x/vmware-nsxlib
Code Issues Proposed changes
vmware-nsxlib/vmware_nsxlib/tests/unit/v3/test_client.py
Shawn Wang 60eaa86892
Expose HTTP Status Code in ManagerError 
When ManagerError is raised due to unavailability of NSX backend, it is
common that return HTTP status is non-200 and no error code is provided
by NSX. In this case, recipient of ManagerError can hardly check the
cause due to not having access to either error code or HTTP status code.

This patch includes the HTTP status code in ManagerError, aiming to
provide recipient a more reliable way to check the cause of error.

Change-Id: Id346b27ad69ca4c3b3b4ef950a8c191ce46cde5e
2020-01-30 13:20:51 -08:00

407 lines
14 KiB
Python
Raw Blame History

# Copyright 2015 VMware, Inc.
# All Rights Reserved
#
# 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 copy
from oslo_log import log
from oslo_serialization import jsonutils
import requests
from vmware_nsxlib.tests.unit.v3 import mocks
from vmware_nsxlib.tests.unit.v3 import nsxlib_testcase
from vmware_nsxlib.v3 import client
from vmware_nsxlib.v3 import exceptions as nsxlib_exc
from vmware_nsxlib.v3 import utils
LOG = log.getLogger(__name__)
DFT_ACCEPT_HEADERS = {
'Accept': '*/*',
'Cookie': 'JSESSIONID=%s;' % nsxlib_testcase.JSESSIONID
}
JSON_DFT_ACCEPT_HEADERS = {
'Accept': 'application/json',
'Content-Type': 'application/json',
'Cookie': 'JSESSIONID=%s;' % nsxlib_testcase.JSESSIONID
}
PARTIAL_UPDATE_HEADERS = {
'Accept': 'application/json',
'Content-Type': 'application/json',
'Cookie': 'JSESSIONID=%s;' % nsxlib_testcase.JSESSIONID,
'nsx-enable-partial-patch': 'true'
}
def _headers(**kwargs):
headers = copy.copy(DFT_ACCEPT_HEADERS)
headers.update(kwargs)
return headers
def assert_call(verb, client_or_resource,
url, verify=nsxlib_testcase.NSX_CERT,
data=None, headers=DFT_ACCEPT_HEADERS,
timeout=(nsxlib_testcase.NSX_HTTP_TIMEOUT,
nsxlib_testcase.NSX_HTTP_READ_TIMEOUT),
single_call=True):
nsx_client = client_or_resource
if getattr(nsx_client, 'client', None) is not None:
nsx_client = nsx_client.client
cluster = nsx_client._conn
if single_call:
cluster.assert_called_once(
verb,
**{'url': url, 'verify': verify, 'body': data,
'headers': headers, 'cert': None, 'timeout': timeout})
else:
cluster.assert_any_call(
verb,
**{'url': url, 'verify': verify, 'body': data,
'headers': headers, 'cert': None, 'timeout': timeout})
def mock_calls_count(verb, client_or_resource):
nsx_client = client_or_resource
if getattr(nsx_client, 'client', None) is not None:
nsx_client = nsx_client.client
cluster = nsx_client._conn
return cluster.call_count(verb)
def assert_json_call(verb, client_or_resource, url,
verify=nsxlib_testcase.NSX_CERT,
data=None,
headers=JSON_DFT_ACCEPT_HEADERS,
single_call=True):
return assert_call(verb, client_or_resource, url,
verify=verify, data=data,
headers=headers, single_call=single_call)
class NsxV3RESTClientTestCase(nsxlib_testcase.NsxClientTestCase):
def test_client_url_prefix(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='/cloud/api')
api.list()
assert_call(
'get', api,
'https://1.2.3.4/cloud/api')
api = self.new_mocked_client(client.RESTClient,
url_prefix='/cloud/api')
api.url_list('v1/ports')
assert_call(
'get', api,
'https://1.2.3.4/cloud/api/v1/ports')
def test_client_headers(self):
default_headers = {'Content-Type': 'application/golang'}
api = self.new_mocked_client(
client.RESTClient, default_headers=default_headers,
url_prefix='/v1/api')
api.list()
assert_call(
'get', api,
'https://1.2.3.4/v1/api',
headers=_headers(**default_headers))
api = self.new_mocked_client(
client.RESTClient,
default_headers=default_headers,
url_prefix='/v1/api')
method_headers = {'X-API-Key': 'strong-crypt'}
api.url_list('ports/33', headers=method_headers)
method_headers.update(default_headers)
assert_call(
'get', api,
'https://1.2.3.4/v1/api/ports/33',
headers=_headers(**method_headers))
def test_client_for(self):
api = self.new_mocked_client(client.RESTClient, url_prefix='api/v1/')
sub_api = api.new_client_for('switch/ports')
sub_api.get('11a2b')
assert_call(
'get', sub_api,
'https://1.2.3.4/api/v1/switch/ports/11a2b')
def test_client_list(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.list()
assert_call(
'get', api,
'https://1.2.3.4/api/v1/ports')
def test_client_get(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.get('unique-id')
assert_call(
'get', api,
'https://1.2.3.4/api/v1/ports/unique-id')
def test_client_delete(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.delete('unique-id')
assert_call(
'delete', api,
'https://1.2.3.4/api/v1/ports/unique-id')
def test_client_update(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.update('unique-id', jsonutils.dumps({'name': 'a-new-name'}))
assert_call(
'put', api,
'https://1.2.3.4/api/v1/ports/unique-id',
data=jsonutils.dumps({'name': 'a-new-name'}))
def test_client_create(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.create(body=jsonutils.dumps({'resource-name': 'port1'}))
assert_call(
'post', api,
'https://1.2.3.4/api/v1/ports',
data=jsonutils.dumps({'resource-name': 'port1'}))
def test_client_url_list(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
json_headers = {'Content-Type': 'application/json'}
api.url_list('/connections', json_headers)
assert_call(
'get', api,
'https://1.2.3.4/api/v1/ports/connections',
headers=_headers(**json_headers))
def test_client_url_get(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.url_get('connections/1')
assert_call(
'get', api,
'https://1.2.3.4/api/v1/ports/connections/1')
def test_client_url_delete(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.url_delete('1')
assert_call(
'delete', api,
'https://1.2.3.4/api/v1/ports/1')
def test_client_url_put(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.url_put('connections/1', jsonutils.dumps({'name': 'conn1'}))
assert_call(
'put', api,
'https://1.2.3.4/api/v1/ports/connections/1',
data=jsonutils.dumps({'name': 'conn1'}))
def test_client_url_post(self):
api = self.new_mocked_client(client.RESTClient,
url_prefix='api/v1/ports')
api.url_post('1/connections', jsonutils.dumps({'name': 'conn1'}))
assert_call(
'post', api,
'https://1.2.3.4/api/v1/ports/1/connections',
data=jsonutils.dumps({'name': 'conn1'}))
def test_client_validate_result(self):
def _verb_response_code(http_verb, status_code, error_code=None):
content = None
if error_code:
content = jsonutils.dumps({'httpStatus': 'dummy',
'error_code': error_code,
'module_name': 'dummy',
'error_message': 'bad',
'related_errors': [{
'error_message': 'bla',
'error_code': 'code'}]})
response = mocks.MockRequestsResponse(
status_code, content)
client_api = self.new_mocked_client(
client.RESTClient, mock_validate=False,
session_response=response)
client_call = getattr(client_api, "url_%s" % http_verb)
client_call('', None)
for verb in ['get', 'post', 'put', 'delete']:
for code in client.RESTClient._VERB_RESP_CODES.get(verb):
_verb_response_code(verb, code)
with self.assertRaises(nsxlib_exc.ManagerError) as e:
_verb_response_code(verb, requests.codes.INTERNAL_SERVER_ERROR)
self.assertEqual(e.exception.status_code,
requests.codes.INTERNAL_SERVER_ERROR)
with self.assertRaises(nsxlib_exc.ResourceNotFound) as e:
_verb_response_code(verb, requests.codes.NOT_FOUND)
self.assertEqual(e.exception.status_code, requests.codes.NOT_FOUND)
with self.assertRaises(nsxlib_exc.BackendResourceNotFound) as e:
_verb_response_code(verb, requests.codes.NOT_FOUND, 202)
self.assertEqual(e.exception.status_code, requests.codes.NOT_FOUND)
def test_inject_headers_callback(self):
self.injected = None
def inject_header():
self.injected = True
return {}
utils.set_inject_headers_callback(inject_header)
api = self.new_mocked_client(
client.RESTClient,
url_prefix='/v1/api')
api.list()
injected_headers = {}
assert_call(
'get', api,
'https://1.2.3.4/v1/api',
headers=_headers(**injected_headers))
api = self.new_mocked_client(
client.RESTClient,
url_prefix='/v1/api')
utils.set_inject_headers_callback(None)
self.assertIsNotNone(self.injected)
def test_http_error_to_exception(self):
exc = client.http_error_to_exception(500, 607)
self.assertEqual(exc, nsxlib_exc.APITransactionAborted)
class NsxV3JSONClientTestCase(nsxlib_testcase.NsxClientTestCase):
def test_json_request(self):
resp = mocks.MockRequestsResponse(
200, jsonutils.dumps({'result': {'ok': 200}}))
api = self.new_mocked_client(client.JSONRESTClient,
session_response=resp,
url_prefix='api/v2/nat')
resp = api.create(body={'name': 'mgmt-egress'})
assert_json_call(
'post', api,
'https://1.2.3.4/api/v2/nat',
data=jsonutils.dumps({'name': 'mgmt-egress'}))
self.assertEqual(resp, {'result': {'ok': 200}})
def test_mask_password(self):
pwds = ('my!pwd0#', 'some0therlong$pwd')
body = {'name_pwd': 'name1',
'password': pwds[0],
'some_list': {'name_password': 'name2',
'password': pwds[1]}}
cl = client.RESTClient(None)
json_body = jsonutils.dumps(body)
masked_body = cl._mask_password(json_body)
for pwd in pwds:
json_body = json_body.replace('"' + pwd + '"', '"********"')
self.assertEqual(json_body, masked_body)
class NsxV3APIClientTestCase(nsxlib_testcase.NsxClientTestCase):
def test_api_call(self):
api = self.new_mocked_client(client.NSX3Client)
api.get('ports')
assert_json_call(
'get', api,
'https://1.2.3.4/api/v1/ports')
def test_raise_error(self):
api = self.new_mocked_client(client.NSX3Client)
with self.assertRaises(nsxlib_exc.ManagerError) as e:
api._raise_error(requests.codes.INTERNAL_SERVER_ERROR, 'GET', '')
self.assertEqual(e.exception.status_code,
requests.codes.INTERNAL_SERVER_ERROR)
# NOTE(boden): remove this when tmp brigding removed
class NsxV3APIClientBridgeTestCase(nsxlib_testcase.NsxClientTestCase):
def test_get_resource(self):
api = self.new_mocked_client(client.NSX3Client)
api.get('ports')
assert_json_call(
'get', api,
'https://1.2.3.4/api/v1/ports')
def test_create_resource(self):
api = self.new_mocked_client(client.NSX3Client)
api.create('ports', {'resource-name': 'port1'})
assert_json_call(
'post', api,
'https://1.2.3.4/api/v1/ports',
data=jsonutils.dumps({'resource-name': 'port1'}))
def test_update_resource(self):
api = self.new_mocked_client(client.NSX3Client)
api.update('ports/1', {'name': 'a-new-name'})
assert_json_call(
'put', api,
'https://1.2.3.4/api/v1/ports/1',
data=jsonutils.dumps({'name': 'a-new-name'}))
def test_delete_resource(self):
api = self.new_mocked_client(client.NSX3Client)
api.delete('ports/11')
assert_json_call(
'delete', api,
'https://1.2.3.4/api/v1/ports/11')
View Git Blame Copy Permalink