x
/
vmware-nsxlib
Code Issues Proposed changes
vmware-nsxlib/vmware_nsxlib/tests/unit/v3/test_client.py

419 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',
'httpStatus': 'BAD_REQUEST'}]})
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)
exc = client.http_error_to_exception(requests.codes.FORBIDDEN, 505)
self.assertEqual(exc, nsxlib_exc.InvalidLicense)
exc = client.http_error_to_exception(requests.codes.FORBIDDEN, 505,
[777])
self.assertEqual(exc, nsxlib_exc.InvalidLicense)
exc = client.http_error_to_exception(requests.codes.BAD_REQUEST,
500157, [777, 500045])
self.assertEqual(exc, nsxlib_exc.NsxPendingDelete)
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('GET', requests.codes.INTERNAL_SERVER_ERROR, '')
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