openstack/python-neutronclient
Code Issues Proposed changes

Use requests_mock instead of mox

Browse Source 
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
This commit is contained in:
Jamie Lennox
2014-10-30 17:17:17 +01:00
parent ed25cba554
commit 5822d619e5
2 changed files with 117 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

276
neutronclient/tests/unit/test_auth.py
View File

@@ -15,25 +15,22 @@
# #
import json import json
import logging
import uuid import uuid
import fixtures import fixtures
from mox3 import mox
from oslo.serialization import jsonutils from oslo.serialization import jsonutils
import requests from requests_mock.contrib import fixture as mock_fixture
import requests_mock
import six
import testtools import testtools
from keystoneclient.auth.identity import v2 as ks_v2_auth from keystoneclient.auth.identity import v2 as ks_v2_auth
from keystoneclient.auth.identity import v3 as ks_v3_auth from keystoneclient.auth.identity import v3 as ks_v3_auth
from keystoneclient import exceptions as ks_exceptions from keystoneclient import exceptions as ks_exceptions
from keystoneclient import fixture from keystoneclient import fixture as ks_fixture
from keystoneclient import session from keystoneclient import session
from neutronclient import client from neutronclient import client
from neutronclient.common import exceptions from neutronclient.common import exceptions
from neutronclient.common import utils
USERNAME = 'testuser' USERNAME = 'testuser'
@@ -41,16 +38,16 @@ USER_ID = 'testuser_id'
TENANT_NAME = 'testtenant' TENANT_NAME = 'testtenant'
TENANT_ID = 'testtenant_id' TENANT_ID = 'testtenant_id'
PASSWORD = 'password' PASSWORD = 'password'
ENDPOINT_URL = 'localurl' ENDPOINT_URL = 'http://localurl'
PUBLIC_ENDPOINT_URL = 'public_%s' % ENDPOINT_URL PUBLIC_ENDPOINT_URL = '%s/public' % ENDPOINT_URL
ADMIN_ENDPOINT_URL = 'admin_%s' % ENDPOINT_URL ADMIN_ENDPOINT_URL = '%s/admin' % ENDPOINT_URL
INTERNAL_ENDPOINT_URL = 'internal_%s' % ENDPOINT_URL INTERNAL_ENDPOINT_URL = '%s/internal' % ENDPOINT_URL
ENDPOINT_OVERRIDE = 'otherurl' ENDPOINT_OVERRIDE = 'http://otherurl'
TOKENID = uuid.uuid4().hex TOKENID = uuid.uuid4().hex
REGION = 'RegionOne' REGION = 'RegionOne'
NOAUTH = 'noauth' NOAUTH = 'noauth'
KS_TOKEN_RESULT = fixture.V2Token() KS_TOKEN_RESULT = ks_fixture.V2Token()
KS_TOKEN_RESULT.set_scope() KS_TOKEN_RESULT.set_scope()
_s = KS_TOKEN_RESULT.add_service('network', 'Neutron Service') _s = KS_TOKEN_RESULT.add_service('network', 'Neutron Service')
_s.add_endpoint(ENDPOINT_URL, region=REGION) _s.add_endpoint(ENDPOINT_URL, region=REGION)
@@ -71,8 +68,8 @@ BASE_URL = "http://keystone.example.com:5000/"
V2_URL = "%sv2.0" % BASE_URL V2_URL = "%sv2.0" % BASE_URL
V3_URL = "%sv3" % BASE_URL V3_URL = "%sv3" % BASE_URL
_v2 = fixture.V2Discovery(V2_URL) _v2 = ks_fixture.V2Discovery(V2_URL)
_v3 = fixture.V3Discovery(V3_URL) _v3 = ks_fixture.V3Discovery(V3_URL)
V3_VERSION_LIST = jsonutils.dumps({'versions': {'values': [_v2, _v3]}}) V3_VERSION_LIST = jsonutils.dumps({'versions': {'values': [_v2, _v3]}})
@@ -80,15 +77,8 @@ V2_VERSION_ENTRY = {'version': _v2}
V3_VERSION_ENTRY = {'version': _v3} 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): 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 = v2_token.add_service('network')
service.add_endpoint(PUBLIC_ENDPOINT_URL, region=REGION) service.add_endpoint(PUBLIC_ENDPOINT_URL, region=REGION)
@@ -106,7 +96,7 @@ def setup_keystone_v3(mrequests):
V3_URL, V3_URL,
json=V3_VERSION_ENTRY) json=V3_VERSION_ENTRY)
v3_token = fixture.V3Token() v3_token = ks_fixture.V3Token()
service = v3_token.add_service('network') service = v3_token.add_service('network')
service.add_standard_endpoints(public=PUBLIC_ENDPOINT_URL, service.add_standard_endpoints(public=PUBLIC_ENDPOINT_URL,
admin=ADMIN_ENDPOINT_URL, admin=ADMIN_ENDPOINT_URL,
@@ -135,28 +125,21 @@ 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.requests = self.useFixture(mock_fixture.Fixture())
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(self.mox.UnsetStubs)
def test_get_noauth(self): def test_get_noauth(self):
self.mox.StubOutWithMock(self.client, "request") url = ENDPOINT_URL + '/resource'
self.requests.get(ENDPOINT_URL + '/resource')
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') self.client.do_request('/resource', 'GET')
self.assertEqual(url, self.requests.last_request.url)
self.assertEqual(self.client.endpoint_url, ENDPOINT_URL) self.assertEqual(self.client.endpoint_url, ENDPOINT_URL)
@@ -165,11 +148,13 @@ 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()
for var in ('http_proxy', 'HTTP_PROXY'): for var in ('http_proxy', 'HTTP_PROXY'):
self.useFixture(fixtures.EnvironmentVariableFixture(var)) 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( self.client = client.construct_http_client(
username=USERNAME, username=USERNAME,
tenant_name=TENANT_NAME, tenant_name=TENANT_NAME,
@@ -177,9 +162,6 @@ class CLITestAuthKeystone(testtools.TestCase):
auth_url=AUTH_URL, auth_url=AUTH_URL,
region_name=REGION) region_name=REGION)
self.addCleanup(self.mox.VerifyAll)
self.addCleanup(self.mox.UnsetStubs)
def test_reused_token_get_auth_info(self): def test_reused_token_get_auth_info(self):
"""Test that Client.get_auth_info() works even if client was """Test that Client.get_auth_info() works even if client was
instantiated with predefined token. instantiated with predefined token.
@@ -197,9 +179,8 @@ 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)
@requests_mock.Mocker() def test_get_token(self):
def test_get_token(self, mrequests): auth_session, auth_plugin = setup_keystone_v2(self.requests)
auth_session, auth_plugin = setup_keystone_v2(mrequests)
self.client = client.construct_http_client( self.client = client.construct_http_client(
username=USERNAME, username=USERNAME,
@@ -210,61 +191,50 @@ class CLITestAuthKeystone(testtools.TestCase):
session=auth_session, session=auth_session,
auth=auth_plugin) auth=auth_plugin)
self.mox.StubOutWithMock(self.client, "request") m = self.requests.get(PUBLIC_ENDPOINT_URL + '/resource',
res200 = get_response(200) request_headers={'X-Auth-Token': TOKENID})
self.client.request(
'/resource', 'GET',
authenticated=True
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET') self.client.do_request('/resource', 'GET')
self.assertTrue(m.called)
def test_refresh_token(self): def test_refresh_token(self):
self.mox.StubOutWithMock(self.client, "request")
token_id = uuid.uuid4().hex token_id = uuid.uuid4().hex
text = uuid.uuid4().hex
self.client.auth_token = token_id self.client.auth_token = token_id
self.client.endpoint_url = ENDPOINT_URL self.client.endpoint_url = ENDPOINT_URL
res200 = get_response(200) res_url = ENDPOINT_URL + '/resource'
res401 = get_response(401) v2_url = AUTH_URL + '/tokens'
# If a token is expired, neutron server retruns 401 # token_id gives 401, KS_TOKEN_RESULT gives 200
self.client.request( self.requests.get(res_url,
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET', request_headers={'X-Auth-Token': token_id},
headers=mox.ContainsKeyValue('X-Auth-Token', token_id) status_code=401)
).AndReturn((res401, ''))
self.client.request( self.requests.get(
AUTH_URL + '/tokens', 'POST', res_url,
body=mox.IsA(str), headers=mox.IsA(dict) text=text,
).AndReturn((res200, json.dumps(KS_TOKEN_RESULT))) status_code=200,
self.client.request( request_headers={'X-Auth-Token': KS_TOKEN_RESULT.token_id})
mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', self.requests.post(v2_url, json=KS_TOKEN_RESULT)
KS_TOKEN_RESULT.token_id)
).AndReturn((res200, '')) resp = self.client.do_request('/resource', 'GET')
self.mox.ReplayAll()
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): def test_refresh_token_no_auth_url(self):
self.mox.StubOutWithMock(self.client, "request")
self.client.auth_url = None self.client.auth_url = None
token_id = uuid.uuid4().hex token_id = uuid.uuid4().hex
self.client.auth_token = token_id self.client.auth_token = token_id
self.client.endpoint_url = ENDPOINT_URL self.client.endpoint_url = ENDPOINT_URL
res401 = get_response(401) self.requests.get(ENDPOINT_URL + '/resource', status_code=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.assertRaises(exceptions.NoAuthURLProvided, self.assertRaises(exceptions.NoAuthURLProvided,
self.client.do_request, self.client.do_request,
'/resource', '/resource',
@@ -277,24 +247,18 @@ class CLITestAuthKeystone(testtools.TestCase):
self.client._get_endpoint_url) self.client._get_endpoint_url)
def test_get_endpoint_url(self): def test_get_endpoint_url(self):
self.mox.StubOutWithMock(self.client, "request")
token_id = uuid.uuid4().hex token_id = uuid.uuid4().hex
self.client.auth_token = token_id self.client.auth_token = token_id
res200 = get_response(200) self.requests.get(AUTH_URL + '/tokens/%s/endpoints' % token_id,
json=ENDPOINTS_RESULT)
self.requests.get(ENDPOINT_URL + '/resource')
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.client.do_request('/resource', 'GET') 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): def test_use_given_endpoint_url(self):
self.client = client.HTTPClient( self.client = client.HTTPClient(
username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD, username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
@@ -302,69 +266,49 @@ class CLITestAuthKeystone(testtools.TestCase):
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")
token_id = uuid.uuid4().hex token_id = uuid.uuid4().hex
self.client.auth_token = token_id self.client.auth_token = token_id
res200 = get_response(200)
self.client.request( self.requests.get(ENDPOINT_OVERRIDE + '/resource')
mox.StrContains(ENDPOINT_OVERRIDE + '/resource'), 'GET',
headers=mox.ContainsKeyValue('X-Auth-Token', token_id)
).AndReturn((res200, ''))
self.mox.ReplayAll()
self.client.do_request('/resource', 'GET') self.client.do_request('/resource', 'GET')
self.assertEqual(self.client.endpoint_url, ENDPOINT_OVERRIDE) 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): def test_get_endpoint_url_other(self):
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")
token_id = uuid.uuid4().hex token_id = uuid.uuid4().hex
self.client.auth_token = token_id self.client.auth_token = token_id
res200 = get_response(200)
self.client.request( self.requests.get(AUTH_URL + '/tokens/%s/endpoints' % token_id,
mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % token_id), json=ENDPOINTS_RESULT)
'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')
def test_get_endpoint_url_failed(self): def test_get_endpoint_url_failed(self):
self.mox.StubOutWithMock(self.client, "request")
token_id = uuid.uuid4().hex token_id = uuid.uuid4().hex
self.client.auth_token = token_id self.client.auth_token = token_id
res200 = get_response(200) self.requests.get(AUTH_URL + '/tokens/%s/endpoints' % token_id,
res401 = get_response(401) status_code=401)
self.requests.post(AUTH_URL + '/tokens', json=KS_TOKEN_RESULT)
m = self.requests.get(ENDPOINT_URL + '/resource')
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.client.do_request('/resource', 'GET') self.client.do_request('/resource', 'GET')
@requests_mock.Mocker() self.assertEqual(KS_TOKEN_RESULT.token_id,
def test_endpoint_type(self, mrequests): m.last_request.headers['X-Auth-Token'])
auth_session, auth_plugin = setup_keystone_v3(mrequests)
def test_endpoint_type(self):
auth_session, auth_plugin = setup_keystone_v3(self.requests)
# Test default behavior is to choose public. # Test default behavior is to choose public.
self.client = client.construct_http_client( self.client = client.construct_http_client(
@@ -413,34 +357,16 @@ class CLITestAuthKeystone(testtools.TestCase):
self.client.authenticate) self.client.authenticate)
def test_strip_credentials_from_log(self): def test_strip_credentials_from_log(self):
def verify_no_credentials(kwargs): m = self.requests.post(AUTH_URL + '/tokens', json=KS_TOKEN_RESULT)
return ('REDACTED' in kwargs['body']) and ( self.requests.get(ENDPOINT_URL + '/resource')
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()
self.client.do_request('/resource', 'GET') 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): class CLITestAuthKeystoneWithId(CLITestAuthKeystone):
@@ -473,14 +399,10 @@ class TestKeystoneClientVersions(testtools.TestCase):
def setUp(self): def setUp(self):
"""Prepare the test environment.""" """Prepare the test environment."""
super(TestKeystoneClientVersions, self).setUp() super(TestKeystoneClientVersions, self).setUp()
self.mox = mox.Mox() self.requests = self.useFixture(mock_fixture.Fixture())
self.addCleanup(self.mox.VerifyAll)
self.addCleanup(self.mox.UnsetStubs)
@requests_mock.Mocker() def test_v2_auth(self):
def test_v2_auth(self, mrequests): auth_session, auth_plugin = setup_keystone_v2(self.requests)
auth_session, auth_plugin = setup_keystone_v2(mrequests)
res200 = get_response(200)
self.client = client.construct_http_client( self.client = client.construct_http_client(
username=USERNAME, username=USERNAME,
@@ -491,20 +413,14 @@ class TestKeystoneClientVersions(testtools.TestCase):
session=auth_session, session=auth_session,
auth=auth_plugin) 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.client.do_request('/resource', 'GET')
@requests_mock.Mocker() self.assertTrue(m.called)
def test_v3_auth(self, mrequests):
auth_session, auth_plugin = setup_keystone_v3(mrequests) def test_v3_auth(self):
res200 = get_response(200) auth_session, auth_plugin = setup_keystone_v3(self.requests)
self.client = client.construct_http_client( self.client = client.construct_http_client(
user_id=USER_ID, user_id=USER_ID,
@@ -515,12 +431,8 @@ class TestKeystoneClientVersions(testtools.TestCase):
session=auth_session, session=auth_session,
auth=auth_plugin) 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.client.do_request('/resource', 'GET')
self.assertTrue(m.called)

65
neutronclient/tests/unit/test_http.py
View File

@@ -15,15 +15,13 @@
import abc import abc
from mox3 import mox from requests_mock.contrib import fixture as mock_fixture
import requests_mock
import six import six
import testtools import testtools
from neutronclient import client from neutronclient import client
from neutronclient.common import exceptions from neutronclient.common import exceptions
from neutronclient.tests.unit import test_auth from neutronclient.tests.unit import test_auth
from neutronclient.tests.unit.test_cli20 import MyResp
AUTH_TOKEN = 'test_token' AUTH_TOKEN = 'test_token'
@@ -39,10 +37,8 @@ class TestHTTPClientMixin(object):
def setUp(self): def setUp(self):
super(TestHTTPClientMixin, self).setUp() super(TestHTTPClientMixin, self).setUp()
self.requests = self.useFixture(mock_fixture.Fixture())
self.clazz, self.http = self.initialize() self.clazz, self.http = self.initialize()
self.mox = mox.Mox()
self.addCleanup(self.mox.UnsetStubs)
self.mox.StubOutWithMock(self.clazz, '_request')
@abc.abstractmethod @abc.abstractmethod
def initialize(self): def initialize(self):
@@ -50,12 +46,10 @@ class TestHTTPClientMixin(object):
def _test_headers(self, expected_headers, **kwargs): def _test_headers(self, expected_headers, **kwargs):
"""Test headers.""" """Test headers."""
self.clazz._request(URL, METHOD, self.requests.register_uri(METHOD, URL,
body=kwargs.get('body'), request_headers=expected_headers)
headers=expected_headers)
self.mox.ReplayAll()
self.http.request(URL, METHOD, **kwargs) 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): def test_headers_without_body(self):
self._test_headers({'Accept': 'application/json'}) self._test_headers({'Accept': 'application/json'})
@@ -82,9 +76,8 @@ class TestHTTPClientMixin(object):
class TestSessionClient(TestHTTPClientMixin, testtools.TestCase): class TestSessionClient(TestHTTPClientMixin, testtools.TestCase):
@requests_mock.Mocker() def initialize(self):
def initialize(self, mrequests): session, auth = test_auth.setup_keystone_v2(self.requests)
session, auth = test_auth.setup_keystone_v2(mrequests)
return [client.SessionClient, return [client.SessionClient,
client.SessionClient(session=session, auth=auth)] client.SessionClient(session=session, auth=auth)]
@@ -96,47 +89,35 @@ class TestHTTPClient(TestHTTPClientMixin, testtools.TestCase):
client.HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL)] client.HTTPClient(token=AUTH_TOKEN, endpoint_url=END_URL)]
def test_request_error(self): def test_request_error(self):
self.clazz._request( def cb(*args, **kwargs):
URL, METHOD, body=None, headers=mox.IgnoreArg() raise Exception('error msg')
).AndRaise(Exception('error msg'))
self.mox.ReplayAll()
self.requests.get(URL, body=cb)
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): def test_request_success(self):
rv_should_be = MyResp(200), 'test content' text = 'test content'
self.requests.register_uri(METHOD, URL, text=text)
self.clazz._request( resp, resp_text = self.http._cs_request(URL, METHOD)
URL, METHOD, body=None, headers=mox.IgnoreArg() self.assertEqual(200, resp.status_code)
).AndReturn(rv_should_be) self.assertEqual(text, resp_text)
self.mox.ReplayAll()
self.assertEqual(rv_should_be, self.http._cs_request(URL, METHOD))
self.mox.VerifyAll()
def test_request_unauthorized(self): def test_request_unauthorized(self):
rv_should_be = MyResp(401), 'unauthorized message' text = 'unauthorized message'
self.clazz._request( self.requests.register_uri(METHOD, URL, status_code=401, text=text)
URL, METHOD, body=None, headers=mox.IgnoreArg()
).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', e.message) self.assertEqual(text, e.message)
self.mox.VerifyAll()
def test_request_forbidden_is_returned_to_caller(self): def test_request_forbidden_is_returned_to_caller(self):
rv_should_be = MyResp(403), 'forbidden message' text = 'forbidden message'
self.clazz._request( self.requests.register_uri(METHOD, URL, status_code=403, text=text)
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)) resp, resp_text = self.http._cs_request(URL, METHOD)
self.mox.VerifyAll() self.assertEqual(403, resp.status_code)
self.assertEqual(text, resp_text)