import logging
import re
-from rest_framework import status
from newton.pub.exceptions import VimDriverNewtonException
-from newton.pub.utils.restcall import req_by_msb,req_to_aai
+from newton.pub.utils import restcall
logger = logging.getLogger(__name__)
if cloud_owner and cloud_region_id:
retcode, content, status_code = \
- req_to_aai("/cloud-infrastructure/cloud-regions/cloud-region/%s/%s"
+ restcall.req_to_aai("/cloud-infrastructure/cloud-regions/cloud-region/%s/%s"
% (cloud_owner,cloud_region_id),"GET")
if retcode != 0:
logger.error("Status code is %s, detail is %s.", status_code, content)
#assume esr-system-info-id is composed by {cloud-owner} _ {cloud-region-id}
retcode2,content2,status_code2 = \
- req_to_aai("/cloud-infrastructure/cloud-regions/cloud-region/%s/%s"
- + "/esr-system-info-list/esr-system-info/%s_%s" \
- % (cloud_owner,cloud_region_id,cloud_owner,cloud_region_id),
- "GET")
+ restcall.req_to_aai(("/cloud-infrastructure/cloud-regions/cloud-region/%(owner)s/%(region)s"
+ "/esr-system-info-list/esr-system-info/%(owner)s_%(region)s" % {
+ "owner": cloud_owner, "region": cloud_region_id}), "GET")
if retcode2 != 0:
logger.error("Status code is %s, detail is %s.", status_code, content)
raise VimDriverNewtonException(
tmp_authinfo = json.JSONDecoder().decode(content2)
#convert vim information
-
- if tmp_viminfo:
+ if tmp_viminfo and tmp_authinfo:
viminfo = {}
viminfo['vimId'] = vim_id
viminfo['cloud_owner'] = cloud_owner
viminfo['cloud_extra_info'] = tmp_viminfo['cloud-extra-info']
viminfo['cloud_epa_caps'] = tmp_viminfo['cloud-epa-caps']
- if tmp_authinfo:
- viminfo['userName'] = tmp_authinfo['user-name']
- viminfo['password'] = tmp_authinfo['password']
- viminfo['domain'] = tmp_authinfo['cloud-domain']
- viminfo['url'] = tmp_authinfo['service-url']
- viminfo['tenant'] = tmp_authinfo['default-tenant']
- viminfo['cacert'] = tmp_authinfo['ssl-cacert']
- viminfo['insecure'] = tmp_authinfo['ssl-insecure']
- else:
- return None
+ viminfo['userName'] = tmp_authinfo['user-name']
+ viminfo['password'] = tmp_authinfo['password']
+ viminfo['domain'] = tmp_authinfo['cloud-domain']
+ viminfo['url'] = tmp_authinfo['service-url']
+ viminfo['tenant'] = tmp_authinfo['default-tenant']
+ viminfo['cacert'] = tmp_authinfo['ssl-cacert']
+ viminfo['insecure'] = tmp_authinfo['ssl-insecure']
return viminfo
- else:
- return None
- else:
- return None
+ return None
def delete_vim_by_id(vim_id):
cloud_owner, cloud_region_id = decode_vim_id(vim_id)
if cloud_owner and cloud_region_id:
retcode, content, status_code = \
- req_to_aai("/cloud-infrastructure/cloud-regions/cloud-region/%s/%s"
+ restcall.req_to_aai("/cloud-infrastructure/cloud-regions/cloud-region/%s/%s"
% ( cloud_owner, cloud_region_id), "DELETE")
if retcode != 0:
logger.error("Status code is %s, detail is %s.", status_code, content)
--- /dev/null
+# Copyright (c) 2017 Intel Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+
+import mock
+import six
+import unittest
+
+from newton.pub.exceptions import VimDriverNewtonException
+from newton.pub.msapi import extsys
+from newton.pub.utils import restcall
+
+MOCK_VIM_INFO = {
+ "cloud-type": "openstack",
+ "complex-name": "complex",
+ "cloud-region-version": "Regionv1",
+ "cloud-extra-info": "type",
+ "cloud-epa-caps": "epa"
+}
+
+MOCK_ESR_SYSTEM_INFO = {
+ "user-name": "test",
+ "password": "secret",
+ "cloud-domain": "default",
+ "service-url": "http://localhost",
+ "default-tenant": "demo",
+ "ssl-cacert": None,
+ "ssl-insecure": None
+}
+
+
+def returnList(items):
+ def func():
+ for item in items:
+ yield item
+ yield mock.DEFAULT
+
+ generator = func()
+
+ def effect(*args, **kwargs):
+ return six.next(generator)
+
+ return effect
+
+
+class TestEpaCaps(unittest.TestCase):
+ cloud_onwer = "windriver-hudson-cd"
+ cloud_region_id = "RegionOne"
+ vim_id = cloud_onwer + "_" + cloud_region_id
+
+ def setUp(self):
+ self.req_to_aai_backup = restcall.req_to_aai
+
+ def tearDown(self):
+ restcall.req_to_aai = self.req_to_aai_backup
+
+ def test_get_vim_by_id(self):
+ values = [
+ (1, "test_content", 500), # Failure first call
+ (0, json.dumps(MOCK_VIM_INFO), None), (1, "test_content", 500), # Failure second call
+ (0, json.dumps(MOCK_VIM_INFO), None), (0, json.dumps(MOCK_ESR_SYSTEM_INFO), None) # Success calls
+ ]
+
+ restcall.req_to_aai = mock.Mock(side_effect=returnList(values))
+ self.assertRaises(VimDriverNewtonException, extsys.get_vim_by_id, self.vim_id)
+ restcall.req_to_aai.assert_called_once()
+
+ self.assertRaises(VimDriverNewtonException, extsys.get_vim_by_id, self.vim_id)
+
+ viminfo = extsys.get_vim_by_id(self.vim_id)
+ self.assertIsNotNone(viminfo)
+ self.assertEquals(self.vim_id, viminfo['vimId'])
+ self.assertEquals(self.cloud_onwer, viminfo['cloud_owner'])
+ self.assertEquals(self.cloud_region_id, viminfo['cloud_region_id'])
+ self.assertEquals(MOCK_VIM_INFO['cloud-type'], viminfo['type'])
+ self.assertEquals(MOCK_VIM_INFO['complex-name'], viminfo['name'])
+ self.assertEquals(MOCK_VIM_INFO['cloud-region-version'], viminfo['version'])
+ self.assertEquals(MOCK_VIM_INFO['cloud-extra-info'], viminfo['cloud_extra_info'])
+ self.assertEquals(MOCK_VIM_INFO['cloud-epa-caps'], viminfo['cloud_epa_caps'])
+
+ self.assertEquals(MOCK_ESR_SYSTEM_INFO['user-name'], viminfo['userName'])
+ self.assertEquals(MOCK_ESR_SYSTEM_INFO['password'], viminfo['password'])
+ self.assertEquals(MOCK_ESR_SYSTEM_INFO['cloud-domain'], viminfo['domain'])
+ self.assertEquals(MOCK_ESR_SYSTEM_INFO['service-url'], viminfo['url'])
+ self.assertEquals(MOCK_ESR_SYSTEM_INFO['default-tenant'], viminfo['tenant'])
+ self.assertEquals(MOCK_ESR_SYSTEM_INFO['ssl-cacert'], viminfo['cacert'])
+ self.assertEquals(MOCK_ESR_SYSTEM_INFO['ssl-insecure'], viminfo['insecure'])
+
+ def test_delete_vim_by_id(self):
+ values = [(1, "test_content", 500),(0, None, None)]
+
+ restcall.req_to_aai = mock.Mock(side_effect=returnList(values))
+ self.assertRaises(VimDriverNewtonException, extsys.delete_vim_by_id, self.vim_id)
+ self.assertEquals(0, extsys.delete_vim_by_id(self.vim_id))
+
+ def test_decode_vim_id_successfuly(self):
+ owner, region_id = extsys.decode_vim_id(self.vim_id)
+ self.assertEquals(self.cloud_onwer, owner)
+ self.assertEquals(self.cloud_region_id, region_id)
\ No newline at end of file
--- /dev/null
+# Copyright (c) 2017 Intel Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from httplib2 import Http
+import mock
+from rest_framework import status
+import unittest
+
+from newton.pub.utils import restcall
+
+
+class TestRestCall(unittest.TestCase):
+ base_url = "http://localhost"
+ resource = "compute"
+
+ @mock.patch.object(Http, 'request')
+ def test_unknown_failure_call_req(self, mock_http):
+ mock_http.raiseError.side_effect = mock.Mock(
+ side_effect=Exception('Test'))
+ args = [
+ self.base_url, "user", "password", "auth_type",
+ self.resource, "get", {"extra": "test"}
+ ]
+
+ ret = restcall._call_req(*args)
+ self.assertEquals(3, ret[0])
+ self.assertEquals(status.HTTP_500_INTERNAL_SERVER_ERROR, ret[2])
+
+
+ @mock.patch.object(Http, 'request')
+ def test_invalid_output_call_req(self, mock_http):
+ args = [
+ self.base_url, "user", "password", "auth_type",
+ self.resource, "get", {"extra": "test"}
+ ]
+
+ mock_http.return_value = ({'status': None},
+ str.encode("test", 'utf-8'))
+
+ ret = restcall._call_req(*args)
+ self.assertEquals(1, ret[0])
+ self.assertEquals("test", ret[1])
+ self.assertIsNone(ret[2])
+
+ @mock.patch.object(Http, 'request')
+ def test_req_by_msb(self, mock_http):
+ resp_body = "test_body"
+ resp_status=status.HTTP_200_OK
+ mock_http.return_value = (
+ {'status': resp_status},
+ str.encode(resp_body, 'utf-8'))
+
+
+ ret = restcall.req_by_msb(self.resource, "delete")
+ self.assertEquals(0, ret[0])
+ self.assertEquals(resp_body, ret[1])
+ self.assertEquals(resp_status, ret[2])
+
+ @mock.patch.object(Http, 'request')
+ def test_req_to_vim(self, mock_http):
+ resp_body = "test_body"
+ resp_status=status.HTTP_200_OK
+ mock_http.return_value = (
+ {'status': resp_status},
+ str.encode(resp_body, 'utf-8'))
+
+ ret = restcall.req_to_vim(self.base_url, self.resource, "get")
+ self.assertEquals(0, ret[0])
+ self.assertEquals(resp_body, ret[1])
+ self.assertEquals(resp_status, ret[2])
+
+ @mock.patch.object(Http, 'request')
+ def test_req_to_aai(self, mock_http):
+ resp_body = "test_body"
+ resp_status=status.HTTP_200_OK
+ mock_http.return_value = (
+ {'status': resp_status},
+ str.encode(resp_body, 'utf-8'))
+
+ ret = restcall.req_to_aai(self.resource, "post")
+ self.assertEquals(0, ret[0])
+ self.assertEquals(resp_body, ret[1])
+ self.assertEquals(resp_status, ret[2])
+
+ def test_combine_url(self):
+ self.assertEquals(self.base_url,
+ restcall._combine_url(self.base_url, ""))
+ self.assertEquals(self.base_url + "/" + self.resource,
+ restcall._combine_url(self.base_url + "/",
+ "/" + self.resource))
+ self.assertEquals(self.base_url + "/" + self.resource,
+ restcall._combine_url(self.base_url + "/",
+ self.resource))
+ self.assertEquals(self.base_url + "/" + self.resource,
+ restcall._combine_url(self.base_url,
+ "/" + self.resource))
+ self.assertEquals(self.base_url + "/" + self.resource,
+ restcall._combine_url(self.base_url,
+ self.resource))
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-import sys
+import codecs
import json
import traceback
+import sys
+
import logging
from six.moves import urllib
-import uuid
-from six.moves import http_client
import httplib2
import uuid
logger = logging.getLogger(__name__)
-def call_req(base_url, user, passwd, auth_type,
+def _call_req(base_url, user, passwd, auth_type,
resource, method, extra_headers='', content=''):
callid = str(uuid.uuid1())
-# logger.debug("[%s]call_req('%s','%s','%s',%s,'%s','%s','%s')" % (
-# callid, base_url, user, passwd, auth_type, resource, method, content))
ret = None
resp_status = None
try:
- full_url = combine_url(base_url, resource)
+ full_url = _combine_url(base_url, resource)
headers = {
'content-type': 'application/json',
'accept': 'application/json'
headers.update(extra_headers)
if user:
headers['Authorization'] = \
- 'Basic ' + ('%s:%s' % (user, passwd)).encode("base64")
+ 'Basic ' + str(codecs.encode('%s:%s' % (user, passwd), "ascii"))
+
ca_certs = None
for retry_times in range(MAX_RETRY_TIME):
http = httplib2.Http(
body=content,
headers=headers)
resp_status, resp_body = \
- resp['status'], resp_content.decode('UTF-8')
-# logger.debug("[%s][%d]status=%s,resp_body=%s)" %
-# (callid, retry_times, resp_status, resp_body))
+ resp['status'], codecs.decode(
+ resp_content, 'UTF-8')
if resp_status in status_ok_list:
ret = [0, resp_body, resp_status]
else:
raise ex
except urllib.error.URLError as err:
ret = [2, str(err), resp_status]
- except Exception as ex:
+ except Exception:
logger.error(traceback.format_exc())
logger.error("[%s]ret=%s" % (callid, str(sys.exc_info())))
if not resp_status:
except:
logger.error(traceback.format_exc())
ret = [4, str(sys.exc_info()), resp_status]
-
-# logger.debug("[%s]ret=%s" % (callid, str(ret)))
return ret
def req_by_msb(resource, method, content=''):
base_url = "http://%s:%s/" % (config.MSB_SERVICE_ADDR, config.MSB_SERVICE_PORT)
-# logger.debug("requests--get::> %s" % "33")
- return call_req(base_url, "", "", rest_no_auth,
+ return _call_req(base_url, "", "", rest_no_auth,
resource, method, "", content)
def req_to_vim(base_url, resource, method, extra_headers='', content=''):
- return call_req(base_url, "", "", rest_no_auth,
+ return _call_req(base_url, "", "", rest_no_auth,
resource, method, extra_headers, content)
+
def req_to_aai(resource, method, content='', appid=config.MULTICLOUD_APP_ID):
tmp_trasaction_id = str(uuid.uuid1())
headers = {
}
logger.debug("req_to_aai--%s::> %s, %s" % (tmp_trasaction_id, method, resource))
- return call_req(config.AAI_BASE_URL, config.AAI_USERNAME, config.AAI_PASSWORD, rest_no_auth,
+ return _call_req(config.AAI_BASE_URL, config.AAI_USERNAME, config.AAI_PASSWORD, rest_no_auth,
resource, method, content=json.dumps(content), extra_headers=headers)
-def combine_url(base_url, resource):
+def _combine_url(base_url, resource):
full_url = None
if not resource:
Code Review / multicloud / openstack.git / commitdiff