From 5822d619e56ece42b711fdc378de772ebda65f38 Mon Sep 17 00:00:00 2001 From: Jamie Lennox Date: Thu, 30 Oct 2014 17:17:17 +0100 Subject: Use requests_mock instead of mox Kill off mox and use requests_mock. There are fixtures available for everything we want to test - there is no reason to resort to using mox. This cleans up and makes clearer a lot of tests. Change-Id: I2633bcaf36388fb4db1de4cd75d5ccd76f961509 --- neutronclient/tests/unit/test_auth.py | 282 ++++++++++++---------------------- neutronclient/tests/unit/test_http.py | 67 +++----- 2 files changed, 121 insertions(+), 228 deletions(-) diff --git a/neutronclient/tests/unit/test_auth.py b/neutronclient/tests/unit/test_auth.py index bd8a492..7cb8c56 100644 --- a/neutronclient/tests/unit/test_auth.py +++ b/neutronclient/tests/unit/test_auth.py @@ -15,25 +15,22 @@ # import json +import logging import uuid import fixtures -from mox3 import mox from oslo.serialization import jsonutils -import requests -import requests_mock -import six +from requests_mock.contrib import fixture as mock_fixture import testtools from keystoneclient.auth.identity import v2 as ks_v2_auth from keystoneclient.auth.identity import v3 as ks_v3_auth from keystoneclient import exceptions as ks_exceptions -from keystoneclient import fixture +from keystoneclient import fixture as ks_fixture from keystoneclient import session from neutronclient import client from neutronclient.common import exceptions -from neutronclient.common import utils USERNAME = 'testuser' @@ -41,16 +38,16 @@ USER_ID = 'testuser_id' TENANT_NAME = 'testtenant' TENANT_ID = 'testtenant_id' PASSWORD = 'password' -ENDPOINT_URL = 'localurl' -PUBLIC_ENDPOINT_URL = 'public_%s' % ENDPOINT_URL -ADMIN_ENDPOINT_URL = 'admin_%s' % ENDPOINT_URL -INTERNAL_ENDPOINT_URL = 'internal_%s' % ENDPOINT_URL -ENDPOINT_OVERRIDE = 'otherurl' +ENDPOINT_URL = 'http://localurl' +PUBLIC_ENDPOINT_URL = '%s/public' % ENDPOINT_URL +ADMIN_ENDPOINT_URL = '%s/admin' % ENDPOINT_URL +INTERNAL_ENDPOINT_URL = '%s/internal' % ENDPOINT_URL +ENDPOINT_OVERRIDE = 'http://otherurl' TOKENID = uuid.uuid4().hex REGION = 'RegionOne' NOAUTH = 'noauth' -KS_TOKEN_RESULT = fixture.V2Token() +KS_TOKEN_RESULT = ks_fixture.V2Token() KS_TOKEN_RESULT.set_scope() _s = KS_TOKEN_RESULT.add_service('network', 'Neutron Service') _s.add_endpoint(ENDPOINT_URL, region=REGION) @@ -71,8 +68,8 @@ BASE_URL = "http://keystone.example.com:5000/" V2_URL = "%sv2.0" % BASE_URL V3_URL = "%sv3" % BASE_URL -_v2 = fixture.V2Discovery(V2_URL) -_v3 = fixture.V3Discovery(V3_URL) +_v2 = ks_fixture.V2Discovery(V2_URL) +_v3 = ks_fixture.V3Discovery(V3_URL) V3_VERSION_LIST = jsonutils.dumps({'versions': {'values': [_v2, _v3]}}) @@ -80,15 +77,8 @@ V2_VERSION_ENTRY = {'version': _v2} V3_VERSION_ENTRY = {'version': _v3} -def get_response(status_code, headers=None): - response = mox.Mox().CreateMock(requests.Response) - response.headers = headers or {} - response.status_code = status_code - return response - - def setup_keystone_v2(mrequests): - v2_token = fixture.V2Token(token_id=TOKENID) + v2_token = ks_fixture.V2Token(token_id=TOKENID) service = v2_token.add_service('network') service.add_endpoint(PUBLIC_ENDPOINT_URL, region=REGION) @@ -106,7 +96,7 @@ def setup_keystone_v3(mrequests): V3_URL, json=V3_VERSION_ENTRY) - v3_token = fixture.V3Token() + v3_token = ks_fixture.V3Token() service = v3_token.add_service('network') service.add_standard_endpoints(public=PUBLIC_ENDPOINT_URL, admin=ADMIN_ENDPOINT_URL, @@ -135,28 +125,21 @@ class CLITestAuthNoAuth(testtools.TestCase): def setUp(self): """Prepare the test environment.""" super(CLITestAuthNoAuth, self).setUp() - self.mox = mox.Mox() + + self.requests = self.useFixture(mock_fixture.Fixture()) + 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) def test_get_noauth(self): - self.mox.StubOutWithMock(self.client, "request") - - res200 = get_response(200) - - self.client.request( - mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', - headers=mox.IsA(dict), - ).AndReturn((res200, '')) - self.mox.ReplayAll() - + url = ENDPOINT_URL + '/resource' + self.requests.get(ENDPOINT_URL + '/resource') self.client.do_request('/resource', 'GET') + self.assertEqual(url, self.requests.last_request.url) self.assertEqual(self.client.endpoint_url, ENDPOINT_URL) @@ -165,11 +148,13 @@ class CLITestAuthKeystone(testtools.TestCase): def setUp(self): """Prepare the test environment.""" super(CLITestAuthKeystone, self).setUp() - self.mox = mox.Mox() for var in ('http_proxy', 'HTTP_PROXY'): self.useFixture(fixtures.EnvironmentVariableFixture(var)) + self.logger = self.useFixture(fixtures.FakeLogger(level=logging.DEBUG)) + self.requests = self.useFixture(mock_fixture.Fixture()) + self.client = client.construct_http_client( username=USERNAME, tenant_name=TENANT_NAME, @@ -177,9 +162,6 @@ class CLITestAuthKeystone(testtools.TestCase): auth_url=AUTH_URL, region_name=REGION) - self.addCleanup(self.mox.VerifyAll) - self.addCleanup(self.mox.UnsetStubs) - def test_reused_token_get_auth_info(self): """Test that Client.get_auth_info() works even if client was instantiated with predefined token. @@ -197,9 +179,8 @@ class CLITestAuthKeystone(testtools.TestCase): 'endpoint_url': self.client.endpoint_url} self.assertEqual(client_.get_auth_info(), expected) - @requests_mock.Mocker() - def test_get_token(self, mrequests): - auth_session, auth_plugin = setup_keystone_v2(mrequests) + def test_get_token(self): + auth_session, auth_plugin = setup_keystone_v2(self.requests) self.client = client.construct_http_client( username=USERNAME, @@ -210,61 +191,50 @@ class CLITestAuthKeystone(testtools.TestCase): session=auth_session, auth=auth_plugin) - self.mox.StubOutWithMock(self.client, "request") - res200 = get_response(200) - - self.client.request( - '/resource', 'GET', - authenticated=True - ).AndReturn((res200, '')) - - self.mox.ReplayAll() - + m = self.requests.get(PUBLIC_ENDPOINT_URL + '/resource', + request_headers={'X-Auth-Token': TOKENID}) self.client.do_request('/resource', 'GET') + self.assertTrue(m.called) def test_refresh_token(self): - self.mox.StubOutWithMock(self.client, "request") - token_id = uuid.uuid4().hex + text = uuid.uuid4().hex self.client.auth_token = token_id self.client.endpoint_url = ENDPOINT_URL - res200 = get_response(200) - res401 = get_response(401) - - # If a token is expired, neutron server retruns 401 - self.client.request( - mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', - headers=mox.ContainsKeyValue('X-Auth-Token', token_id) - ).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', - KS_TOKEN_RESULT.token_id) - ).AndReturn((res200, '')) - self.mox.ReplayAll() - self.client.do_request('/resource', 'GET') + res_url = ENDPOINT_URL + '/resource' + v2_url = AUTH_URL + '/tokens' + + # token_id gives 401, KS_TOKEN_RESULT gives 200 + self.requests.get(res_url, + request_headers={'X-Auth-Token': token_id}, + status_code=401) + + self.requests.get( + res_url, + text=text, + status_code=200, + request_headers={'X-Auth-Token': KS_TOKEN_RESULT.token_id}) + + self.requests.post(v2_url, json=KS_TOKEN_RESULT) + + resp = self.client.do_request('/resource', 'GET') + + self.assertEqual(text, resp[1]) + self.assertEqual(3, len(self.requests.request_history)) + + self.assertEqual(res_url, self.requests.request_history[0].url) + self.assertEqual(v2_url, self.requests.request_history[1].url) + self.assertEqual(res_url, self.requests.request_history[2].url) def test_refresh_token_no_auth_url(self): - self.mox.StubOutWithMock(self.client, "request") self.client.auth_url = None token_id = uuid.uuid4().hex self.client.auth_token = token_id self.client.endpoint_url = ENDPOINT_URL - res401 = get_response(401) - - # If a token is expired, neutron server returns 401 - self.client.request( - mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', - headers=mox.ContainsKeyValue('X-Auth-Token', token_id) - ).AndReturn((res401, '')) - self.mox.ReplayAll() + self.requests.get(ENDPOINT_URL + '/resource', status_code=401) self.assertRaises(exceptions.NoAuthURLProvided, self.client.do_request, '/resource', @@ -277,24 +247,18 @@ class CLITestAuthKeystone(testtools.TestCase): self.client._get_endpoint_url) def test_get_endpoint_url(self): - self.mox.StubOutWithMock(self.client, "request") - token_id = uuid.uuid4().hex self.client.auth_token = token_id - res200 = get_response(200) - - self.client.request( - mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % token_id), - '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_id) - ).AndReturn((res200, '')) - self.mox.ReplayAll() + self.requests.get(AUTH_URL + '/tokens/%s/endpoints' % token_id, + json=ENDPOINTS_RESULT) + self.requests.get(ENDPOINT_URL + '/resource') + self.client.do_request('/resource', 'GET') + self.assertEqual(token_id, + self.requests.last_request.headers['X-Auth-Token']) + def test_use_given_endpoint_url(self): self.client = client.HTTPClient( username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD, @@ -302,69 +266,49 @@ class CLITestAuthKeystone(testtools.TestCase): endpoint_url=ENDPOINT_OVERRIDE) self.assertEqual(self.client.endpoint_url, ENDPOINT_OVERRIDE) - self.mox.StubOutWithMock(self.client, "request") - token_id = uuid.uuid4().hex - self.client.auth_token = token_id - res200 = get_response(200) - self.client.request( - mox.StrContains(ENDPOINT_OVERRIDE + '/resource'), 'GET', - headers=mox.ContainsKeyValue('X-Auth-Token', token_id) - ).AndReturn((res200, '')) - self.mox.ReplayAll() + self.requests.get(ENDPOINT_OVERRIDE + '/resource') + self.client.do_request('/resource', 'GET') + self.assertEqual(self.client.endpoint_url, ENDPOINT_OVERRIDE) + self.assertEqual(token_id, + self.requests.last_request.headers['X-Auth-Token']) def test_get_endpoint_url_other(self): 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") token_id = uuid.uuid4().hex self.client.auth_token = token_id - res200 = get_response(200) - self.client.request( - mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % token_id), - 'GET', headers=mox.IsA(dict) - ).AndReturn((res200, json.dumps(ENDPOINTS_RESULT))) - self.mox.ReplayAll() + self.requests.get(AUTH_URL + '/tokens/%s/endpoints' % token_id, + json=ENDPOINTS_RESULT) + self.assertRaises(exceptions.EndpointTypeNotFound, self.client.do_request, '/resource', 'GET') def test_get_endpoint_url_failed(self): - self.mox.StubOutWithMock(self.client, "request") - token_id = uuid.uuid4().hex self.client.auth_token = token_id - res200 = get_response(200) - res401 = get_response(401) - - self.client.request( - mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % token_id), - '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', - KS_TOKEN_RESULT.token_id) - ).AndReturn((res200, '')) - self.mox.ReplayAll() + self.requests.get(AUTH_URL + '/tokens/%s/endpoints' % token_id, + status_code=401) + self.requests.post(AUTH_URL + '/tokens', json=KS_TOKEN_RESULT) + m = self.requests.get(ENDPOINT_URL + '/resource') + self.client.do_request('/resource', 'GET') - @requests_mock.Mocker() - def test_endpoint_type(self, mrequests): - auth_session, auth_plugin = setup_keystone_v3(mrequests) + self.assertEqual(KS_TOKEN_RESULT.token_id, + m.last_request.headers['X-Auth-Token']) + + def test_endpoint_type(self): + auth_session, auth_plugin = setup_keystone_v3(self.requests) # Test default behavior is to choose public. self.client = client.construct_http_client( @@ -413,34 +357,16 @@ class CLITestAuthKeystone(testtools.TestCase): self.client.authenticate) 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']) - - def verify_credentials(body): - return 'REDACTED' not in body and self.client.password in body - - self.mox.StubOutWithMock(self.client, "request") - self.mox.StubOutWithMock(utils, "http_log_req") - - res200 = get_response(200) - - utils.http_log_req(mox.IgnoreArg(), mox.IgnoreArg(), mox.Func( - verify_no_credentials)) - self.client.request( - mox.IsA(six.string_types), mox.IsA(six.string_types), - 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(six.string_types), mox.IsA(six.string_types), - headers=mox.IsA(dict) - ).AndReturn((res200, '')) - self.mox.ReplayAll() - + m = self.requests.post(AUTH_URL + '/tokens', json=KS_TOKEN_RESULT) + self.requests.get(ENDPOINT_URL + '/resource') self.client.do_request('/resource', 'GET') + self.assertIn('REDACTED', self.logger.output) + self.assertNotIn(self.client.password, self.logger.output) + + self.assertNotIn('REDACTED', m.last_request.body) + self.assertIn(self.client.password, m.last_request.body) + class CLITestAuthKeystoneWithId(CLITestAuthKeystone): @@ -473,14 +399,10 @@ class TestKeystoneClientVersions(testtools.TestCase): def setUp(self): """Prepare the test environment.""" super(TestKeystoneClientVersions, self).setUp() - self.mox = mox.Mox() - self.addCleanup(self.mox.VerifyAll) - self.addCleanup(self.mox.UnsetStubs) + self.requests = self.useFixture(mock_fixture.Fixture()) - @requests_mock.Mocker() - def test_v2_auth(self, mrequests): - auth_session, auth_plugin = setup_keystone_v2(mrequests) - res200 = get_response(200) + def test_v2_auth(self): + auth_session, auth_plugin = setup_keystone_v2(self.requests) self.client = client.construct_http_client( username=USERNAME, @@ -491,20 +413,14 @@ class TestKeystoneClientVersions(testtools.TestCase): session=auth_session, auth=auth_plugin) - self.mox.StubOutWithMock(self.client, "request") + m = self.requests.get(PUBLIC_ENDPOINT_URL + '/resource') - self.client.request( - '/resource', 'GET', - authenticated=True - ).AndReturn((res200, '')) - - self.mox.ReplayAll() self.client.do_request('/resource', 'GET') - @requests_mock.Mocker() - def test_v3_auth(self, mrequests): - auth_session, auth_plugin = setup_keystone_v3(mrequests) - res200 = get_response(200) + self.assertTrue(m.called) + + def test_v3_auth(self): + auth_session, auth_plugin = setup_keystone_v3(self.requests) self.client = client.construct_http_client( user_id=USER_ID, @@ -515,12 +431,8 @@ class TestKeystoneClientVersions(testtools.TestCase): session=auth_session, auth=auth_plugin) - self.mox.StubOutWithMock(self.client, "request") + m = self.requests.get(PUBLIC_ENDPOINT_URL + '/resource') - self.client.request( - '/resource', 'GET', - authenticated=True - ).AndReturn((res200, '')) - - self.mox.ReplayAll() self.client.do_request('/resource', 'GET') + + self.assertTrue(m.called) diff --git a/neutronclient/tests/unit/test_http.py b/neutronclient/tests/unit/test_http.py index 51c6cdc..6e04b62 100644 --- a/neutronclient/tests/unit/test_http.py +++ b/neutronclient/tests/unit/test_http.py @@ -15,15 +15,13 @@ import abc -from mox3 import mox -import requests_mock +from requests_mock.contrib import fixture as mock_fixture import six import testtools from neutronclient import client from neutronclient.common import exceptions from neutronclient.tests.unit import test_auth -from neutronclient.tests.unit.test_cli20 import MyResp AUTH_TOKEN = 'test_token' @@ -39,10 +37,8 @@ class TestHTTPClientMixin(object): def setUp(self): super(TestHTTPClientMixin, self).setUp() + self.requests = self.useFixture(mock_fixture.Fixture()) self.clazz, self.http = self.initialize() - self.mox = mox.Mox() - self.addCleanup(self.mox.UnsetStubs) - self.mox.StubOutWithMock(self.clazz, '_request') @abc.abstractmethod def initialize(self): @@ -50,12 +46,10 @@ class TestHTTPClientMixin(object): def _test_headers(self, expected_headers, **kwargs): """Test headers.""" - self.clazz._request(URL, METHOD, - body=kwargs.get('body'), - headers=expected_headers) - self.mox.ReplayAll() + self.requests.register_uri(METHOD, URL, + request_headers=expected_headers) self.http.request(URL, METHOD, **kwargs) - self.mox.VerifyAll() + self.assertEqual(kwargs.get('body'), self.requests.last_request.body) def test_headers_without_body(self): self._test_headers({'Accept': 'application/json'}) @@ -82,9 +76,8 @@ class TestHTTPClientMixin(object): class TestSessionClient(TestHTTPClientMixin, testtools.TestCase): - @requests_mock.Mocker() - def initialize(self, mrequests): - session, auth = test_auth.setup_keystone_v2(mrequests) + def initialize(self): + session, auth = test_auth.setup_keystone_v2(self.requests) return [client.SessionClient, client.SessionClient(session=session, auth=auth)] @@ -96,47 +89,35 @@ class TestHTTPClient(TestHTTPClientMixin, testtools.TestCase): client.HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL)] def test_request_error(self): - self.clazz._request( - URL, METHOD, body=None, headers=mox.IgnoreArg() - ).AndRaise(Exception('error msg')) - self.mox.ReplayAll() + def cb(*args, **kwargs): + raise Exception('error msg') + self.requests.get(URL, body=cb) self.assertRaises( exceptions.ConnectionFailed, self.http._cs_request, URL, METHOD ) - self.mox.VerifyAll() def test_request_success(self): - rv_should_be = MyResp(200), 'test content' - - self.clazz._request( - URL, METHOD, body=None, headers=mox.IgnoreArg() - ).AndReturn(rv_should_be) - self.mox.ReplayAll() + text = 'test content' + self.requests.register_uri(METHOD, URL, text=text) - self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD)) - self.mox.VerifyAll() + resp, resp_text = self.http._cs_request(URL, METHOD) + self.assertEqual(200, resp.status_code) + self.assertEqual(text, resp_text) def test_request_unauthorized(self): - rv_should_be = MyResp(401), 'unauthorized message' - self.clazz._request( - URL, METHOD, body=None, headers=mox.IgnoreArg() - ).AndReturn(rv_should_be) - self.mox.ReplayAll() - + text = 'unauthorized message' + self.requests.register_uri(METHOD, URL, status_code=401, text=text) e = self.assertRaises(exceptions.Unauthorized, self.http._cs_request, URL, METHOD) - self.assertEqual('unauthorized message', e.message) - self.mox.VerifyAll() + self.assertEqual(text, e.message) def test_request_forbidden_is_returned_to_caller(self): - rv_should_be = MyResp(403), 'forbidden message' - self.clazz._request( - URL, METHOD, body=None, headers=mox.IgnoreArg() - ).AndReturn(rv_should_be) - self.mox.ReplayAll() - - self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD)) - self.mox.VerifyAll() + text = 'forbidden message' + self.requests.register_uri(METHOD, URL, status_code=403, text=text) + + resp, resp_text = self.http._cs_request(URL, METHOD) + self.assertEqual(403, resp.status_code) + self.assertEqual(text, resp_text) -- cgit v1.2.1