openstack/python-tackerclient
Code Issues Proposed changes

Refactor mox references to use the mock library in pythontackerclient

Browse Source 
use mock instead of mox

Co-Authored-By: Trinath Somanchi <trinanth.somachi@nxp.com>
Change-Id: I9e5327a4e4e6cba220670b096f718e7343fd43dc
Closes-Bug: #1572706
This commit is contained in:
aksingh 2017-02-06 23:22:45 +05:30 committed by Trinath Somanchi
parent bd3294415e
commit 4c3adee97b
6 changed files with 207 additions and 322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.testr.conf
View File

@ -1,4 +1,4 @@
[DEFAULT] [DEFAULT]
test_command=OS_STDOUT_CAPTURE=1 OS_STDERR_CAPTURE=1 ${PYTHON:-python} -m subunit.run discover -t ./ ${OS_TEST_PATH:-./tackerclient/tests/unit/vm} $LISTOPT $IDOPTION test_command=OS_STDOUT_CAPTURE=1 OS_STDERR_CAPTURE=1 ${PYTHON:-python} -m subunit.run discover -t ./ ${OS_TEST_PATH:-./tackerclient/tests/unit} $LISTOPT $IDOPTION
test_id_option=--load-list $IDFILE test_id_option=--load-list $IDFILE
test_list_option=--list test_list_option=--list

238
tackerclient/tests/unit/test_auth.py
View File

@ -19,13 +19,12 @@ import json
import uuid import uuid
from keystoneclient import exceptions as k_exceptions from keystoneclient import exceptions as k_exceptions
import mox import mock
import requests import requests
import testtools import testtools
from tackerclient import client from tackerclient import client
from tackerclient.common import exceptions from tackerclient.common import exceptions
from tackerclient.common import utils
USERNAME = 'testuser' USERNAME = 'testuser'
@ -71,39 +70,44 @@ ENDPOINTS_RESULT = {
def get_response(status_code, headers=None): def get_response(status_code, headers=None):
response = mox.Mox().CreateMock(requests.Response) response = mock.Mock().CreateMock(requests.Response)
response.headers = headers or {} response.headers = headers or {}
response.status_code = status_code response.status_code = status_code
return response return response
resp_200 = get_response(200)
resp_401 = get_response(401)
headers = {'X-Auth-Token': '',
'User-Agent': 'python-tackerclient'}
expected_headers = {'X-Auth-Token': TOKEN,
'User-Agent': 'python-tackerclient'}
agent_header = {'User-Agent': 'python-tackerclient'}
class CLITestAuthNoAuth(testtools.TestCase): class CLITestAuthNoAuth(testtools.TestCase):
def setUp(self): def setUp(self):
"""Prepare the test environment.""" """Prepare the test environment."""
super(CLITestAuthNoAuth, self).setUp() super(CLITestAuthNoAuth, self).setUp()
self.mox = mox.Mox()
self.client = client.HTTPClient(username=USERNAME, self.client = client.HTTPClient(username=USERNAME,
tenant_name=TENANT_NAME, tenant_name=TENANT_NAME,
password=PASSWORD, password=PASSWORD,
endpoint_url=ENDPOINT_URL, endpoint_url=ENDPOINT_URL,
auth_strategy=NOAUTH, auth_strategy=NOAUTH,
region_name=REGION) region_name=REGION)
self.addCleanup(self.mox.VerifyAll) self.addCleanup(mock.patch.stopall)
self.addCleanup(self.mox.UnsetStubs)
def test_get_noauth(self): @mock.patch('tackerclient.client.HTTPClient.request')
self.mox.StubOutWithMock(self.client, "request") def test_get_noauth(self, mock_request):
res200 = get_response(200) mock_request.return_value = (resp_200, '')
self.client.do_request('/resource', 'GET',
self.client.request( headers=headers)
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', mock_request.assert_called_once_with(
headers=mox.IsA(dict), ENDPOINT_URL + '/resource',
).AndReturn((res200, '')) 'GET',
self.mox.ReplayAll() headers=headers)
self.client.do_request('/resource', 'GET')
self.assertEqual(self.client.endpoint_url, ENDPOINT_URL) self.assertEqual(self.client.endpoint_url, ENDPOINT_URL)
@ -117,14 +121,12 @@ class CLITestAuthKeystone(testtools.TestCase):
def setUp(self): def setUp(self):
"""Prepare the test environment.""" """Prepare the test environment."""
super(CLITestAuthKeystone, self).setUp() super(CLITestAuthKeystone, self).setUp()
self.mox = mox.Mox()
self.client = client.HTTPClient(username=USERNAME, self.client = client.HTTPClient(username=USERNAME,
tenant_name=TENANT_NAME, tenant_name=TENANT_NAME,
password=PASSWORD, password=PASSWORD,
auth_url=AUTH_URL, auth_url=AUTH_URL,
region_name=REGION) region_name=REGION)
self.addCleanup(self.mox.VerifyAll) self.addCleanup(mock.patch.stopall)
self.addCleanup(self.mox.UnsetStubs)
def test_reused_token_get_auth_info(self): def test_reused_token_get_auth_info(self):
"""Test Client.get_auth_info(). """Test Client.get_auth_info().
@ -144,69 +146,53 @@ class CLITestAuthKeystone(testtools.TestCase):
'endpoint_url': self.client.endpoint_url} 'endpoint_url': self.client.endpoint_url}
self.assertEqual(client_.get_auth_info(), expected) self.assertEqual(client_.get_auth_info(), expected)
def test_get_token(self): @mock.patch('tackerclient.client.HTTPClient.request')
self.mox.StubOutWithMock(self.client, "request") def test_get_token(self, mock_request):
res200 = get_response(200)
self.client.request(
AUTH_URL + '/tokens', 'POST',
body=self.auth_body, headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res200, ''))
self.mox.ReplayAll()
mock_request.return_value = (resp_200, json.dumps(KS_TOKEN_RESULT))
self.client.do_request('/resource', 'GET') self.client.do_request('/resource', 'GET')
mock_request.assert_called_with(
ENDPOINT_URL + '/resource', 'GET',
headers=expected_headers)
self.assertEqual(self.client.endpoint_url, ENDPOINT_URL) self.assertEqual(self.client.endpoint_url, ENDPOINT_URL)
self.assertEqual(self.client.auth_token, TOKEN) self.assertEqual(self.client.auth_token, TOKEN)
def test_refresh_token(self): @mock.patch('tackerclient.client.HTTPClient.request')
self.mox.StubOutWithMock(self.client, "request") def test_refresh_token(self, mock_request):
self.client.auth_token = TOKEN self.client.auth_token = TOKEN
self.client.endpoint_url = ENDPOINT_URL self.client.endpoint_url = ENDPOINT_URL
res200 = get_response(200)
res401 = get_response(401)
# If a token is expired, tacker server retruns 401 # If a token is expired, tacker server retruns 401
self.client.request( mock_request.return_value = (resp_401, '')
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', self.assertRaises(exceptions.Unauthorized,
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN) self.client.do_request,
).AndReturn((res401, '')) '/resource',
self.client.request( 'GET')
AUTH_URL + '/tokens', 'POST',
body=mox.IsA(str), headers=mox.IsA(dict) mock_request.return_value = (resp_200, json.dumps(KS_TOKEN_RESULT))
).AndReturn((res200, json.dumps(KS_TOKEN_RESULT))) self.client.do_request('/resource', 'GET')
self.client.request( mock_request.assert_called_with(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', ENDPOINT_URL + '/resource', 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN) headers=expected_headers)
).AndReturn((res200, ''))
self.mox.ReplayAll() @mock.patch('tackerclient.client.HTTPClient.request')
self.client.do_request('/resource', 'GET') def test_refresh_token_no_auth_url(self, mock_request):
def test_refresh_token_no_auth_url(self):
self.mox.StubOutWithMock(self.client, "request")
self.client.auth_url = None self.client.auth_url = None
self.client.auth_token = TOKEN self.client.auth_token = TOKEN
self.client.endpoint_url = ENDPOINT_URL self.client.endpoint_url = ENDPOINT_URL
res401 = get_response(401) # If a token is expired, tacker server retruns 401
mock_request.return_value = (resp_401, '')
# If a token is expired, tacker server returns 401
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res401, ''))
self.mox.ReplayAll()
self.assertRaises(exceptions.NoAuthURLProvided, self.assertRaises(exceptions.NoAuthURLProvided,
self.client.do_request, self.client.do_request,
'/resource', '/resource',
'GET') 'GET')
expected_url = ENDPOINT_URL + '/resource'
mock_request.assert_called_with(expected_url, 'GET',
headers=expected_headers)
def test_get_endpoint_url_with_invalid_auth_url(self): def test_get_endpoint_url_with_invalid_auth_url(self):
# Handle the case when auth_url is not provided # Handle the case when auth_url is not provided
@ -214,85 +200,70 @@ class CLITestAuthKeystone(testtools.TestCase):
self.assertRaises(exceptions.NoAuthURLProvided, self.assertRaises(exceptions.NoAuthURLProvided,
self.client._get_endpoint_url) self.client._get_endpoint_url)
def test_get_endpoint_url(self): @mock.patch('tackerclient.client.HTTPClient.request')
self.mox.StubOutWithMock(self.client, "request") def test_get_endpoint_url(self, mock_request):
self.client.auth_token = TOKEN self.client.auth_token = TOKEN
res200 = get_response(200) mock_request.return_value = (resp_200, json.dumps(ENDPOINTS_RESULT))
self.client.request(
mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET',
headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET') self.client.do_request('/resource', 'GET')
mock_request.assert_called_with(
ENDPOINT_URL + '/resource', 'GET',
headers=expected_headers)
def test_use_given_endpoint_url(self): mock_request.return_value = (resp_200, '')
self.client.do_request('/resource', 'GET',
headers=headers)
mock_request.assert_called_with(
ENDPOINT_URL + '/resource', 'GET',
headers=headers)
@mock.patch('tackerclient.client.HTTPClient.request')
def test_use_given_endpoint_url(self, mock_request):
self.client = client.HTTPClient( self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD, username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
auth_url=AUTH_URL, region_name=REGION, auth_url=AUTH_URL, region_name=REGION,
endpoint_url=ENDPOINT_OVERRIDE) endpoint_url=ENDPOINT_OVERRIDE)
self.assertEqual(self.client.endpoint_url, ENDPOINT_OVERRIDE) self.assertEqual(self.client.endpoint_url, ENDPOINT_OVERRIDE)
self.mox.StubOutWithMock(self.client, "request")
self.client.auth_token = TOKEN self.client.auth_token = TOKEN
res200 = get_response(200) mock_request.return_value = (resp_200, '')
self.client.request( self.client.do_request('/resource', 'GET',
mox.StrContains(ENDPOINT_OVERRIDE + '/resource'), 'GET', headers=headers)
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN) mock_request.assert_called_with(
).AndReturn((res200, '')) ENDPOINT_OVERRIDE + '/resource', 'GET',
self.mox.ReplayAll() headers=headers)
self.client.do_request('/resource', 'GET')
self.assertEqual(self.client.endpoint_url, ENDPOINT_OVERRIDE) self.assertEqual(self.client.endpoint_url, ENDPOINT_OVERRIDE)
def test_get_endpoint_url_other(self): @mock.patch('tackerclient.client.HTTPClient.request')
def test_get_endpoint_url_other(self, mock_request):
self.client = client.HTTPClient( self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD, username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
auth_url=AUTH_URL, region_name=REGION, endpoint_type='otherURL') auth_url=AUTH_URL, region_name=REGION, endpoint_type='otherURL')
self.mox.StubOutWithMock(self.client, "request")
self.client.auth_token = TOKEN self.client.auth_token = TOKEN
res200 = get_response(200) mock_request.return_value = (resp_200, json.dumps(ENDPOINTS_RESULT))
self.client.request(
mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET',
headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
self.mox.ReplayAll()
self.assertRaises(exceptions.EndpointTypeNotFound, self.assertRaises(exceptions.EndpointTypeNotFound,
self.client.do_request, self.client.do_request,
'/resource', '/resource',
'GET') 'GET')
expected_url = AUTH_URL + '/tokens/%s/endpoints' % TOKEN
headers = {'User-Agent': 'python-tackerclient'}
mock_request.assert_called_with(expected_url, 'GET',
headers=headers)
def test_get_endpoint_url_failed(self): @mock.patch('tackerclient.client.HTTPClient.request')
self.mox.StubOutWithMock(self.client, "request") def test_get_endpoint_url_failed(self, mock_request):
self.client.auth_token = TOKEN self.client.auth_token = TOKEN
self.client.auth_url = AUTH_URL + '/tokens/%s/endpoints' % TOKEN
res200 = get_response(200) mock_request.return_value = (resp_401, '')
res401 = get_response(401) self.assertRaises(exceptions.Unauthorized,
self.client.do_request,
self.client.request( '/resource',
mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET', 'GET')
headers=mox.IsA(dict)
).AndReturn((res401, ''))
self.client.request(
AUTH_URL + '/tokens', 'POST',
body=mox.IsA(str), headers=mox.IsA(dict)
).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
self.client.request(
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET')
def test_endpoint_type(self): def test_endpoint_type(self):
resources = copy.deepcopy(KS_TOKEN_RESULT) resources = copy.deepcopy(KS_TOKEN_RESULT)
@ -342,32 +313,25 @@ class CLITestAuthKeystone(testtools.TestCase):
self.client._extract_service_catalog, self.client._extract_service_catalog,
resources) resources)
def test_strip_credentials_from_log(self): @mock.patch('tackerclient.client.HTTPClient.request')
def verify_no_credentials(kwargs): @mock.patch('tackerclient.common.utils.http_log_req')
return ('REDACTED' in kwargs['body']) and ( def test_strip_credentials_from_log(self, mock_http_log_req,
self.client.password not in kwargs['body']) mock_request,):
def verify_credentials(body): body = ('{"auth": {"tenantId": "testtenant_id",'
return 'REDACTED' not in body and self.client.password in body '"passwordCredentials": {"password": "password",'
'"userId": "testuser_id"}}}')
expected_body = ('{"auth": {"tenantId": "testtenant_id",'
'"REDACTEDCredentials": {"REDACTED": "REDACTED",'
'"userId": "testuser_id"}}}')
_headers = {'headers': expected_headers, 'body': expected_body}
self.mox.StubOutWithMock(self.client, "request") mock_request.return_value = (resp_200, json.dumps(KS_TOKEN_RESULT))
self.mox.StubOutWithMock(utils, "http_log_req") self.client.do_request('/resource', 'GET', body=body)
res200 = get_response(200) args, kwargs = mock_http_log_req.call_args
# Check that credentials are stripped while logging.
utils.http_log_req(mox.IgnoreArg(), mox.IgnoreArg(), mox.Func( self.assertEqual(_headers, args[2])
verify_no_credentials))
self.client.request(
mox.IsA(str), mox.IsA(str), body=mox.Func(verify_credentials),
headers=mox.IgnoreArg()
).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
utils.http_log_req(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg())
self.client.request(
mox.IsA(str), mox.IsA(str), headers=mox.IsA(dict)
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET')
class CLITestAuthKeystoneWithId(CLITestAuthKeystone): class CLITestAuthKeystoneWithId(CLITestAuthKeystone):

25
tackerclient/tests/unit/test_cli10.py
View File

@ -645,36 +645,27 @@ class ClientV1TestJson(CLITestV10Base):
def test_do_request_unicode(self): def test_do_request_unicode(self):
self.client.format = self.format self.client.format = self.format
unicode_text = u'\u7f51\u7edc' unicode_text = u'\u7f51\u7edc'
# url with unicode
action = u'/test' action = u'/test'
expected_action = action.encode('utf-8')
# query string with unicode
params = {'test': unicode_text} params = {'test': unicode_text}
expect_query = urllib.urlencode({'test':
unicode_text.encode('utf-8')})
# request body with unicode
body = params body = params
expect_body = self.client.serialize(body) expect_body = self.client.serialize(body)
# headers with unicode
self.client.httpclient.auth_token = unicode_text self.client.httpclient.auth_token = unicode_text
expected_auth_token = unicode_text.encode('utf-8')
with mock.patch.object(self.client.httpclient, 'request') as mock_req: with mock.patch.object(self.client.httpclient, 'request') as mock_req:
mock_req.return_value = (MyResp(200), expect_body) mock_req.return_value = (MyResp(200), expect_body)
res_body = self.client.do_request('PUT', action, body=body, res_body = self.client.do_request('PUT', action, body=body,
params=params) params=params)
mock_req.assert_called_once_with( expected_uri = u'localurl/v1.0/test.json?test=%E7%BD%91%E7%BB%9C'
end_url(expected_action, query=expect_query, mock_req.assert_called_with(
format=self.format), expected_uri, 'PUT', body=expect_body,
'PUT', body=expect_body, headers={'X-Auth-Token': unicode_text,
headers=test_utils.ContainsKeyValue('X-Auth-Token', 'User-Agent': 'python-tackerclient'})
expected_auth_token))
# test response with unicode # test response with unicode
self.assertEqual(res_body, body) self.assertEqual(res_body, body)
def test_do_request_error_without_response_body(self): def test_do_request_error_without_response_body(self):
self.client.format = self.format self.client.format = self.format
params = {'test': 'value'} params = {'test': 'value'}
expect_query = urllib.urlencode(params) expect_query = urlparse.urlencode(params)
self.client.httpclient.auth_token = 'token' self.client.httpclient.auth_token = 'token'
with mock.patch.object(self.client.httpclient, 'request') as mock_req: with mock.patch.object(self.client.httpclient, 'request') as mock_req:
mock_req.return_value = (MyResp(400, reason='An error'), '') mock_req.return_value = (MyResp(400, reason='An error'), '')
@ -704,13 +695,15 @@ class CLITestV10ExceptionHandler(CLITestV10Base):
client.exception_handler_v10, client.exception_handler_v10,
status_code, error_content) status_code, error_content)
self.assertEqual(status_code, e.status_code) self.assertEqual(status_code, e.status_code)
self.assertEqual(expected_exception.__name__,
e.__class__.__name__)
if expected_msg is None: if expected_msg is None:
if error_detail: if error_detail:
expected_msg = '\n'.join([error_msg, error_detail]) expected_msg = '\n'.join([error_msg, error_detail])
else: else:
expected_msg = error_msg expected_msg = error_msg
self.assertEqual(expected_msg, str(e)) self.assertEqual(expected_msg, e.message)
def test_exception_handler_v10_ip_address_in_use(self): def test_exception_handler_v10_ip_address_in_use(self):
err_msg = ('Unable to complete operation for network ' err_msg = ('Unable to complete operation for network '

55
tackerclient/tests/unit/test_http.py
View File

@ -13,7 +13,7 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import mox import mock
import testtools import testtools
from tackerclient.client import HTTPClient from tackerclient.client import HTTPClient
@ -25,60 +25,47 @@ AUTH_TOKEN = 'test_token'
END_URL = 'test_url' END_URL = 'test_url'
METHOD = 'GET' METHOD = 'GET'
URL = 'http://test.test:1234/v1.0/test' URL = 'http://test.test:1234/v1.0/test'
headers = {'User-Agent': 'python-tackerclient'}
class TestHTTPClient(testtools.TestCase): class TestHTTPClient(testtools.TestCase):
def setUp(self): def setUp(self):
super(TestHTTPClient, self).setUp() super(TestHTTPClient, self).setUp()
self.addCleanup(mock.patch.stopall)
self.mox = mox.Mox()
self.mox.StubOutWithMock(HTTPClient, 'request')
self.addCleanup(self.mox.UnsetStubs)
self.http = HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL) self.http = HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL)
def test_request_error(self): @mock.patch('tackerclient.client.HTTPClient.request')
HTTPClient.request( def test_request_error(self, mock_request):
URL, METHOD, headers=mox.IgnoreArg()
).AndRaise(Exception('error msg'))
self.mox.ReplayAll()
mock_request.side_effect = Exception('error msg')
self.assertRaises( self.assertRaises(
exceptions.ConnectionFailed, exceptions.ConnectionFailed,
self.http._cs_request, self.http._cs_request,
URL, METHOD URL, METHOD
) )
self.mox.VerifyAll()
def test_request_success(self): @mock.patch('tackerclient.client.HTTPClient.request')
def test_request_success(self, mock_request):
rv_should_be = MyResp(200), 'test content' rv_should_be = MyResp(200), 'test content'
mock_request.return_value = rv_should_be
HTTPClient.request(
URL, METHOD, headers=mox.IgnoreArg()
).AndReturn(rv_should_be)
self.mox.ReplayAll()
self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD)) self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD))
self.mox.VerifyAll()
def test_request_unauthorized(self): @mock.patch('tackerclient.client.HTTPClient.request')
rv_should_be = MyResp(401), 'unauthorized message' def test_request_unauthorized(self, mock_request):
HTTPClient.request(
URL, METHOD, headers=mox.IgnoreArg() mock_request.return_value = MyResp(401), 'unauthorized message'
).AndReturn(rv_should_be)
self.mox.ReplayAll()
e = self.assertRaises(exceptions.Unauthorized, e = self.assertRaises(exceptions.Unauthorized,
self.http._cs_request, URL, METHOD) self.http._cs_request, URL, METHOD)
self.assertEqual('unauthorized message', str(e)) self.assertEqual('unauthorized message', str(e))
self.mox.VerifyAll() mock_request.assert_called_with(URL, METHOD, headers=headers)
@mock.patch('tackerclient.client.HTTPClient.request')
def test_request_forbidden_is_returned_to_caller(self, mock_request):
def test_request_forbidden_is_returned_to_caller(self):
rv_should_be = MyResp(403), 'forbidden message' rv_should_be = MyResp(403), 'forbidden message'
HTTPClient.request( mock_request.return_value = rv_should_be
URL, METHOD, headers=mox.IgnoreArg()
).AndReturn(rv_should_be)
self.mox.ReplayAll()
self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD)) self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD))
self.mox.VerifyAll()

73
tackerclient/tests/unit/test_shell.py
View File

@ -21,7 +21,8 @@ import six
import sys import sys
import fixtures import fixtures
import mox from keystoneclient import session
import mock
import testtools import testtools
from testtools import matchers from testtools import matchers
@ -36,6 +37,7 @@ DEFAULT_TENANT_NAME = 'tenant_name'
DEFAULT_AUTH_URL = 'http://127.0.0.1:5000/v1.0/' DEFAULT_AUTH_URL = 'http://127.0.0.1:5000/v1.0/'
DEFAULT_TOKEN = '3bcc3d3a03f44e3d8377f9247b0ad155' DEFAULT_TOKEN = '3bcc3d3a03f44e3d8377f9247b0ad155'
DEFAULT_URL = 'http://tacker.example.org:9890/' DEFAULT_URL = 'http://tacker.example.org:9890/'
DEFAULT_API_VERSION = '1.0'
class ShellTest(testtools.TestCase): class ShellTest(testtools.TestCase):
@ -50,7 +52,6 @@ class ShellTest(testtools.TestCase):
# Patch os.environ to avoid required auth info. # Patch os.environ to avoid required auth info.
def setUp(self): def setUp(self):
super(ShellTest, self).setUp() super(ShellTest, self).setUp()
self.mox = mox.Mox()
for var in self.FAKE_ENV: for var in self.FAKE_ENV:
self.useFixture( self.useFixture(
fixtures.EnvironmentVariable( fixtures.EnvironmentVariable(
@ -63,7 +64,7 @@ class ShellTest(testtools.TestCase):
try: try:
sys.stdout = six.StringIO() sys.stdout = six.StringIO()
sys.stderr = six.StringIO() sys.stderr = six.StringIO()
_shell = openstack_shell.TackerShell('1.0') _shell = openstack_shell.TackerShell(DEFAULT_API_VERSION)
_shell.run(argstr.split()) _shell.run(argstr.split())
except SystemExit: except SystemExit:
exc_type, exc_value, exc_traceback = sys.exc_info() exc_type, exc_value, exc_traceback = sys.exc_info()
@ -103,7 +104,7 @@ class ShellTest(testtools.TestCase):
def test_help_command(self): def test_help_command(self):
required = 'usage:' required = 'usage:'
help_text, stderr = self.shell('help network-create') help_text, stderr = self.shell('help vnfd-create')
self.assertThat( self.assertThat(
help_text, help_text,
matchers.MatchesRegex(required)) matchers.MatchesRegex(required))
@ -114,15 +115,27 @@ class ShellTest(testtools.TestCase):
stdout, stderr = self.shell('--os-auth-strategy fake ' stdout, stderr = self.shell('--os-auth-strategy fake '
'vnfd-list') 'vnfd-list')
self.assertFalse(stdout) self.assertFalse(stdout)
self.assertEqual('You must provide a service URL via '
'either --os-url or env[OS_URL]', stderr.strip())
def test_auth(self): def test_auth(self):
tacker_shell = openstack_shell.TackerShell('1.0') with mock.patch.object(openstack_shell.TackerShell,
self.addCleanup(self.mox.UnsetStubs) 'run_subcommand'), \
self.mox.StubOutWithMock(clientmanager.ClientManager, '__init__') mock.patch.object(session, 'Session'), \
self.mox.StubOutWithMock(tacker_shell, 'run_subcommand') mock.patch.object(clientmanager, 'ClientManager') as mock_cmgr:
clientmanager.ClientManager.__init__(
shell = openstack_shell.TackerShell(DEFAULT_API_VERSION)
shell.options = mock.Mock()
auth_session = shell._get_keystone_session()
cmdline = ('--os-username test '
'--os-password test '
'--os-tenant-name test '
'--os-auth-url http://127.0.0.1:5000/ '
'--os-auth-strategy keystone vnfd-list')
shell.authenticate_user()
shell.run(cmdline.split())
mock_cmgr.assert_called_with(
raise_errors=False, retries=0, timeout=None,
token='', url='', auth_url='http://127.0.0.1:5000/', token='', url='', auth_url='http://127.0.0.1:5000/',
tenant_name='test', tenant_id='tenant_id', tenant_name='test', tenant_id='tenant_id',
username='test', user_id='', username='test', user_id='',
@ -131,38 +144,26 @@ class ShellTest(testtools.TestCase):
auth_strategy='keystone', auth_strategy='keystone',
service_type='nfv-orchestration', service_type='nfv-orchestration',
endpoint_type='publicURL', insecure=False, ca_cert=None, endpoint_type='publicURL', insecure=False, ca_cert=None,
log_credentials=True) log_credentials=True, session=auth_session, auth=auth_session.auth)
tacker_shell.run_subcommand(['vnfd-list'])
self.mox.ReplayAll()
cmdline = ('--os-username test '
'--os-password test '
'--os-tenant-name test '
'--os-auth-url http://127.0.0.1:5000/ '
'--os-auth-strategy keystone vnfd-list')
tacker_shell.run(cmdline.split())
self.mox.VerifyAll()
def test_build_option_parser(self): def test_build_option_parser(self):
tacker_shell = openstack_shell.TackerShell('1.0') tacker_shell = openstack_shell.TackerShell(DEFAULT_API_VERSION)
result = tacker_shell.build_option_parser('descr', '1.0') result = tacker_shell.build_option_parser('descr', DEFAULT_API_VERSION)
self.assertIsInstance(result, argparse.ArgumentParser) self.assertIsInstance(result, argparse.ArgumentParser)
def test_main_with_unicode(self): @mock.patch.object(openstack_shell.TackerShell, 'run')
self.mox.StubOutClassWithMocks(openstack_shell, 'TackerShell') def test_main_with_unicode(self, mock_run):
qshell_mock = openstack_shell.TackerShell('1.0') mock_run.return_value = 0
unicode_text = u'\u7f51\u7edc' unicode_text = u'\u7f51\u7edc'
argv = ['net-list', unicode_text, unicode_text.encode('utf-8')] argv = ['net-list', unicode_text, unicode_text.encode('utf-8')]
qshell_mock.run([u'net-list', unicode_text,
unicode_text]).AndReturn(0)
self.mox.ReplayAll()
ret = openstack_shell.main(argv=argv) ret = openstack_shell.main(argv=argv)
self.mox.VerifyAll() mock_run.assert_called_once_with([u'net-list', unicode_text,
self.mox.UnsetStubs() unicode_text])
self.assertEqual(ret, 0) self.assertEqual(0, ret)
def test_endpoint_option(self): def test_endpoint_option(self):
shell = openstack_shell.TackerShell('1.0') shell = openstack_shell.TackerShell(DEFAULT_API_VERSION)
parser = shell.build_option_parser('descr', '1.0') parser = shell.build_option_parser('descr', DEFAULT_API_VERSION)
# Neither $OS_ENDPOINT_TYPE nor --endpoint-type # Neither $OS_ENDPOINT_TYPE nor --endpoint-type
namespace = parser.parse_args([]) namespace = parser.parse_args([])
@ -177,8 +178,8 @@ class ShellTest(testtools.TestCase):
"public") "public")
self.useFixture(fixture) self.useFixture(fixture)
shell = openstack_shell.TackerShell('1.0') shell = openstack_shell.TackerShell(DEFAULT_API_VERSION)
parser = shell.build_option_parser('descr', '1.0') parser = shell.build_option_parser('descr', DEFAULT_API_VERSION)
# $OS_ENDPOINT_TYPE but not --endpoint-type # $OS_ENDPOINT_TYPE but not --endpoint-type
namespace = parser.parse_args([]) namespace = parser.parse_args([])

128
tackerclient/tests/unit/test_ssl.py
View File

@ -14,12 +14,13 @@
# under the License. # under the License.
import fixtures import fixtures
import mox from keystoneclient import session
import mock
import requests import requests
import testtools import testtools
from tackerclient.client import HTTPClient from tackerclient import client
from tackerclient.common.clientmanager import ClientManager from tackerclient.common import clientmanager
from tackerclient.common import exceptions from tackerclient.common import exceptions
from tackerclient import shell as openstack_shell from tackerclient import shell as openstack_shell
@ -29,6 +30,7 @@ END_URL = 'test_url'
METHOD = 'GET' METHOD = 'GET'
URL = 'http://test.test:1234/v1.0/' URL = 'http://test.test:1234/v1.0/'
CA_CERT = '/tmp/test/path' CA_CERT = '/tmp/test/path'
DEFAULT_API_VERSION = '1.0'
class TestSSL(testtools.TestCase): class TestSSL(testtools.TestCase):
@ -37,106 +39,44 @@ class TestSSL(testtools.TestCase):
self.useFixture(fixtures.EnvironmentVariable('OS_TOKEN', AUTH_TOKEN)) self.useFixture(fixtures.EnvironmentVariable('OS_TOKEN', AUTH_TOKEN))
self.useFixture(fixtures.EnvironmentVariable('OS_URL', END_URL)) self.useFixture(fixtures.EnvironmentVariable('OS_URL', END_URL))
self.addCleanup(mock.patch.stopall)
self.mox = mox.Mox() def _test_verify_client_manager(self, cacert):
self.addCleanup(self.mox.UnsetStubs) with mock.patch.object(session, 'Session'), \
mock.patch.object(clientmanager, 'ClientManager') as mock_cmgr:
mock_cmgr.return_value = 0
shell = openstack_shell.TackerShell(DEFAULT_API_VERSION)
shell.options = mock.Mock()
auth_session = shell._get_keystone_session()
shell.run(cacert)
mock_cmgr.assert_called_with(
api_version={'nfv-orchestration': '1.0'},
auth=auth_session.auth, auth_strategy='keystone',
auth_url='', ca_cert=CA_CERT, endpoint_type='publicURL',
insecure=False, log_credentials=True, password='',
raise_errors=False, region_name='', retries=0,
service_type='nfv-orchestration', session=auth_session,
tenant_id='', tenant_name='', timeout=None,
token='test_token', url='test_url', user_id='', username='')
def test_ca_cert_passed(self): def test_ca_cert_passed(self):
self.mox.StubOutWithMock(ClientManager, '__init__') cacert = ['--os-cacert', CA_CERT]
self.mox.StubOutWithMock(openstack_shell.TackerShell, 'interact') self._test_verify_client_manager(cacert)
ClientManager.__init__(
ca_cert=CA_CERT,
# we are not really interested in other args
api_version=mox.IgnoreArg(),
auth_strategy=mox.IgnoreArg(),
auth_url=mox.IgnoreArg(),
service_type=mox.IgnoreArg(),
endpoint_type=mox.IgnoreArg(),
insecure=mox.IgnoreArg(),
password=mox.IgnoreArg(),
region_name=mox.IgnoreArg(),
tenant_id=mox.IgnoreArg(),
tenant_name=mox.IgnoreArg(),
token=mox.IgnoreArg(),
url=mox.IgnoreArg(),
username=mox.IgnoreArg(),
user_id=mox.IgnoreArg(),
log_credentials=mox.IgnoreArg(),
)
openstack_shell.TackerShell.interact().AndReturn(0)
self.mox.ReplayAll()
openstack_shell.TackerShell('1.0').run(['--os-cacert', CA_CERT])
self.mox.VerifyAll()
def test_ca_cert_passed_as_env_var(self): def test_ca_cert_passed_as_env_var(self):
self.useFixture(fixtures.EnvironmentVariable('OS_CACERT', CA_CERT)) self.useFixture(fixtures.EnvironmentVariable('OS_CACERT', CA_CERT))
self._test_verify_client_manager([])
self.mox.StubOutWithMock(ClientManager, '__init__') @mock.patch.object(client.HTTPClient, 'request')
self.mox.StubOutWithMock(openstack_shell.TackerShell, 'interact') def test_proper_exception_is_raised_when_cert_validation_fails(self,
mock_req):
ClientManager.__init__( http = client.HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL)
ca_cert=CA_CERT, mock_req.side_effect = requests.exceptions.SSLError()
# we are not really interested in other args
api_version=mox.IgnoreArg(),
auth_strategy=mox.IgnoreArg(),
auth_url=mox.IgnoreArg(),
service_type=mox.IgnoreArg(),
endpoint_type=mox.IgnoreArg(),
insecure=mox.IgnoreArg(),
password=mox.IgnoreArg(),
region_name=mox.IgnoreArg(),
tenant_id=mox.IgnoreArg(),
tenant_name=mox.IgnoreArg(),
token=mox.IgnoreArg(),
url=mox.IgnoreArg(),
username=mox.IgnoreArg(),
user_id=mox.IgnoreArg(),
log_credentials=mox.IgnoreArg(),
)
openstack_shell.TackerShell.interact().AndReturn(0)
self.mox.ReplayAll()
openstack_shell.TackerShell('1.0').run([])
self.mox.VerifyAll()
def test_client_manager_properly_creates_httpclient_instance(self):
self.mox.StubOutWithMock(HTTPClient, '__init__')
HTTPClient.__init__(
ca_cert=CA_CERT,
# we are not really interested in other args
auth_strategy=mox.IgnoreArg(),
auth_url=mox.IgnoreArg(),
endpoint_url=mox.IgnoreArg(),
insecure=mox.IgnoreArg(),
password=mox.IgnoreArg(),
region_name=mox.IgnoreArg(),
tenant_name=mox.IgnoreArg(),
token=mox.IgnoreArg(),
username=mox.IgnoreArg(),
)
self.mox.ReplayAll()
version = {'nfv-orchestration': '1.0'}
ClientManager(ca_cert=CA_CERT,
api_version=version,
url=END_URL,
token=AUTH_TOKEN).tacker
self.mox.VerifyAll()
def test_proper_exception_is_raised_when_cert_validation_fails(self):
http = HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL)
self.mox.StubOutWithMock(HTTPClient, 'request')
HTTPClient.request(
URL, METHOD, headers=mox.IgnoreArg()
).AndRaise(requests.exceptions.SSLError)
self.mox.ReplayAll()
self.assertRaises( self.assertRaises(
exceptions.SslCertificateValidationError, exceptions.SslCertificateValidationError,
http._cs_request, http._cs_request,
URL, METHOD URL, METHOD
) )
self.mox.VerifyAll()