diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/unit/apiclient/__init__.py | 0 | ||||
| -rw-r--r-- | tests/unit/apiclient/test_auth.py | 178 | ||||
| -rw-r--r-- | tests/unit/apiclient/test_base.py | 348 | ||||
| -rw-r--r-- | tests/unit/apiclient/test_client.py | 335 | ||||
| -rw-r--r-- | tests/unit/apiclient/test_exceptions.py | 137 | ||||
| -rw-r--r-- | tests/unit/apiclient/test_utils.py | 115 |
6 files changed, 0 insertions, 1113 deletions
diff --git a/tests/unit/apiclient/__init__.py b/tests/unit/apiclient/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/tests/unit/apiclient/__init__.py +++ /dev/null diff --git a/tests/unit/apiclient/test_auth.py b/tests/unit/apiclient/test_auth.py deleted file mode 100644 index a9806d8b..00000000 --- a/tests/unit/apiclient/test_auth.py +++ /dev/null @@ -1,178 +0,0 @@ -# Copyright 2012 OpenStack Foundation -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. - -import argparse - -import fixtures -import mock -from oslotest import base as test_base -import requests -from stevedore import extension - -try: - import json -except ImportError: - import simplejson as json - -from openstack.common.apiclient import auth -from openstack.common.apiclient import client -from openstack.common.apiclient import fake_client - - -TEST_REQUEST_BASE = { - 'verify': True, -} - - -def mock_http_request(resp=None): - """Mock an HTTP Request.""" - if not resp: - resp = { - "access": { - "token": { - "expires": "12345", - "id": "FAKE_ID", - "tenant": { - "id": "FAKE_TENANT_ID", - } - }, - "serviceCatalog": [ - { - "type": "compute", - "endpoints": [ - { - "region": "RegionOne", - "adminURL": "http://localhost:8774/v1.1", - "internalURL": "http://localhost:8774/v1.1", - "publicURL": "http://localhost:8774/v1.1/", - }, - ], - }, - ], - }, - } - - auth_response = fake_client.TestResponse({ - "status_code": 200, - "text": json.dumps(resp), - }) - return mock.Mock(return_value=(auth_response)) - - -def requested_headers(cs): - """Return requested passed headers.""" - return { - 'User-Agent': cs.user_agent, - 'Content-Type': 'application/json', - } - - -class BaseFakePlugin(auth.BaseAuthPlugin): - def _do_authenticate(self, http_client): - pass - - def token_and_endpoint(self, endpoint_type, service_type): - pass - - -class GlobalFunctionsTest(test_base.BaseTestCase): - - def test_load_auth_system_opts(self): - self.useFixture(fixtures.MonkeyPatch( - "os.environ", - {"OS_TENANT_NAME": "fake-project", - "OS_USERNAME": "fake-username"})) - parser = argparse.ArgumentParser() - auth.load_auth_system_opts(parser) - options = parser.parse_args( - ["--os-auth-url=fake-url", "--os_auth_system=fake-system"]) - self.assertEqual(options.os_tenant_name, "fake-project") - self.assertEqual(options.os_username, "fake-username") - self.assertEqual(options.os_auth_url, "fake-url") - self.assertEqual(options.os_auth_system, "fake-system") - - -class MockEntrypoint(object): - def __init__(self, name, plugin): - self.name = name - self.plugin = plugin - - -class AuthPluginTest(test_base.BaseTestCase): - @mock.patch.object(requests.Session, "request") - @mock.patch.object(extension.ExtensionManager, "map") - def test_auth_system_success(self, mock_mgr_map, mock_request): - """Test that we can authenticate using the auth system.""" - class FakePlugin(BaseFakePlugin): - def authenticate(self, cls): - cls.request( - "POST", "http://auth/tokens", - json={"fake": "me"}, allow_redirects=True) - - mock_mgr_map.side_effect = ( - lambda func: func(MockEntrypoint("fake", FakePlugin))) - - mock_request.side_effect = mock_http_request() - - auth.discover_auth_systems() - plugin = auth.load_plugin("fake") - cs = client.HTTPClient(auth_plugin=plugin) - cs.authenticate() - - headers = requested_headers(cs) - - mock_request.assert_called_with( - "POST", - "http://auth/tokens", - headers=headers, - data='{"fake": "me"}', - allow_redirects=True, - **TEST_REQUEST_BASE) - - @mock.patch.object(extension.ExtensionManager, "map") - def test_discover_auth_system_options(self, mock_mgr_map): - """Test that we can load the auth system options.""" - class FakePlugin(BaseFakePlugin): - @classmethod - def add_opts(cls, parser): - parser.add_argument('--auth_system_opt', - default=False, - action='store_true', - help="Fake option.") - - mock_mgr_map.side_effect = ( - lambda func: func(MockEntrypoint("fake", FakePlugin))) - - parser = argparse.ArgumentParser() - auth.discover_auth_systems() - auth.load_auth_system_opts(parser) - opts, _args = parser.parse_known_args(['--auth_system_opt']) - - self.assertTrue(opts.auth_system_opt) - - @mock.patch.object(extension.ExtensionManager, "map") - def test_parse_auth_system_options(self, mock_mgr_map): - """Test that we can parse the auth system options.""" - class FakePlugin(BaseFakePlugin): - opt_names = ["fake_argument"] - - mock_mgr_map.side_effect = ( - lambda func: func(MockEntrypoint("fake", FakePlugin))) - - auth.discover_auth_systems() - plugin = auth.load_plugin("fake") - - plugin.parse_opts([]) - self.assertIn("fake_argument", plugin.opts) diff --git a/tests/unit/apiclient/test_base.py b/tests/unit/apiclient/test_base.py deleted file mode 100644 index f7bff083..00000000 --- a/tests/unit/apiclient/test_base.py +++ /dev/null @@ -1,348 +0,0 @@ -# Copyright 2013 OpenStack Foundation -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. - -import mock -from oslotest import base as test_base -from testtools import matchers - -from openstack.common.apiclient import base -from openstack.common.apiclient import client -from openstack.common.apiclient import exceptions -from openstack.common.apiclient import fake_client - - -class HumanResource(base.Resource): - HUMAN_ID = True - - -class HumanResourceManager(base.ManagerWithFind): - resource_class = HumanResource - - def list(self): - return self._list("/human_resources", "human_resources") - - def get(self, human_resource): - return self._get( - "/human_resources/%s" % base.getid(human_resource), - "human_resource") - - def update(self, human_resource, name): - body = { - "human_resource": { - "name": name, - }, - } - return self._put( - "/human_resources/%s" % base.getid(human_resource), - body, - "human_resource") - - -class CrudResource(base.Resource): - pass - - -class CrudResourceManager(base.CrudManager): - """Manager class for manipulating Identity crud_resources.""" - resource_class = CrudResource - collection_key = 'crud_resources' - key = 'crud_resource' - - def get(self, crud_resource): - return super(CrudResourceManager, self).get( - crud_resource_id=base.getid(crud_resource)) - - -class FakeHTTPClient(fake_client.FakeHTTPClient): - crud_resource_json = {"id": "1", "domain_id": "my-domain"} - - def get_human_resources(self, **kw): - return (200, {}, - {'human_resources': [ - {'id': 1, 'name': '256 MB Server'}, - {'id': 2, 'name': '512 MB Server'}, - {'id': 'aa1', 'name': '128 MB Server'}, - {'id': 3, 'name': '4 MB Server'}, - ]}) - - def get_human_resources_1(self, **kw): - res = self.get_human_resources()[2]['human_resources'][0] - return (200, {}, {'human_resource': res}) - - def get_human_resources_3(self, **kw): - res = self.get_human_resources()[2]['human_resources'][3] - return (200, {'x-openstack-request-id': 'req-flappy'}, - {'human_resource': res}) - - def put_human_resources_1(self, **kw): - kw = kw["json"]["human_resource"].copy() - kw["id"] = "1" - return (200, {}, {'human_resource': kw}) - - def post_crud_resources(self, **kw): - return (200, {}, {"crud_resource": {"id": "1"}}) - - def get_crud_resources(self, **kw): - crud_resources = [] - if kw.get("domain_id") == self.crud_resource_json["domain_id"]: - crud_resources = [self.crud_resource_json] - else: - crud_resources = [] - return (200, {}, {"crud_resources": crud_resources}) - - def get_crud_resources_1(self, **kw): - return (200, {}, {"crud_resource": self.crud_resource_json}) - - def head_crud_resources_1(self, **kw): - return (204, {}, None) - - def patch_crud_resources_1(self, **kw): - self.crud_resource_json.update(kw) - return (200, {}, {"crud_resource": self.crud_resource_json}) - - def delete_crud_resources_1(self, **kw): - return (202, {}, None) - - -class TestClient(client.BaseClient): - - service_type = "test" - - def __init__(self, http_client, extensions=None): - super(TestClient, self).__init__( - http_client, extensions=extensions) - - self.human_resources = HumanResourceManager(self) - self.crud_resources = CrudResourceManager(self) - - -class ResourceTest(test_base.BaseTestCase): - def test_resource_repr(self): - r = base.Resource(None, dict(foo="bar", baz="spam")) - self.assertEqual(repr(r), "<Resource baz=spam, foo=bar>") - - def test_getid(self): - class TmpObject(base.Resource): - id = "4" - self.assertEqual(base.getid(TmpObject(None, {})), "4") - - def test_human_id(self): - r = base.Resource(None, {"name": "1"}) - self.assertIsNone(r.human_id) - r = HumanResource(None, {"name": "1"}) - self.assertEqual(r.human_id, "1") - r = HumanResource(None, {"name": None}) - self.assertIsNone(r.human_id) - - -class BaseManagerTest(test_base.BaseTestCase): - - def setUp(self): - super(BaseManagerTest, self).setUp() - self.http_client = FakeHTTPClient() - self.tc = TestClient(self.http_client) - - def test_resource_lazy_getattr(self): - f = HumanResource(self.tc.human_resources, {'id': 1}) - self.assertEqual(f.name, '256 MB Server') - self.http_client.assert_called('GET', '/human_resources/1') - - # Missing stuff still fails after a second get - self.assertRaises(AttributeError, getattr, f, 'blahblah') - - def test_resource_req_id(self): - f = HumanResource(self.tc.human_resources, {'id': 1}) - self.assertEqual(f.name, '256 MB Server') - self.assertFalse(hasattr(f, 'x_request_id')) - self.http_client.assert_called('GET', '/human_resources/1') - - def test_resource_req_id_header(self): - f = HumanResource(self.tc.human_resources, {'id': 3}) - self.assertEqual(f.name, '4 MB Server') - self.assertEqual(f.x_request_id, 'req-flappy') - self.http_client.assert_called('GET', '/human_resources/3') - - def test_eq(self): - # Two resources of the same type with the same id: equal - r1 = base.Resource(None, {'id': 1, 'name': 'hi'}) - r2 = base.Resource(None, {'id': 1, 'name': 'hello'}) - self.assertEqual(r1, r2) - - # Two resources of different types: never equal - r1 = base.Resource(None, {'id': 1}) - r2 = HumanResource(None, {'id': 1}) - self.assertNotEqual(r1, r2) - - # Two resources with no ID: equal if their info is equal - r1 = base.Resource(None, {'name': 'joe', 'age': 12}) - r2 = base.Resource(None, {'name': 'joe', 'age': 12}) - self.assertEqual(r1, r2) - - def test_findall_invalid_attribute(self): - # Make sure findall with an invalid attribute doesn't cause errors. - # The following should not raise an exception. - self.tc.human_resources.findall(vegetable='carrot') - - # However, find() should raise an error - self.assertRaises(exceptions.NotFound, - self.tc.human_resources.find, - vegetable='carrot') - - def test_update(self): - name = "new-name" - human_resource = self.tc.human_resources.update("1", name) - self.assertEqual(human_resource.id, "1") - self.assertEqual(human_resource.name, name) - - -class BaseManagerTestCase(test_base.BaseTestCase): - - def setUp(self): - super(BaseManagerTestCase, self).setUp() - - self.response = mock.MagicMock() - self.http_client = mock.MagicMock() - self.http_client.get.return_value = self.response - self.http_client.post.return_value = self.response - - self.manager = base.BaseManager(self.http_client) - self.manager.resource_class = HumanResource - - def test_list(self): - self.response.json.return_value = {'human_resources': [{'id': 42}]} - expected = [HumanResource(self.manager, {'id': 42}, loaded=True)] - result = self.manager._list("/human_resources", "human_resources") - self.assertEqual(expected, result) - - def test_list_no_response_key(self): - self.response.json.return_value = [{'id': 42}] - expected = [HumanResource(self.manager, {'id': 42}, loaded=True)] - result = self.manager._list("/human_resources") - self.assertEqual(expected, result) - - def test_list_get(self): - self.manager._list("/human_resources", "human_resources") - self.manager.client.get.assert_called_with("/human_resources") - - def test_list_post(self): - self.manager._list("/human_resources", "human_resources", - json={'id': 42}) - self.manager.client.post.assert_called_with("/human_resources", - json={'id': 42}) - - def test_get(self): - self.response.json.return_value = {'human_resources': {'id': 42}} - expected = HumanResource(self.manager, {'id': 42}, loaded=True) - result = self.manager._get("/human_resources/42", "human_resources") - self.manager.client.get.assert_called_with("/human_resources/42") - self.assertEqual(expected, result) - - def test_get_no_response_key(self): - self.response.json.return_value = {'id': 42} - expected = HumanResource(self.manager, {'id': 42}, loaded=True) - result = self.manager._get("/human_resources/42") - self.manager.client.get.assert_called_with("/human_resources/42") - self.assertEqual(expected, result) - - def test_post(self): - self.response.json.return_value = {'human_resources': {'id': 42}} - expected = HumanResource(self.manager, {'id': 42}, loaded=True) - result = self.manager._post("/human_resources", - response_key="human_resources", - json={'id': 42}) - self.manager.client.post.assert_called_with("/human_resources", - json={'id': 42}) - self.assertEqual(expected, result) - - def test_post_return_raw(self): - self.response.json.return_value = {'human_resources': {'id': 42}} - result = self.manager._post("/human_resources", - response_key="human_resources", - json={'id': 42}, return_raw=True) - self.manager.client.post.assert_called_with("/human_resources", - json={'id': 42}) - self.assertEqual(result, {'id': 42}) - - def test_post_no_response_key(self): - self.response.json.return_value = {'id': 42} - expected = HumanResource(self.manager, {'id': 42}, loaded=True) - result = self.manager._post("/human_resources", json={'id': 42}) - self.manager.client.post.assert_called_with("/human_resources", - json={'id': 42}) - self.assertEqual(expected, result) - - -class CrudManagerTest(test_base.BaseTestCase): - - domain_id = "my-domain" - crud_resource_id = "1" - - def setUp(self): - super(CrudManagerTest, self).setUp() - self.http_client = FakeHTTPClient() - self.tc = TestClient(self.http_client) - - def test_create(self): - crud_resource = self.tc.crud_resources.create() - self.assertEqual(crud_resource.id, self.crud_resource_id) - - def test_list(self, domain=None, user=None): - crud_resources = self.tc.crud_resources.list( - base_url=None, - domain_id=self.domain_id) - self.assertEqual(len(crud_resources), 1) - self.assertEqual(crud_resources[0].id, self.crud_resource_id) - self.assertEqual(crud_resources[0].domain_id, self.domain_id) - crud_resources = self.tc.crud_resources.list( - base_url=None, - domain_id="another-domain", - another_attr=None) - self.assertEqual(len(crud_resources), 0) - - def test_get(self): - crud_resource = self.tc.crud_resources.get(self.crud_resource_id) - self.assertEqual(crud_resource.id, self.crud_resource_id) - fake_client.assert_has_keys( - crud_resource._info, - required=["id", "domain_id"], - optional=["missing-attr"]) - - def test_update(self): - crud_resource = self.tc.crud_resources.update( - crud_resource_id=self.crud_resource_id, - domain_id=self.domain_id) - self.assertEqual(crud_resource.id, self.crud_resource_id) - self.assertEqual(crud_resource.domain_id, self.domain_id) - - def test_delete(self): - resp = self.tc.crud_resources.delete( - crud_resource_id=self.crud_resource_id) - self.assertEqual(resp.status_code, 202) - - def test_head(self): - ret = self.tc.crud_resources.head( - crud_resource_id=self.crud_resource_id) - self.assertTrue(ret) - - def test_find(self): - ret = self.tc.crud_resources.find(domain_id=self.domain_id) - self.assertTrue(ret) - - def test_find_receives_a_NotFound_error(self): - e = self.assertRaises(exceptions.NotFound, - self.tc.crud_resources.find, - resource_id=self.crud_resource_id) - self.assertThat(str(e), - matchers.StartsWith('No CrudResource matching ')) diff --git a/tests/unit/apiclient/test_client.py b/tests/unit/apiclient/test_client.py deleted file mode 100644 index 1911c4e7..00000000 --- a/tests/unit/apiclient/test_client.py +++ /dev/null @@ -1,335 +0,0 @@ -# -*- coding: utf-8 -*- -# -# Copyright 2012 OpenStack Foundation -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. - -import logging - -import fixtures -import mock -from oslotest import base as test_base -import requests - -from openstack.common.apiclient import auth -from openstack.common.apiclient import client -from openstack.common.apiclient import exceptions -from openstack.common.apiclient import fake_client - - -class TestClient(client.BaseClient): - service_type = "test" - cached_endpoint = 'old_endpoint' - - -class FakeAuthPlugin(auth.BaseAuthPlugin): - auth_system = "fake" - attempt = -1 - - def _do_authenticate(self, http_client): - pass - - def token_and_endpoint(self, endpoint_type, service_type): - self.attempt = self.attempt + 1 - return ("token-%s" % self.attempt, "/endpoint-%s" % self.attempt) - - -class ClientTest(test_base.BaseTestCase): - - def test_client_with_timeout(self): - http_client = client.HTTPClient(None, timeout=2) - self.assertEqual(http_client.timeout, 2) - mock_request = mock.Mock() - mock_request.return_value = requests.Response() - mock_request.return_value.status_code = 200 - with mock.patch("requests.Session.request", mock_request): - http_client.request("GET", "/", json={"1": "2"}) - requests.Session.request.assert_called_with( - "GET", - "/", - timeout=2, - headers=mock.ANY, - verify=mock.ANY, - data=mock.ANY) - - def test_concat_url(self): - self.assertEqual(client.HTTPClient.concat_url("/a", "/b"), "/a/b") - self.assertEqual(client.HTTPClient.concat_url("/a", "b"), "/a/b") - self.assertEqual(client.HTTPClient.concat_url("/a/", "/b"), "/a/b") - - def test_client_request(self): - http_client = client.HTTPClient(FakeAuthPlugin()) - mock_request = mock.Mock() - mock_request.return_value = requests.Response() - mock_request.return_value.status_code = 200 - with mock.patch("requests.Session.request", mock_request): - http_client.client_request( - TestClient(http_client), "GET", "/resource", json={"1": "2"}) - requests.Session.request.assert_called_with( - "GET", - "/endpoint-0/resource", - headers={ - "User-Agent": http_client.user_agent, - "Content-Type": "application/json", - "X-Auth-Token": "token-0" - }, - data='{"1": "2"}', - verify=True) - - def test_client_request_expired_token(self): - side_effect_rv = [False, True] - - def side_effect(*args, **kwargs): - if side_effect_rv.pop(): - raise exceptions.Unauthorized() - return - - def new_creds(*args, **kwargs): - return 'new_token', 'new_endpoint' - - with mock.patch('%s.FakeAuthPlugin.token_and_endpoint' % __name__, - mock.MagicMock()) as mocked_token_and_endpoint: - mocked_token_and_endpoint.side_effect = new_creds - - http_client = client.HTTPClient(FakeAuthPlugin()) - http_client.cached_token = 'old_token' - http_client.request = mock.MagicMock(side_effect=side_effect) - - http_client.client_request( - TestClient(http_client), "GET", "/resource", json={"1": "2"}) - http_client.request.assert_called_with( - 'GET', - 'new_endpoint/resource', - headers={'X-Auth-Token': 'new_token'}, - json={'1': '2'}) - self.assertEqual('new_token', http_client.cached_token) - - def test_client_request_reissue(self): - reject_token = None - - def fake_request(method, url, **kwargs): - if kwargs["headers"]["X-Auth-Token"] == reject_token: - raise exceptions.Unauthorized(method=method, url=url) - return "%s %s" % (method, url) - - http_client = client.HTTPClient(FakeAuthPlugin()) - test_client = TestClient(http_client) - http_client.request = fake_request - - self.assertEqual( - http_client.client_request( - test_client, "GET", "/resource"), - "GET /endpoint-0/resource") - reject_token = "token-0" - self.assertEqual( - http_client.client_request( - test_client, "GET", "/resource"), - "GET /endpoint-1/resource") - - def test_client_with_response_400_status_code(self): - http_client = client.HTTPClient(FakeAuthPlugin()) - mock_request = mock.Mock() - mock_request.return_value = requests.Response() - mock_request.return_value.status_code = 400 - with mock.patch("requests.Session.request", mock_request): - self.assertRaises( - exceptions.BadRequest, http_client.client_request, - TestClient(http_client), "GET", "/resource", - json={"bad": "request"}) - - def test_client_with_no_token_and_no_endpoint(self): - with mock.patch('%s.FakeAuthPlugin.token_and_endpoint' % __name__, - mock.MagicMock()) as mocked_token_and_endpoint: - mocked_token_and_endpoint.return_value = (None, None) - - http_client = client.HTTPClient(FakeAuthPlugin()) - self.assertRaises( - exceptions.AuthorizationFailure, http_client.client_request, - TestClient(http_client), "GET", "/resource", json={"1": "2"}) - - def test_client_raising_unauthorized(self): - side_effect_rv = [True, False] - - def side_effect(*args, **kwargs): - if side_effect_rv.pop(): - raise exceptions.EndpointException() - return ("token-%s" % len(side_effect_rv), - "/endpoint-%s" % len(side_effect_rv)) - - with mock.patch('%s.FakeAuthPlugin.token_and_endpoint' % __name__, - mock.MagicMock()) as mocked_token_and_endpoint: - mocked_token_and_endpoint.side_effect = side_effect - - http_client = client.HTTPClient(FakeAuthPlugin()) - http_client.request = mock.MagicMock( - side_effect=exceptions.Unauthorized()) - self.assertRaises( - exceptions.Unauthorized, http_client.client_request, - TestClient(http_client), "GET", "/resource", json={"1": "2"}) - - def test_client_raising_unauthorized_with_equal_token_and_endpoint(self): - with mock.patch('%s.FakeAuthPlugin.token_and_endpoint' % __name__, - mock.MagicMock()) as mocked_token_and_endpoint: - mocked_token_and_endpoint.return_value = ('token-0', '/endpoint-0') - http_client = client.HTTPClient(FakeAuthPlugin()) - http_client.request = mock.MagicMock( - side_effect=exceptions.Unauthorized()) - self.assertRaises( - exceptions.Unauthorized, http_client.client_request, - TestClient(http_client), "GET", "/resource", json={"1": "2"}) - - def test_client_raising_unauthorized_with_just_authenticated(self): - side_effect_rv = [True, False, True] - - def side_effect(*args, **kwargs): - if side_effect_rv.pop(): - raise exceptions.EndpointException() - return ("token-%s" % len(side_effect_rv), - "/endpoint-%s" % len(side_effect_rv)) - - with mock.patch('%s.FakeAuthPlugin.token_and_endpoint' % __name__, - mock.MagicMock()) as mocked_token_and_endpoint: - mocked_token_and_endpoint.side_effect = side_effect - - http_client = client.HTTPClient(FakeAuthPlugin()) - http_client.request = mock.MagicMock( - side_effect=exceptions.Unauthorized()) - self.assertRaises( - exceptions.Unauthorized, http_client.client_request, - TestClient(http_client), "GET", "/resource", json={"1": "2"}) - - def test_log_req(self): - self.logger = self.useFixture( - fixtures.FakeLogger( - format="%(message)s", - level=logging.DEBUG, - nuke_handlers=True - ) - ) - cs = client.HTTPClient(FakeAuthPlugin()) - cs.debug = True - cs._http_log_req('GET', '/foo', {'headers': {}}) - BOGUS_HEADERS_1 = {'headers': {'X-Auth-Token': 'totally_bogus'}} - cs._http_log_req('GET', '/foo', BOGUS_HEADERS_1) - BOGUS_HEADERS_2 = {'headers': { - 'X-Subject-Token': 'totally_bogus_subject_token'} - } - cs._http_log_req('GET', '/foo/bar', BOGUS_HEADERS_2) - BOGUS_HEADERS_3 = {'headers': { - 'X-Foo': 'bar', - 'X-Auth-Token': 'totally_bogus', - 'X-Subject-Token': 'totally_bogus_subject_token' - } - } - cs._http_log_req('GET', '/foo', BOGUS_HEADERS_3) - - output = self.logger.output.split('\n') - - self.assertIn("REQ: curl -g -i -X 'GET' '/foo'", output) - self.assertIn("REQ: curl -g -i -X 'GET' '/foo' -H 'X-Auth-Token: " - "{SHA1}b42162b6ffdbd7c3c37b7c95b7ba9f51dda0236d'", - output) - self.assertIn("REQ: curl -g -i -X 'GET' '/foo/bar'" - " -H 'X-Subject-Token: " - "{SHA1}1999873bee4b1aed81b7abb168044a8cbbbccbd6'", - output) - self.assertIn("REQ: curl -g -i -X 'GET' '/foo' " - "-H 'X-Subject-Token: " - "{SHA1}1999873bee4b1aed81b7abb168044a8cbbbccbd6' " - "-H 'X-Foo: bar' " - "-H 'X-Auth-Token: " - "{SHA1}b42162b6ffdbd7c3c37b7c95b7ba9f51dda0236d'", - output) - - def _test_log_req_insecure(self, verify, expected): - self.logger = self.useFixture( - fixtures.FakeLogger( - format="%(message)s", - level=logging.DEBUG, - nuke_handlers=True - ) - ) - - cs = client.HTTPClient(FakeAuthPlugin(), debug=True, verify=verify) - cs._http_log_req('GET', '/foo', {'headers': {}}) - output = self.logger.output.split('\n') - self.assertIn(expected, output) - - def test_log_req_insecure(self): - expected = "REQ: curl -g -i -X 'GET' '/foo'" - self._test_log_req_insecure(True, expected) - expected = "REQ: curl -g -i --insecure -X 'GET' '/foo'" - self._test_log_req_insecure(False, expected) - - -class FakeClientTest(test_base.BaseTestCase): - - def setUp(self): - super(FakeClientTest, self).setUp() - fixtures = { - '/endpoint/resource': { - 'GET': ( - {}, - {'foo': u'bär'} - ) - } - } - fake_http_client = fake_client.FakeHTTPClient(fixtures=fixtures) - self.test_client = TestClient(fake_http_client) - - def test_fake_client_request(self): - resp = self.test_client.get('/endpoint/resource') - self.assertEqual(resp.status_code, 200) - resp_data = resp.json() - self.assertEqual(u'bär', resp_data['foo']) - - def test_fake_client_encode(self): - def guess_json_utf(data): - self.assertIsInstance(data, bytes) - return 'utf-8' - - with mock.patch("requests.utils.guess_json_utf", guess_json_utf): - resp = self.test_client.get('/endpoint/resource') - self.assertEqual(resp.status_code, 200) - - -class FakeClient1(object): - pass - - -class FakeClient21(object): - pass - - -class GetClientClassTestCase(test_base.BaseTestCase): - version_map = { - "1": "%s.FakeClient1" % __name__, - "2.1": "%s.FakeClient21" % __name__, - } - - def test_get_int(self): - self.assertEqual( - client.BaseClient.get_class("fake", 1, self.version_map), - FakeClient1) - - def test_get_str(self): - self.assertEqual( - client.BaseClient.get_class("fake", "2.1", self.version_map), - FakeClient21) - - def test_unsupported_version(self): - self.assertRaises( - exceptions.UnsupportedVersion, - client.BaseClient.get_class, - "fake", "7", self.version_map) diff --git a/tests/unit/apiclient/test_exceptions.py b/tests/unit/apiclient/test_exceptions.py deleted file mode 100644 index ead088e0..00000000 --- a/tests/unit/apiclient/test_exceptions.py +++ /dev/null @@ -1,137 +0,0 @@ -# Copyright 2012 OpenStack Foundation -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. - -from oslotest import base as test_base -import six - -from openstack.common.apiclient import exceptions - - -class FakeResponse(object): - json_data = {} - - def __init__(self, **kwargs): - for key, value in six.iteritems(kwargs): - setattr(self, key, value) - - def json(self): - return self.json_data - - -class ExceptionsArgsTest(test_base.BaseTestCase): - - def assert_exception(self, ex_cls, method, url, status_code, json_data, - error_msg=None, error_details=None, - check_description=True): - ex = exceptions.from_response( - FakeResponse(status_code=status_code, - headers={"Content-Type": "application/json"}, - json_data=json_data), - method, - url) - self.assertTrue(isinstance(ex, ex_cls)) - if check_description: - expected_msg = error_msg or json_data["error"]["message"] - expected_details = error_details or json_data["error"]["details"] - self.assertEqual(ex.message, expected_msg) - self.assertEqual(ex.details, expected_details) - self.assertEqual(ex.method, method) - self.assertEqual(ex.url, url) - self.assertEqual(ex.http_status, status_code) - - def test_from_response_known(self): - method = "GET" - url = "/fake" - status_code = 400 - json_data = {"error": {"message": "fake message", - "details": "fake details"}} - self.assert_exception( - exceptions.BadRequest, method, url, status_code, json_data) - - def test_from_response_unknown(self): - method = "POST" - url = "/fake-unknown" - status_code = 499 - json_data = {"error": {"message": "fake unknown message", - "details": "fake unknown details"}} - self.assert_exception( - exceptions.HTTPClientError, method, url, status_code, json_data) - status_code = 600 - self.assert_exception( - exceptions.HttpError, method, url, status_code, json_data) - - def test_from_response_non_openstack(self): - method = "POST" - url = "/fake-unknown" - status_code = 400 - json_data = {"alien": 123} - self.assert_exception( - exceptions.BadRequest, method, url, status_code, json_data, - check_description=False) - - def test_from_response_with_different_response_format(self): - method = "GET" - url = "/fake-wsme" - status_code = 400 - json_data1 = {"error_message": {"debuginfo": None, - "faultcode": "Client", - "faultstring": "fake message"}} - message = six.text_type( - json_data1["error_message"]["faultstring"]) - details = six.text_type(json_data1) - self.assert_exception( - exceptions.BadRequest, method, url, status_code, json_data1, - message, details) - - json_data2 = {"badRequest": {"message": "fake message", "code": 400}} - message = six.text_type(json_data2["badRequest"]["message"]) - details = six.text_type(json_data2) - self.assert_exception( - exceptions.BadRequest, method, url, status_code, json_data2, - message, details) - - def test_from_response_with_text_response_format(self): - method = "GET" - url = "/fake-wsme" - status_code = 400 - text_data1 = "error_message: fake message" - - ex = exceptions.from_response( - FakeResponse(status_code=status_code, - headers={"Content-Type": "text/html"}, - text=text_data1), - method, - url) - self.assertTrue(isinstance(ex, exceptions.BadRequest)) - self.assertEqual(ex.details, text_data1) - self.assertEqual(ex.method, method) - self.assertEqual(ex.url, url) - self.assertEqual(ex.http_status, status_code) - - def test_from_response_with_text_response_format_with_no_body(self): - method = "GET" - url = "/fake-wsme" - status_code = 401 - - ex = exceptions.from_response( - FakeResponse(status_code=status_code, - headers={"Content-Type": "text/html"}), - method, - url) - self.assertTrue(isinstance(ex, exceptions.Unauthorized)) - self.assertEqual(ex.details, '') - self.assertEqual(ex.method, method) - self.assertEqual(ex.url, url) - self.assertEqual(ex.http_status, status_code) diff --git a/tests/unit/apiclient/test_utils.py b/tests/unit/apiclient/test_utils.py deleted file mode 100644 index 3c8af512..00000000 --- a/tests/unit/apiclient/test_utils.py +++ /dev/null @@ -1,115 +0,0 @@ -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. - -from oslotest import base as test_base - -from openstack.common.apiclient import base -from openstack.common.apiclient import exceptions -from openstack.common.apiclient import utils - - -UUID = '8e8ec658-c7b0-4243-bdf8-6f7f2952c0d0' - - -class FakeResource(object): - NAME_ATTR = 'name' - - def __init__(self, _id, properties): - self.id = _id - try: - self.name = properties['name'] - except KeyError: - pass - - -class FakeManager(base.ManagerWithFind): - - resource_class = FakeResource - - resources = [ - FakeResource('1234', {'name': 'entity_one'}), - FakeResource(UUID, {'name': 'entity_two'}), - FakeResource('5678', {'name': '9876'}) - ] - - def get(self, resource_id): - for resource in self.resources: - if resource.id == str(resource_id): - return resource - raise exceptions.NotFound(resource_id) - - def list(self): - return self.resources - - -class FakeDisplayResource(object): - NAME_ATTR = 'display_name' - - def __init__(self, _id, properties): - self.id = _id - try: - self.display_name = properties['display_name'] - except KeyError: - pass - - -class FakeDisplayManager(FakeManager): - - resource_class = FakeDisplayResource - - resources = [ - FakeDisplayResource('4242', {'display_name': 'entity_three'}), - ] - - -class FindResourceTestCase(test_base.BaseTestCase): - - def setUp(self): - super(FindResourceTestCase, self).setUp() - self.manager = FakeManager(None) - - def test_find_none(self): - """Test a few non-valid inputs.""" - self.assertRaises(exceptions.CommandError, - utils.find_resource, - self.manager, - 'asdf') - self.assertRaises(exceptions.CommandError, - utils.find_resource, - self.manager, - None) - self.assertRaises(exceptions.CommandError, - utils.find_resource, - self.manager, - {}) - - def test_find_by_integer_id(self): - output = utils.find_resource(self.manager, 1234) - self.assertEqual(output, self.manager.get('1234')) - - def test_find_by_str_id(self): - output = utils.find_resource(self.manager, '1234') - self.assertEqual(output, self.manager.get('1234')) - - def test_find_by_uuid(self): - output = utils.find_resource(self.manager, UUID) - self.assertEqual(output, self.manager.get(UUID)) - - def test_find_by_str_name(self): - output = utils.find_resource(self.manager, 'entity_one') - self.assertEqual(output, self.manager.get('1234')) - - def test_find_by_str_displayname(self): - display_manager = FakeDisplayManager(None) - output = utils.find_resource(display_manager, 'entity_three') - self.assertEqual(output, display_manager.get('4242')) |