diff --git a/.testr.conf b/.testr.conf index 159cb2b8..316dc9ec 100644 --- a/.testr.conf +++ b/.testr.conf @@ -1,4 +1,4 @@ [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_list_option=--list diff --git a/tackerclient/tests/unit/test_auth.py b/tackerclient/tests/unit/test_auth.py index d41dee55..80b4d15f 100644 --- a/tackerclient/tests/unit/test_auth.py +++ b/tackerclient/tests/unit/test_auth.py @@ -19,13 +19,12 @@ import json import uuid from keystoneclient import exceptions as k_exceptions -import mox +import mock import requests import testtools from tackerclient import client from tackerclient.common import exceptions -from tackerclient.common import utils USERNAME = 'testuser' @@ -71,39 +70,44 @@ ENDPOINTS_RESULT = { def get_response(status_code, headers=None): - response = mox.Mox().CreateMock(requests.Response) + response = mock.Mock().CreateMock(requests.Response) response.headers = headers or {} response.status_code = status_code 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): def setUp(self): """Prepare the test environment.""" super(CLITestAuthNoAuth, self).setUp() - self.mox = mox.Mox() self.client = client.HTTPClient(username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD, endpoint_url=ENDPOINT_URL, auth_strategy=NOAUTH, region_name=REGION) - self.addCleanup(self.mox.VerifyAll) - self.addCleanup(self.mox.UnsetStubs) + self.addCleanup(mock.patch.stopall) - def test_get_noauth(self): - self.mox.StubOutWithMock(self.client, "request") + @mock.patch('tackerclient.client.HTTPClient.request') + def test_get_noauth(self, mock_request): - res200 = get_response(200) - - self.client.request( - mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', - headers=mox.IsA(dict), - ).AndReturn((res200, '')) - self.mox.ReplayAll() - - self.client.do_request('/resource', 'GET') + mock_request.return_value = (resp_200, '') + self.client.do_request('/resource', 'GET', + headers=headers) + mock_request.assert_called_once_with( + ENDPOINT_URL + '/resource', + 'GET', + headers=headers) self.assertEqual(self.client.endpoint_url, ENDPOINT_URL) @@ -117,14 +121,12 @@ class CLITestAuthKeystone(testtools.TestCase): def setUp(self): """Prepare the test environment.""" super(CLITestAuthKeystone, self).setUp() - self.mox = mox.Mox() self.client = client.HTTPClient(username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD, auth_url=AUTH_URL, region_name=REGION) - self.addCleanup(self.mox.VerifyAll) - self.addCleanup(self.mox.UnsetStubs) + self.addCleanup(mock.patch.stopall) def test_reused_token_get_auth_info(self): """Test Client.get_auth_info(). @@ -144,69 +146,53 @@ class CLITestAuthKeystone(testtools.TestCase): 'endpoint_url': self.client.endpoint_url} self.assertEqual(client_.get_auth_info(), expected) - def test_get_token(self): - self.mox.StubOutWithMock(self.client, "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.patch('tackerclient.client.HTTPClient.request') + def test_get_token(self, mock_request): + mock_request.return_value = (resp_200, json.dumps(KS_TOKEN_RESULT)) 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.auth_token, TOKEN) - def test_refresh_token(self): - self.mox.StubOutWithMock(self.client, "request") + @mock.patch('tackerclient.client.HTTPClient.request') + def test_refresh_token(self, mock_request): self.client.auth_token = TOKEN self.client.endpoint_url = ENDPOINT_URL - res200 = get_response(200) - res401 = get_response(401) - # If a token is expired, tacker server retruns 401 - self.client.request( - mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', - headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN) - ).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') + mock_request.return_value = (resp_401, '') + self.assertRaises(exceptions.Unauthorized, + self.client.do_request, + '/resource', + 'GET') + + mock_request.return_value = (resp_200, json.dumps(KS_TOKEN_RESULT)) + self.client.do_request('/resource', 'GET') + mock_request.assert_called_with( + ENDPOINT_URL + '/resource', 'GET', + headers=expected_headers) + + @mock.patch('tackerclient.client.HTTPClient.request') + 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_token = TOKEN self.client.endpoint_url = ENDPOINT_URL - res401 = get_response(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() + # If a token is expired, tacker server retruns 401 + mock_request.return_value = (resp_401, '') self.assertRaises(exceptions.NoAuthURLProvided, self.client.do_request, '/resource', '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): # Handle the case when auth_url is not provided @@ -214,85 +200,70 @@ class CLITestAuthKeystone(testtools.TestCase): self.assertRaises(exceptions.NoAuthURLProvided, self.client._get_endpoint_url) - def test_get_endpoint_url(self): - self.mox.StubOutWithMock(self.client, "request") + @mock.patch('tackerclient.client.HTTPClient.request') + def test_get_endpoint_url(self, mock_request): self.client.auth_token = TOKEN - res200 = get_response(200) - - 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() + mock_request.return_value = (resp_200, json.dumps(ENDPOINTS_RESULT)) 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( username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD, auth_url=AUTH_URL, region_name=REGION, endpoint_url=ENDPOINT_OVERRIDE) self.assertEqual(self.client.endpoint_url, ENDPOINT_OVERRIDE) - self.mox.StubOutWithMock(self.client, "request") - self.client.auth_token = TOKEN - res200 = get_response(200) + mock_request.return_value = (resp_200, '') - self.client.request( - mox.StrContains(ENDPOINT_OVERRIDE + '/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', + headers=headers) + mock_request.assert_called_with( + ENDPOINT_OVERRIDE + '/resource', 'GET', + headers=headers) 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( username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD, auth_url=AUTH_URL, region_name=REGION, endpoint_type='otherURL') - self.mox.StubOutWithMock(self.client, "request") self.client.auth_token = TOKEN - res200 = get_response(200) - - 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() + mock_request.return_value = (resp_200, json.dumps(ENDPOINTS_RESULT)) self.assertRaises(exceptions.EndpointTypeNotFound, self.client.do_request, '/resource', '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): - self.mox.StubOutWithMock(self.client, "request") + @mock.patch('tackerclient.client.HTTPClient.request') + def test_get_endpoint_url_failed(self, mock_request): self.client.auth_token = TOKEN + self.client.auth_url = AUTH_URL + '/tokens/%s/endpoints' % TOKEN - res200 = get_response(200) - res401 = get_response(401) - - self.client.request( - mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), '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') + mock_request.return_value = (resp_401, '') + self.assertRaises(exceptions.Unauthorized, + self.client.do_request, + '/resource', + 'GET') def test_endpoint_type(self): resources = copy.deepcopy(KS_TOKEN_RESULT) @@ -342,32 +313,25 @@ class CLITestAuthKeystone(testtools.TestCase): self.client._extract_service_catalog, resources) - def test_strip_credentials_from_log(self): - def verify_no_credentials(kwargs): - return ('REDACTED' in kwargs['body']) and ( - self.client.password not in kwargs['body']) + @mock.patch('tackerclient.client.HTTPClient.request') + @mock.patch('tackerclient.common.utils.http_log_req') + def test_strip_credentials_from_log(self, mock_http_log_req, + mock_request,): - def verify_credentials(body): - return 'REDACTED' not in body and self.client.password in body + body = ('{"auth": {"tenantId": "testtenant_id",' + '"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") - self.mox.StubOutWithMock(utils, "http_log_req") + mock_request.return_value = (resp_200, json.dumps(KS_TOKEN_RESULT)) + self.client.do_request('/resource', 'GET', body=body) - res200 = get_response(200) - - utils.http_log_req(mox.IgnoreArg(), mox.IgnoreArg(), mox.Func( - 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') + args, kwargs = mock_http_log_req.call_args + # Check that credentials are stripped while logging. + self.assertEqual(_headers, args[2]) class CLITestAuthKeystoneWithId(CLITestAuthKeystone): diff --git a/tackerclient/tests/unit/test_cli10.py b/tackerclient/tests/unit/test_cli10.py index 667c6261..3e31832e 100644 --- a/tackerclient/tests/unit/test_cli10.py +++ b/tackerclient/tests/unit/test_cli10.py @@ -587,7 +587,7 @@ class CLITestV10Base(testtools.TestCase): resstr = self.client.serialize(expected_res) path = getattr(self.client, resource + "_path") with mock.patch.object(self.client.httpclient, 'request') as\ - mock_req: + mock_req: mock_req.return_value = (MyResp(200), resstr) args.extend(['--request-format', self.format]) cmd_parser = cmd.get_parser("show_" + resource) @@ -645,36 +645,27 @@ class ClientV1TestJson(CLITestV10Base): def test_do_request_unicode(self): self.client.format = self.format unicode_text = u'\u7f51\u7edc' - # url with unicode action = u'/test' - expected_action = action.encode('utf-8') - # query string with unicode params = {'test': unicode_text} - expect_query = urllib.urlencode({'test': - unicode_text.encode('utf-8')}) - # request body with unicode body = params expect_body = self.client.serialize(body) - # headers with unicode 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: mock_req.return_value = (MyResp(200), expect_body) res_body = self.client.do_request('PUT', action, body=body, params=params) - mock_req.assert_called_once_with( - end_url(expected_action, query=expect_query, - format=self.format), - 'PUT', body=expect_body, - headers=test_utils.ContainsKeyValue('X-Auth-Token', - expected_auth_token)) + expected_uri = u'localurl/v1.0/test.json?test=%E7%BD%91%E7%BB%9C' + mock_req.assert_called_with( + expected_uri, 'PUT', body=expect_body, + headers={'X-Auth-Token': unicode_text, + 'User-Agent': 'python-tackerclient'}) # test response with unicode self.assertEqual(res_body, body) def test_do_request_error_without_response_body(self): self.client.format = self.format params = {'test': 'value'} - expect_query = urllib.urlencode(params) + expect_query = urlparse.urlencode(params) self.client.httpclient.auth_token = 'token' with mock.patch.object(self.client.httpclient, 'request') as mock_req: mock_req.return_value = (MyResp(400, reason='An error'), '') @@ -692,9 +683,9 @@ class ClientV1TestJson(CLITestV10Base): class CLITestV10ExceptionHandler(CLITestV10Base): def _test_exception_handler_v10( - self, expected_exception, status_code, expected_msg, - error_type=None, error_msg=None, error_detail=None, - error_content=None): + self, expected_exception, status_code, expected_msg, + error_type=None, error_msg=None, error_detail=None, + error_content=None): if error_content is None: error_content = {'TackerError': {'type': error_type, 'message': error_msg, @@ -704,13 +695,15 @@ class CLITestV10ExceptionHandler(CLITestV10Base): client.exception_handler_v10, status_code, error_content) self.assertEqual(status_code, e.status_code) + self.assertEqual(expected_exception.__name__, + e.__class__.__name__) if expected_msg is None: if error_detail: expected_msg = '\n'.join([error_msg, error_detail]) else: 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): err_msg = ('Unable to complete operation for network ' diff --git a/tackerclient/tests/unit/test_http.py b/tackerclient/tests/unit/test_http.py index 6aa0fade..196eee44 100644 --- a/tackerclient/tests/unit/test_http.py +++ b/tackerclient/tests/unit/test_http.py @@ -13,7 +13,7 @@ # License for the specific language governing permissions and limitations # under the License. -import mox +import mock import testtools from tackerclient.client import HTTPClient @@ -25,60 +25,47 @@ AUTH_TOKEN = 'test_token' END_URL = 'test_url' METHOD = 'GET' URL = 'http://test.test:1234/v1.0/test' +headers = {'User-Agent': 'python-tackerclient'} class TestHTTPClient(testtools.TestCase): + def setUp(self): + super(TestHTTPClient, self).setUp() - - self.mox = mox.Mox() - self.mox.StubOutWithMock(HTTPClient, 'request') - self.addCleanup(self.mox.UnsetStubs) - + self.addCleanup(mock.patch.stopall) self.http = HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL) - def test_request_error(self): - HTTPClient.request( - URL, METHOD, headers=mox.IgnoreArg() - ).AndRaise(Exception('error msg')) - self.mox.ReplayAll() + @mock.patch('tackerclient.client.HTTPClient.request') + def test_request_error(self, mock_request): + mock_request.side_effect = Exception('error msg') self.assertRaises( exceptions.ConnectionFailed, self.http._cs_request, 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' - - HTTPClient.request( - URL, METHOD, headers=mox.IgnoreArg() - ).AndReturn(rv_should_be) - self.mox.ReplayAll() - + mock_request.return_value = rv_should_be self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD)) - self.mox.VerifyAll() - def test_request_unauthorized(self): - rv_should_be = MyResp(401), 'unauthorized message' - HTTPClient.request( - URL, METHOD, headers=mox.IgnoreArg() - ).AndReturn(rv_should_be) - self.mox.ReplayAll() + @mock.patch('tackerclient.client.HTTPClient.request') + def test_request_unauthorized(self, mock_request): + + mock_request.return_value = MyResp(401), 'unauthorized message' e = self.assertRaises(exceptions.Unauthorized, self.http._cs_request, URL, METHOD) 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' - HTTPClient.request( - URL, METHOD, headers=mox.IgnoreArg() - ).AndReturn(rv_should_be) - self.mox.ReplayAll() - + mock_request.return_value = rv_should_be self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD)) - self.mox.VerifyAll() diff --git a/tackerclient/tests/unit/test_shell.py b/tackerclient/tests/unit/test_shell.py index b9b9ac66..ab9bd995 100644 --- a/tackerclient/tests/unit/test_shell.py +++ b/tackerclient/tests/unit/test_shell.py @@ -21,7 +21,8 @@ import six import sys import fixtures -import mox +from keystoneclient import session +import mock import testtools 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_TOKEN = '3bcc3d3a03f44e3d8377f9247b0ad155' DEFAULT_URL = 'http://tacker.example.org:9890/' +DEFAULT_API_VERSION = '1.0' class ShellTest(testtools.TestCase): @@ -50,7 +52,6 @@ class ShellTest(testtools.TestCase): # Patch os.environ to avoid required auth info. def setUp(self): super(ShellTest, self).setUp() - self.mox = mox.Mox() for var in self.FAKE_ENV: self.useFixture( fixtures.EnvironmentVariable( @@ -63,7 +64,7 @@ class ShellTest(testtools.TestCase): try: sys.stdout = six.StringIO() sys.stderr = six.StringIO() - _shell = openstack_shell.TackerShell('1.0') + _shell = openstack_shell.TackerShell(DEFAULT_API_VERSION) _shell.run(argstr.split()) except SystemExit: exc_type, exc_value, exc_traceback = sys.exc_info() @@ -103,7 +104,7 @@ class ShellTest(testtools.TestCase): def test_help_command(self): required = 'usage:' - help_text, stderr = self.shell('help network-create') + help_text, stderr = self.shell('help vnfd-create') self.assertThat( help_text, matchers.MatchesRegex(required)) @@ -114,15 +115,27 @@ class ShellTest(testtools.TestCase): stdout, stderr = self.shell('--os-auth-strategy fake ' 'vnfd-list') 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): - tacker_shell = openstack_shell.TackerShell('1.0') - self.addCleanup(self.mox.UnsetStubs) - self.mox.StubOutWithMock(clientmanager.ClientManager, '__init__') - self.mox.StubOutWithMock(tacker_shell, 'run_subcommand') - clientmanager.ClientManager.__init__( + with mock.patch.object(openstack_shell.TackerShell, + 'run_subcommand'), \ + mock.patch.object(session, 'Session'), \ + mock.patch.object(clientmanager, 'ClientManager') as mock_cmgr: + + 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/', tenant_name='test', tenant_id='tenant_id', username='test', user_id='', @@ -131,38 +144,26 @@ class ShellTest(testtools.TestCase): auth_strategy='keystone', service_type='nfv-orchestration', endpoint_type='publicURL', insecure=False, ca_cert=None, - log_credentials=True) - 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() + log_credentials=True, session=auth_session, auth=auth_session.auth) def test_build_option_parser(self): - tacker_shell = openstack_shell.TackerShell('1.0') - result = tacker_shell.build_option_parser('descr', '1.0') + tacker_shell = openstack_shell.TackerShell(DEFAULT_API_VERSION) + result = tacker_shell.build_option_parser('descr', DEFAULT_API_VERSION) self.assertIsInstance(result, argparse.ArgumentParser) - def test_main_with_unicode(self): - self.mox.StubOutClassWithMocks(openstack_shell, 'TackerShell') - qshell_mock = openstack_shell.TackerShell('1.0') + @mock.patch.object(openstack_shell.TackerShell, 'run') + def test_main_with_unicode(self, mock_run): + mock_run.return_value = 0 unicode_text = u'\u7f51\u7edc' 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) - self.mox.VerifyAll() - self.mox.UnsetStubs() - self.assertEqual(ret, 0) + mock_run.assert_called_once_with([u'net-list', unicode_text, + unicode_text]) + self.assertEqual(0, ret) def test_endpoint_option(self): - shell = openstack_shell.TackerShell('1.0') - parser = shell.build_option_parser('descr', '1.0') + shell = openstack_shell.TackerShell(DEFAULT_API_VERSION) + parser = shell.build_option_parser('descr', DEFAULT_API_VERSION) # Neither $OS_ENDPOINT_TYPE nor --endpoint-type namespace = parser.parse_args([]) @@ -177,8 +178,8 @@ class ShellTest(testtools.TestCase): "public") self.useFixture(fixture) - shell = openstack_shell.TackerShell('1.0') - parser = shell.build_option_parser('descr', '1.0') + shell = openstack_shell.TackerShell(DEFAULT_API_VERSION) + parser = shell.build_option_parser('descr', DEFAULT_API_VERSION) # $OS_ENDPOINT_TYPE but not --endpoint-type namespace = parser.parse_args([]) diff --git a/tackerclient/tests/unit/test_ssl.py b/tackerclient/tests/unit/test_ssl.py index 95c522b4..b9b278bc 100644 --- a/tackerclient/tests/unit/test_ssl.py +++ b/tackerclient/tests/unit/test_ssl.py @@ -14,12 +14,13 @@ # under the License. import fixtures -import mox +from keystoneclient import session +import mock import requests import testtools -from tackerclient.client import HTTPClient -from tackerclient.common.clientmanager import ClientManager +from tackerclient import client +from tackerclient.common import clientmanager from tackerclient.common import exceptions from tackerclient import shell as openstack_shell @@ -29,6 +30,7 @@ END_URL = 'test_url' METHOD = 'GET' URL = 'http://test.test:1234/v1.0/' CA_CERT = '/tmp/test/path' +DEFAULT_API_VERSION = '1.0' 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_URL', END_URL)) + self.addCleanup(mock.patch.stopall) - self.mox = mox.Mox() - self.addCleanup(self.mox.UnsetStubs) + def _test_verify_client_manager(self, cacert): + 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): - self.mox.StubOutWithMock(ClientManager, '__init__') - self.mox.StubOutWithMock(openstack_shell.TackerShell, 'interact') - - 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() + cacert = ['--os-cacert', CA_CERT] + self._test_verify_client_manager(cacert) def test_ca_cert_passed_as_env_var(self): self.useFixture(fixtures.EnvironmentVariable('OS_CACERT', CA_CERT)) + self._test_verify_client_manager([]) - self.mox.StubOutWithMock(ClientManager, '__init__') - self.mox.StubOutWithMock(openstack_shell.TackerShell, 'interact') - - 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([]) - 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() - + @mock.patch.object(client.HTTPClient, 'request') + def test_proper_exception_is_raised_when_cert_validation_fails(self, + mock_req): + http = client.HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL) + mock_req.side_effect = requests.exceptions.SSLError() self.assertRaises( exceptions.SslCertificateValidationError, http._cs_request, URL, METHOD ) - self.mox.VerifyAll()