From 61a756a014eac7a9a479031d2d76187e668c158f Mon Sep 17 00:00:00 2001 From: Sushil Kumar Date: Mon, 8 Jun 2015 07:13:36 +0000 Subject: [PATCH] Added more unit-tests to improve code coverage Following files were too low on code coverage, so the coverage has been enhanced with addition of more unit-tests. - troveclient/client.py - troveclient/v1/shell.py The methodology to test shell.py has been motivated from the one used by python-novaclient to test novaclient/v2/shell.py Also, found an unmocked entry in test_secgroups.py, which was causing intermittent failure while writing the tests for shell.py. Closes-Bug: #1455856 Change-Id: I6a148fb5caad6a83b6ac23411b6458f231064ead --- test-requirements.txt | 1 + troveclient/tests/fakes.py | 517 ++++++++++++++++++++++++++++ troveclient/tests/test_client.py | 454 +++++++++++++++++++++++- troveclient/tests/test_secgroups.py | 14 +- troveclient/tests/test_v1_shell.py | 443 ++++++++++++++++++++++++ troveclient/tests/utils.py | 81 +++++ 6 files changed, 1498 insertions(+), 12 deletions(-) create mode 100644 troveclient/tests/fakes.py create mode 100644 troveclient/tests/test_v1_shell.py create mode 100644 troveclient/tests/utils.py diff --git a/test-requirements.txt b/test-requirements.txt index d0ea1de5..bd8308a0 100644 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -8,6 +8,7 @@ oslosphinx>=2.5.0 # Apache-2.0 requests-mock>=0.6.0 # Apache-2.0 sphinx>=1.1.2,!=1.2.0,!=1.3b1,<1.3 testrepository>=0.0.18 +testscenarios>=0.4 testtools>=0.9.36,!=1.2.0 mock>=1.0 httplib2>=0.7.5 diff --git a/troveclient/tests/fakes.py b/troveclient/tests/fakes.py new file mode 100644 index 00000000..2fe389c3 --- /dev/null +++ b/troveclient/tests/fakes.py @@ -0,0 +1,517 @@ +# Copyright [2015] Hewlett-Packard Development Company, L.P. +# +# 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 six.moves.urllib import parse +from troveclient import client as base_client +from troveclient.tests import utils +from troveclient.v1 import client + + +def get_version_map(): + return { + '1.0': 'troveclient.tests.fakes.FakeClient', + } + + +def assert_has_keys(dict, required=[], optional=[]): + keys = dict.keys() + for k in required: + try: + assert k in keys + except AssertionError: + raise AssertionError("key: %s not found." % k) + + +class FakeClient(client.Client): + + def __init__(self, *args, **kwargs): + client.Client.__init__(self, 'username', 'password', + 'project_id', 'auth_url', + extensions=kwargs.get('extensions')) + self.client = FakeHTTPClient(**kwargs) + + def assert_called(self, method, url, body=None, pos=-1): + """Assert than an API method was just called.""" + expected = (method, url) + called = self.client.callstack[pos][0:2] + + assert self.client.callstack, \ + "Expected %s %s but no calls were made." % expected + + assert expected == called, \ + 'Expected %s %s; got %s %s' % (expected + called) + + if body is not None: + if self.client.callstack[pos][2] != body: + raise AssertionError('%r != %r' % + (self.client.callstack[pos][2], body)) + + def assert_called_anytime(self, method, url, body=None): + """Assert than an API method was called anytime in the test.""" + expected = (method, url) + + assert self.client.callstack, \ + "Expected %s %s but no calls were made." % expected + + found = False + for entry in self.client.callstack: + if expected == entry[0:2]: + found = True + break + + assert found, 'Expected %s; got %s' % (expected, self.client.callstack) + if body is not None: + try: + assert entry[2] == body + except AssertionError: + print(entry[2]) + print("!=") + print(body) + raise + + self.client.callstack = [] + + +class FakeHTTPClient(base_client.HTTPClient): + + def __init__(self, **kwargs): + self.username = 'username' + self.password = 'password' + self.auth_url = 'auth_url' + self.management_url = ( + 'http://trove-api:8779/v1.0/14630bc0e9ef4e248c9753eaf57b0f6e') + self.tenant_id = 'tenant_id' + self.callstack = [] + self.projectid = 'projectid' + self.user = 'user' + self.region_name = 'region_name' + self.endpoint_type = 'endpoint_type' + self.service_type = 'service_type' + self.service_name = 'service_name' + self.volume_service_name = 'volume_service_name' + self.timings = 'timings' + self.bypass_url = 'bypass_url' + self.os_cache = 'os_cache' + self.http_log_debug = 'http_log_debug' + + def _cs_request(self, url, method, **kwargs): + # Check that certain things are called correctly + if method in ['GET', 'DELETE']: + assert 'body' not in kwargs + elif method == 'PUT': + assert 'body' in kwargs + + if url is not None: + # Call the method + args = parse.parse_qsl(parse.urlparse(url)[4]) + kwargs.update(args) + munged_url = url.rsplit('?', 1)[0] + munged_url = munged_url.strip('/').replace('/', '_') + munged_url = munged_url.replace('.', '_') + munged_url = munged_url.replace('-', '_') + munged_url = munged_url.replace(' ', '_') + callback = "%s_%s" % (method.lower(), munged_url) + + if not hasattr(self, callback): + raise AssertionError('Called unknown API method: %s %s, ' + 'expected fakes method name: %s' % + (method, url, callback)) + + # Note the call + self.callstack.append((method, url, kwargs.get('body'))) + + status, headers, body = getattr(self, callback)(**kwargs) + r = utils.TestResponse({ + "status_code": status, + "text": body, + "headers": headers, + }) + return r, body + + def get_instances(self, **kw): + return (200, {}, {"instances": [ + { + "id": "1234", + "name": "test-member-1", + "status": "ACTIVE", + "ip": ["10.0.0.13"], + "volume": {"size": 2}, + "flavor": {"id": "2"}, + "datastore": {"version": "5.6", "type": "mysql"}}, + { + "id": "5678", + "name": "test-member-2", + "status": "ACTIVE", + "ip": ["10.0.0.14"], + "volume": {"size": 2}, + "flavor": {"id": "2"}, + "datastore": {"version": "5.6", "type": "mysql"}}]}) + + def get_instances_1234(self, **kw): + r = {'instance': self.get_instances()[2]['instances'][0]} + return (200, {}, r) + + def post_instances(self, body, **kw): + assert_has_keys( + body['instance'], + required=['name', 'flavorRef'], + optional=['volume']) + if 'volume' in body['instance']: + assert_has_keys(body['instance']['volume'], required=['size']) + return (202, {}, self.get_instances_1234()[2]) + + def get_flavors(self, **kw): + return (200, {}, {"flavors": [ + { + "str_id": "1", + "ram": 512, + "id": 1, + "name": "m1.tiny"}, + { + "str_id": "10", + "ram": 768, + "id": 10, + "name": "eph.rd-smaller"}, + { + "str_id": "2", + "ram": 2048, + "id": 2, + "name": "m1.small"}, + { + "str_id": "3", + "ram": 4096, + "id": 3, + "name": "m1.medium"}]}) + + def get_datastores_mysql_versions_some_version_id_flavors(self, **kw): + return self.get_flavors() + + def get_flavors_1(self, **kw): + r = {'flavor': self.get_flavors()[2]['flavors'][0]} + return (200, {}, r) + + def get_clusters(self, **kw): + return (200, {}, {"clusters": [ + { + "instances": [ + { + "type": "member", + "id": "member-1", + "ip": ["10.0.0.3"], + "flavor": {"id": "2"}, + "name": "test-clstr-member-1" + }, + { + "type": "member", + "id": "member-2", + "ip": ["10.0.0.4"], + "flavor": {"id": "2"}, + "name": "test-clstr-member-2" + }], + "updated": "2015-05-02T11:06:19", + "task": {"description": "No tasks for the cluster.", "id": 1, + "name": "NONE"}, + "name": "test-clstr", + "created": "2015-05-02T10:37:04", + "datastore": {"version": "7.1", "type": "vertica"}, + "id": "cls-1234"}]}) + + def get_clusters_cls_1234(self, **kw): + r = {'cluster': self.get_clusters()[2]['clusters'][0]} + return (200, {}, r) + + def delete_instances_1234(self, **kw): + return (202, {}, None) + + def delete_clusters_cls_1234(self, **kw): + return (202, {}, None) + + def patch_instances_1234(self, **kw): + return (202, {}, None) + + def post_clusters(self, body, **kw): + assert_has_keys( + body['cluster'], + required=['instances', 'datastore', 'name']) + if 'instances' in body['cluster']: + for instance in body['cluster']['instances']: + assert_has_keys(instance, required=['volume', 'flavorRef']) + return (202, {}, self.get_clusters_cls_1234()[2]) + + def post_instances_1234_action(self, **kw): + return (202, {}, None) + + def get_datastores(self, **kw): + return (200, {}, {"datastores": [ + { + "default_version": "v-56", + "versions": [{"id": "v-56", "name": "5.6"}], + "id": "d-123", + "name": "mysql"}, + { + "default_version": "v-71", + "versions": [{"id": "v-71", "name": "7.1"}], + "id": "d-456", + "name": "vertica" + }]}) + + def get_datastores_d_123(self, **kw): + r = {'datastore': self.get_datastores()[2]['datastores'][0]} + return (200, {}, r) + + def get_datastores_d_123_versions(self, **kw): + return (200, {}, {"versions": [ + { + "datastore": "d-123", + "id": "v-56", + "name": "5.6"}]}) + + def get_datastores_d_123_versions_v_56(self, **kw): + r = {'version': self.get_datastores_d_123_versions()[2]['versions'][0]} + return (200, {}, r) + + def get_configurations(self, **kw): + return (200, {}, {"configurations": [ + { + "datastore_name": "mysql", + "updated": "2015-05-16T10:24:29", + "name": "test_config", + "created": "2015-05-16T10:24:28", + "datastore_version_name": "5.6", + "id": "c-123", + "values": {"max_connections": 5}, + "datastore_version_id": "d-123", "description": ''}]}) + + def get_configurations_c_123(self, **kw): + r = {'configuration': self.get_configurations()[2]['configurations'][0] + } + return (200, {}, r) + + def get_datastores_d_123_versions_v_156_parameters(self, **kw): + return (200, {}, {"configuration-parameters": [ + { + "type": "string", + "name": "character_set_results", + "datastore_version_id": "d-123", + "restart_required": "false"}, + { + "name": "connect_timeout", + "min": 2, + "max": 31536000, + "restart_required": "false", + "type": "integer", + "datastore_version_id": "d-123"}, + { + "type": "string", + "name": "character_set_client", + "datastore_version_id": "d-123", + "restart_required": "false"}, + { + "name": "max_connections", + "min": 1, + "max": 100000, + "restart_required": "false", + "type": "integer", + "datastore_version_id": "d-123"}]}) + + def get_datastores_d_123_versions_v_56_parameters_max_connections(self, + **kw): + r = self.get_datastores_d_123_versions_v_156_parameters()[ + 2]['configuration-parameters'][3] + return (200, {}, r) + + def get_configurations_c_123_instances(self, **kw): + return (200, {}, {"instances": []}) + + def delete_configurations_c_123(self, **kw): + return (202, {}, None) + + def get_instances_1234_configuration(self, **kw): + return (200, {}, {"instance": {"configuration": { + "tmp_table_size": "15M", + "innodb_log_files_in_group": "2", + "skip-external-locking": "1", + "max_user_connections": "98"}}}) + + def put_instances_1234(self, **kw): + return (202, {}, None) + + def patch_instances_1234_metadata_key_123(self, **kw): + return (202, {}, None) + + def put_instances_1234_metadata_key_123(self, **kw): + return (202, {}, None) + + def delete_instances_1234_metadata_key_123(self, **kw): + return (202, {}, None) + + def post_instances_1234_metadata_key123(self, body, **kw): + return (202, {}, {'metadata': {}}) + + def get_instances_1234_metadata(self, **kw): + return (200, {}, {"metadata": {}}) + + def get_instances_1234_metadata_key123(self, **kw): + return (200, {}, {"metadata": {}}) + + def get_limits(self, **kw): + return (200, {}, {"limits": [ + { + "max_backups": 50, + "verb": "ABSOLUTE", + "max_volumes": 20, + "max_instances": 5}]}) + + def get_backups(self, **kw): + return (200, {}, {"backups": [ + { + "status": "COMPLETED", + "updated": "2015-05-16T14:23:08", + "description": None, + "datastore": {"version": "5.6", "type": "mysql", + "version_id": "v-56"}, + "id": "bk-1234", + "size": 0.11, + "name": "bkp_1", + "created": "2015-05-16T14:22:28", + "instance_id": "1234", + "parent_id": None, + "locationRef": ("http://backup_srvr/database_backups/" + "bk-1234.xbstream.gz.enc")}, + { + "status": "COMPLETED", + "updated": "2015-05-16T14:22:12", + "description": None, + "datastore": {"version": "5.6", "type": "mysql", + "version_id": "v-56"}, + "id": "bk-5678", + "size": 0.11, + "name": "test_bkp", + "created": "2015-05-16T14:21:27", + "instance_id": "5678", + "parent_id": None, + "locationRef": ("http://backup_srvr/database_backups/" + "bk-5678.xbstream.gz.enc")}]}) + + def get_backups_bk_1234(self, **kw): + r = {'backup': self.get_backups()[2]['backups'][0]} + return (200, {}, r) + + def get_instances_1234_backups(self, **kw): + r = {'backups': [self.get_backups()[2]['backups'][0]]} + return (200, {}, r) + + def delete_backups_bk_1234(self, **kw): + return (202, {}, None) + + def post_backups(self, body, **kw): + assert_has_keys( + body['backup'], + required=['name'], + optional=['description', 'parent']) + return (202, {}, self.get_backups_bk_1234()[2]) + + def get_instances_1234_databases(self, **kw): + return (200, {}, {"databases": [ + {"name": "db_1"}, + {"name": "db_2"}, + {"name": "performance_schema"}]}) + + def delete_instances_1234_databases_db_1(self, **kw): + return (202, {}, None) + + def post_instances_1234_databases(self, body, **kw): + assert_has_keys( + body, + required=['databases']) + for database in body['databases']: + assert_has_keys(database, required=['name'], + optional=['character_set', 'collate']) + return (202, {}, + self.get_instances_1234_databases()[2]['databases'][0]) + + def get_instances_1234_users(self, **kw): + return (200, {}, {"users": [ + {"host": "%", "name": "jacob", "databases": []}, + {"host": "%", "name": "rocky", "databases": []}, + {"host": "%", "name": "harry", "databases": [{"name": "db1"}]}]}) + + def get_instances_1234_users_jacob(self, **kw): + r = {'user': self.get_instances_1234_users()[2]['users'][0]} + return (200, {}, r) + + def delete_instances_1234_users_jacob(self, **kw): + return (202, {}, None) + + def post_instances_1234_users(self, body, **kw): + assert_has_keys( + body, + required=['users']) + for database in body['users']: + assert_has_keys(database, required=['name', 'password'], + optional=['databases']) + return (202, {}, self.get_instances_1234_users()[2]['users'][0]) + + def get_instances_1234_users_jacob_databases(self, **kw): + r = {'databases': [ + self.get_instances_1234_databases()[2]['databases'][0], + self.get_instances_1234_databases()[2]['databases'][1]]} + return (200, {}, r) + + def put_instances_1234_users_jacob(self, **kw): + return (202, {}, None) + + def put_instances_1234_users_jacob_databases(self, **kw): + return (202, {}, None) + + def delete_instances_1234_users_jacob_databases_db1(self, **kw): + return (202, {}, None) + + def post_instances_1234_root(self, **kw): + return (202, {}, {"user": {"password": "password", "name": "root"}}) + + def get_instances_1234_root(self, **kw): + return (200, {}, {"rootEnabled": 'True'}) + + def get_security_groups(self, **kw): + return (200, {}, {"security_groups": [ + { + "instance_id": "1234", + "updated": "2015-05-16T17:29:45", + "name": "SecGroup_1234", + "created": "2015-05-16T17:29:45", + "rules": [{"to_port": 3306, "cidr": "0.0.0.0/0", + "from_port": 3306, + "protocol": "tcp", "id": "1"}], + "id": "2", + "description": "Security Group for 1234"}]}) + + def get_security_groups_2(self, **kw): + r = {'security_group': self.get_security_groups()[ + 2]['security_groups'][0]} + return (200, {}, r) + + def delete_security_group_rules_2(self, **kw): + return (202, {}, None) + + def post_security_group_rules(self, body, **kw): + assert_has_keys(body['security_group_rule'], required=['cidr', 'cidr']) + return (202, {}, {"security_group_rule": [ + { + "from_port": 3306, + "protocol": "tcp", + "created": "2015-05-16T17:55:05", + "to_port": 3306, + "security_group_id": "2", + "cidr": "15.0.0.0/24", "id": 3}]}) diff --git a/troveclient/tests/test_client.py b/troveclient/tests/test_client.py index a22d4702..f901c139 100644 --- a/troveclient/tests/test_client.py +++ b/troveclient/tests/test_client.py @@ -15,13 +15,18 @@ # License for the specific language governing permissions and limitations # under the License. +import fixtures +from keystoneclient import adapter +import logging +import mock +import requests import testtools -import troveclient.v1.client - from troveclient import client as other_client +from troveclient import exceptions from troveclient.openstack.common.apiclient import client -from troveclient.openstack.common.apiclient import exceptions +from troveclient import service_catalog +import troveclient.v1.client class ClientTest(testtools.TestCase): @@ -37,3 +42,446 @@ class ClientTest(testtools.TestCase): self.assertRaises(exceptions.UnsupportedVersion, client.BaseClient.get_class, 'database', '0', version_map) + + def test_client_with_auth_system_without_auth_plugin(self): + self.assertRaisesRegexp( + exceptions.AuthSystemNotFound, "AuthSystemNotFound: 'something'", + other_client.HTTPClient, user='user', password='password', + projectid='project', timeout=2, auth_url="http://www.blah.com", + auth_system='something') + + def test_client_with_auth_system_without_endpoint(self): + auth_plugin = mock.Mock() + auth_plugin.get_auth_url = mock.Mock(return_value=None) + self.assertRaises( + exceptions.EndpointNotFound, + other_client.HTTPClient, user='user', password='password', + projectid='project', timeout=2, auth_plugin=auth_plugin, + auth_url=None, auth_system='something') + + def test_client_with_timeout(self): + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project', + timeout=2, + auth_url="http://www.blah.com", + insecure=True) + self.assertEqual(2, instance.timeout) + mock_request = mock.Mock() + mock_request.return_value = requests.Response() + mock_request.return_value.status_code = 200 + mock_request.return_value.headers = { + 'x-server-management-url': 'blah.com', + 'x-auth-token': 'blah', + } + with mock.patch('requests.request', mock_request): + instance.authenticate() + requests.request.assert_called_with( + mock.ANY, mock.ANY, timeout=2, headers=mock.ANY, + verify=mock.ANY) + + def test_client_unauthorized(self): + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project', + timeout=2, + auth_url="http://www.blah.com", + cacert=mock.Mock()) + instance.auth_token = 'foobar' + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v2.0' + mock_request = mock.Mock() + mock_request.side_effect = other_client.exceptions.Unauthorized(401) + with mock.patch('requests.request', mock_request): + self.assertRaises( + exceptions.Unauthorized, instance.get, '/instances') + + def test_client_bad_request(self): + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project', + timeout=2, + auth_url="http://www.blah.com") + instance.auth_token = 'foobar' + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v2.0' + mock_request = mock.Mock() + mock_request.side_effect = other_client.exceptions.BadRequest() + with mock.patch('requests.request', mock_request): + self.assertRaises( + exceptions.BadRequest, instance.get, '/instances') + + def test_client_with_client_exception(self): + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project', + timeout=2, + auth_url="http://www.blah.com", + retries=2) + instance.auth_token = 'foobar' + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v2.0' + mock_request = mock.Mock() + mock_request.side_effect = other_client.exceptions.ClientException() + type(mock_request.side_effect).code = mock.PropertyMock( + side_effect=[501, 111]) + with mock.patch('requests.request', mock_request): + self.assertRaises( + exceptions.ClientException, instance.get, '/instances') + + def test_client_connection_error(self): + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project', + timeout=2, + auth_url="http://www.blah.com", + retries=2) + instance.auth_token = 'foobar' + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v2.0' + mock_request = mock.Mock() + mock_request.side_effect = requests.exceptions.ConnectionError( + 'connection refused') + with mock.patch('requests.request', mock_request): + self.assertRaisesRegexp( + exceptions.ClientException, + 'Unable to establish connection: connection refused', + instance.get, '/instances') + + @mock.patch.object(other_client.HTTPClient, 'request', + return_value=(200, "{'versions':[]}")) + def _check_version_url(self, management_url, version_url, mock_request): + projectid = '25e469aa1848471b875e68cde6531bc5' + instance = other_client.HTTPClient(user='user', + password='password', + projectid=projectid, + auth_url="http://www.blah.com") + instance.auth_token = 'foobar' + instance.management_url = management_url % projectid + mock_get_service_url = mock.Mock(return_value=instance.management_url) + instance.get_service_url = mock_get_service_url + instance.version = 'v2.0' + + # If passing None as the part of url, a client accesses the url which + # doesn't include "v2/" for getting API version info. + instance.get('') + mock_request.assert_called_once_with(instance.management_url, 'GET', + headers=mock.ANY) + mock_request.reset_mock() + + # Otherwise, a client accesses the url which includes "v2/". + instance.get('/instances') + url = instance.management_url + '/instances' + mock_request.assert_called_once_with(url, 'GET', headers=mock.ANY) + + def test_client_version_url(self): + self._check_version_url('http://foo.com/v1/%s', 'http://foo.com/') + + def test_client_version_url_with_tenant_name(self): + self._check_version_url('http://foo.com/trove/v1/%s', + 'http://foo.com/trove/') + + def test_log_req(self): + self.logger = self.useFixture( + fixtures.FakeLogger( + format="%(message)s", + level=logging.DEBUG, + nuke_handlers=True + ) + ) + cs = other_client.HTTPClient(user='user', + password='password', + projectid=None, + auth_url="http://www.blah.com", + http_log_debug=True) + cs.http_log_req(('/foo', 'GET'), {'headers': {}}) + cs.http_log_req(('/foo', 'GET'), + {'headers': {'X-Auth-Token': 'totally_bogus'}}) + cs.http_log_req( + ('/foo', 'GET'), + {'headers': {}, + 'data': '{"auth": {"passwordCredentials": ' + '{"password": "password"}}}'}) + + output = self.logger.output.split('\n') + + self.assertIn("REQ: curl -i /foo -X GET", output) + self.assertIn( + "REQ: curl -i /foo -X GET -H " + '"X-Auth-Token: totally_bogus"', + output) + self.assertIn( + "REQ: curl -i /foo -X GET -d " + '\'{"auth": {"passwordCredentials": {"password":' + ' "password"}}}\'', + output) + + @mock.patch.object(service_catalog, 'ServiceCatalog') + def test_client_auth_token(self, mock_service_catalog): + auth_url = 'http://www.blah.com' + proxy_token = 'foobar' + proxy_tenant_id = 'user' + mock_service_catalog.return_value.get_token = mock.Mock( + return_value=proxy_token) + instance = other_client.HTTPClient(proxy_token=proxy_token, + proxy_tenant_id=proxy_tenant_id, + user=None, + password=None, + tenant_id=proxy_tenant_id, + projectid=None, + timeout=2, + auth_url=auth_url) + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v2.0' + mock_request = mock.Mock() + mock_request.return_value = requests.Response() + mock_request.return_value.status_code = 200 + mock_request.return_value.headers = { + 'x-server-management-url': 'blah.com', + 'x-auth-token': 'blah', + } + + with mock.patch('requests.request', mock_request): + instance.authenticate() + mock_request.assert_called_with( + 'GET', auth_url + '/tokens/foobar?belongsTo=user', + headers={'User-Agent': 'python-troveclient', + 'Accept': 'application/json', + 'X-Auth-Token': proxy_token}, + timeout=2, verify=True) + + @mock.patch.object(service_catalog, 'ServiceCatalog', side_effect=KeyError) + def test_client_auth_token_authorization_failure(self, + mock_service_catalog): + auth_url = 'http://www.blah.com' + proxy_token = 'foobar' + proxy_tenant_id = 'user' + mock_service_catalog.return_value.get_token = mock.Mock( + return_value=proxy_token) + instance = other_client.HTTPClient(proxy_token=proxy_token, + proxy_tenant_id=proxy_tenant_id, + user=None, + password=None, + tenant_id=proxy_tenant_id, + projectid=None, + timeout=2, + auth_url=auth_url) + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v2.0' + mock_request = mock.Mock() + mock_request.return_value = requests.Response() + mock_request.return_value.status_code = 200 + mock_request.return_value.headers = { + 'x-server-management-url': 'blah.com', + 'x-auth-token': 'blah', + } + + with mock.patch('requests.request', mock_request): + self.assertRaises(exceptions.AuthorizationFailure, + instance.authenticate) + + @mock.patch.object(service_catalog, 'ServiceCatalog', + side_effect=other_client.exceptions.EndpointNotFound) + def test_client_auth_token_endpoint_not_found(self, mock_service_catalog): + auth_url = 'http://www.blah.com' + proxy_token = 'foobar' + proxy_tenant_id = 'user' + mock_service_catalog.return_value.get_token = mock.Mock( + return_value=proxy_token) + instance = other_client.HTTPClient(proxy_token=proxy_token, + proxy_tenant_id=proxy_tenant_id, + user=None, + password=None, + tenant_id=proxy_tenant_id, + projectid=None, + timeout=2, + auth_url=auth_url) + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v2.0' + mock_request = mock.Mock() + mock_request.return_value = requests.Response() + mock_request.return_value.status_code = 200 + mock_request.return_value.headers = { + 'x-server-management-url': 'blah.com', + 'x-auth-token': 'blah', + } + + with mock.patch('requests.request', mock_request): + self.assertRaises(exceptions.EndpointNotFound, + instance.authenticate) + + @mock.patch.object(service_catalog, 'ServiceCatalog') + def test_client_auth_token_v1_auth_failure(self, mock_service_catalog): + auth_url = 'http://www.blah.com' + proxy_token = 'foobar' + proxy_tenant_id = 'user' + mock_service_catalog.return_value.get_token = mock.Mock( + return_value=proxy_token) + instance = other_client.HTTPClient(proxy_token=proxy_token, + proxy_tenant_id=proxy_tenant_id, + user=None, + password=None, + tenant_id=proxy_tenant_id, + projectid=None, + timeout=2, + auth_url=auth_url) + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v1.0' + mock_request = mock.Mock() + mock_request.return_value = requests.Response() + mock_request.return_value.status_code = 200 + mock_request.return_value.headers = { + 'x-server-management-url': 'blah.com', + 'x-auth-token': 'blah', + } + + with mock.patch('requests.request', mock_request): + self.assertRaises(exceptions.NoTokenLookupException, + instance.authenticate) + + @mock.patch.object(service_catalog, 'ServiceCatalog') + def test_client_auth_token_v1_auth(self, mock_service_catalog): + auth_url = 'http://www.blah.com' + proxy_token = 'foobar' + mock_service_catalog.return_value.get_token = mock.Mock( + return_value=proxy_token) + instance = other_client.HTTPClient(user='user', + password='password', + projectid='projectid', + timeout=2, + auth_url=auth_url) + instance.management_url = 'http://example.com' + instance.get_service_url = mock.Mock(return_value='http://example.com') + instance.version = 'v1.0' + mock_request = mock.Mock() + mock_request.return_value = requests.Response() + mock_request.return_value.status_code = 200 + mock_request.return_value.headers = { + 'x-server-management-url': 'blah.com', + } + headers = {'Content-Type': 'application/json', + 'Accept': 'application/json', + 'User-Agent': 'python-troveclient'} + with mock.patch('requests.request', mock_request): + instance.authenticate() + called_args, called_kwargs = mock_request.call_args + self.assertEqual(('POST', 'http://www.blah.com/v2.0/tokens'), + called_args) + self.assertDictEqual(headers, called_kwargs['headers']) + + def test_client_get(self): + auth_url = 'http://www.blah.com' + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project_id', + timeout=2, + auth_url=auth_url) + instance._cs_request = mock.Mock() + + instance.get('clusters') + instance._cs_request.assert_called_with('clusters', 'GET') + + def test_client_patch(self): + auth_url = 'http://www.blah.com' + body = mock.Mock() + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project_id', + timeout=2, + auth_url=auth_url) + instance._cs_request = mock.Mock() + + instance.patch('instances/dummy-instance-id', body=body) + instance._cs_request.assert_called_with( + 'instances/dummy-instance-id', 'PATCH', body=body) + + def test_client_post(self): + auth_url = 'http://www.blah.com' + body = {"add_shard": {}} + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project_id', + timeout=2, + auth_url=auth_url) + instance._cs_request = mock.Mock() + + instance.post('clusters/dummy-cluster-id', body=body) + instance._cs_request.assert_called_with( + 'clusters/dummy-cluster-id', 'POST', body=body) + + def test_client_put(self): + auth_url = 'http://www.blah.com' + body = {"user": {"password": "new_password"}} + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project_id', + timeout=2, + auth_url=auth_url) + instance._cs_request = mock.Mock() + + instance.put('instances/dummy-instance-id/user/dummy-user', body=body) + instance._cs_request.assert_called_with( + 'instances/dummy-instance-id/user/dummy-user', 'PUT', body=body) + + def test_client_delete(self): + auth_url = 'http://www.blah.com' + instance = other_client.HTTPClient(user='user', + password='password', + projectid='project_id', + timeout=2, + auth_url=auth_url) + instance._cs_request = mock.Mock() + + instance.delete('/backups/dummy-backup-id') + instance._cs_request.assert_called_with('/backups/dummy-backup-id', + 'DELETE') + + @mock.patch.object(adapter.LegacyJsonAdapter, 'request') + def test_database_service_name(self, m_request): + m_request.return_value = (mock.MagicMock(status_code=200), None) + + client = other_client.SessionClient(session=mock.MagicMock(), + auth=mock.MagicMock()) + client.request("http://no.where", 'GET') + self.assertIsNone(client.database_service_name) + + client = other_client.SessionClient(session=mock.MagicMock(), + auth=mock.MagicMock(), + database_service_name='myservice') + client.request("http://no.where", 'GET') + self.assertEqual('myservice', client.database_service_name) + + @mock.patch.object(adapter.LegacyJsonAdapter, 'request') + @mock.patch.object(adapter.LegacyJsonAdapter, 'get_endpoint', + return_value=None) + def test_error_sessionclient(self, m_end_point, m_request): + m_request.return_value = (mock.MagicMock(status_code=200), None) + + self.assertRaises(exceptions.EndpointNotFound, + other_client.SessionClient, + session=mock.MagicMock(), + auth=mock.MagicMock()) + + def test_construct_http_client(self): + mock_request = mock.Mock() + mock_request.return_value = requests.Response() + mock_request.return_value.status_code = 200 + mock_request.return_value.headers = { + 'x-server-management-url': 'blah.com', + 'x-auth-token': 'blah', + } + with mock.patch('requests.request', mock_request): + self.assertIsInstance(other_client._construct_http_client(), + other_client.HTTPClient) + self.assertIsInstance( + other_client._construct_http_client(session=mock.Mock(), + auth=mock.Mock()), + other_client.SessionClient) diff --git a/troveclient/tests/test_secgroups.py b/troveclient/tests/test_secgroups.py index 5c658a9f..63d421d0 100644 --- a/troveclient/tests/test_secgroups.py +++ b/troveclient/tests/test_secgroups.py @@ -60,21 +60,17 @@ class SecGroupTest(testtools.TestCase): class SecGroupRuleTest(testtools.TestCase): - def setUp(self): + @mock.patch.object(security_groups.SecurityGroupRules, '__init__', + mock.Mock(return_value=None)) + @mock.patch.object(security_groups.SecurityGroupRule, '__init__', + mock.Mock(return_value=None)) + def setUp(self, *args): super(SecGroupRuleTest, self).setUp() - self.orig__init = security_groups.SecurityGroupRule.__init__ - security_groups.SecurityGroupRule.__init__ = mock.Mock( - return_value=None - ) - security_groups.SecurityGroupRules.__init__ = mock.Mock( - return_value=None - ) self.security_group_rule = security_groups.SecurityGroupRule() self.security_group_rules = security_groups.SecurityGroupRules() def tearDown(self): super(SecGroupRuleTest, self).tearDown() - security_groups.SecurityGroupRule.__init__ = self.orig__init def test___repr__(self): self.security_group_rule.group_id = 1 diff --git a/troveclient/tests/test_v1_shell.py b/troveclient/tests/test_v1_shell.py new file mode 100644 index 00000000..5c377bb9 --- /dev/null +++ b/troveclient/tests/test_v1_shell.py @@ -0,0 +1,443 @@ +# Copyright [2015] Hewlett-Packard Development Company, L.P. +# 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 fixtures +import mock +import six + +import troveclient.client +from troveclient import exceptions +import troveclient.shell +from troveclient.tests import fakes +from troveclient.tests import utils +import troveclient.v1.shell + + +class ShellFixture(fixtures.Fixture): + def setUp(self): + super(ShellFixture, self).setUp() + self.shell = troveclient.shell.OpenStackTroveShell() + + def tearDown(self): + if hasattr(self.shell, 'cs'): + self.shell.cs.clear_callstack() + super(ShellFixture, self).tearDown() + + +class ShellTest(utils.TestCase): + FAKE_ENV = { + 'OS_USERNAME': 'username', + 'OS_PASSWORD': 'password', + 'OS_PROJECT_ID': 'project_id', + 'OS_AUTH_URL': 'http://no.where/v2.0', + } + + def setUp(self, *args): + """Run before each test.""" + super(ShellTest, self).setUp() + + for var in self.FAKE_ENV: + self.useFixture(fixtures.EnvironmentVariable(var, + self.FAKE_ENV[var])) + self.shell = self.useFixture(ShellFixture()).shell + + @mock.patch('sys.stdout', new_callable=six.StringIO) + @mock.patch('troveclient.client.get_version_map', + return_value=fakes.get_version_map()) + def run_command(self, cmd, mock_stdout, mock_get_version_map): + if isinstance(cmd, list): + self.shell.main(cmd) + else: + self.shell.main(cmd.split()) + return mock_stdout.getvalue() + + def assert_called(self, method, url, body=None, **kwargs): + return self.shell.cs.assert_called(method, url, body, **kwargs) + + def assert_called_anytime(self, method, url, body=None): + return self.shell.cs.assert_called_anytime(method, url, body) + + def test_instance_list(self): + self.run_command('list') + self.assert_called('GET', '/instances') + + def test_instance_show(self): + self.run_command('show 1234') + self.assert_called('GET', '/instances/1234') + + def test_instance_delete(self): + self.run_command('delete 1234') + self.assert_called('DELETE', '/instances/1234') + + def test_instance_update(self): + self.run_command('update 1234') + self.assert_called('PATCH', '/instances/1234') + + def test_resize_flavor(self): + self.run_command('resize-flavor 1234 1') + self.assert_called('POST', '/instances/1234/action') + + def test_resize_volume(self): + self.run_command('resize-volume 1234 3') + self.assert_called('POST', '/instances/1234/action') + + def test_restart(self): + self.run_command('restart 1234') + self.assert_called('POST', '/instances/1234/action') + + def test_detach_replica(self): + self.run_command('detach-replica 1234') + self.assert_called('PATCH', '/instances/1234') + + def test_promote_to_replica_source(self): + self.run_command('promote-to-replica-source 1234') + self.assert_called('POST', '/instances/1234/action') + + def test_eject_replica_source(self): + self.run_command('eject-replica-source 1234') + self.assert_called('POST', '/instances/1234/action') + + def test_flavor_list(self): + self.run_command('flavor-list') + self.assert_called('GET', '/flavors') + + def test_flavor_list_with_datastore(self): + cmd = ('flavor-list --datastore_type mysql ' + '--datastore_version_id some-version-id') + self.run_command(cmd) + self.assert_called( + 'GET', '/datastores/mysql/versions/some-version-id/flavors') + + def test_flavor_list_error(self): + cmd = 'flavor-list --datastore_type mysql' + exepcted_error_msg = ('Specify both and ' + ' to list datastore ' + 'version associated flavors') + self.assertRaisesRegexp( + exceptions.CommandError, exepcted_error_msg, self.run_command, cmd) + + def test_flavor_show(self): + self.run_command('flavor-show 1') + self.assert_called('GET', '/flavors/1') + + def test_cluster_list(self): + self.run_command('cluster-list') + self.assert_called('GET', '/clusters') + + def test_cluster_show(self): + self.run_command('cluster-show cls-1234') + self.assert_called('GET', '/clusters/cls-1234') + + def test_cluster_instances(self): + self.run_command('cluster-instances cls-1234') + self.assert_called('GET', '/clusters/cls-1234') + + def test_cluster_delete(self): + self.run_command('cluster-delete cls-1234') + self.assert_called('DELETE', '/clusters/cls-1234') + + def test_boot(self): + self.run_command('create test-member-1 1 --size 1') + self.assert_called_anytime( + 'POST', '/instances', + {'instance': { + 'volume': {'size': 1}, + 'flavorRef': '1', + 'name': 'test-member-1', + 'replica_count': 1 + }}) + + def test_boot_nic_error(self): + cmd = ('create test-member-1 1 --size 1 ' + '--nic net-id=some-id,port-id=some-id') + self.assertRaisesRegexp( + exceptions.CommandError, 'Invalid nic argument', + self.run_command, cmd) + + def test_cluster_create(self): + cmd = ('cluster-create test-clstr vertica 7.1 ' + '--instance flavor_id=2,volume=2 ' + '--instance flavor_id=2,volume=1') + self.run_command(cmd) + self.assert_called_anytime( + 'POST', '/clusters', + {'cluster': { + 'instances': [ + { + 'volume': {'size': '2'}, + 'flavorRef': '2' + }, + { + 'volume': {'size': '1'}, + 'flavorRef': '2' + }], + 'datastore': {'version': '7.1', 'type': 'vertica'}, + 'name': 'test-clstr'}}) + + def test_cluster_create_error(self): + cmd = ('cluster-create test-clstr vertica 7.1 --instance volume=2 ' + '--instance flavor_id=2,volume=1') + self.assertRaisesRegexp( + exceptions.CommandError, 'flavor_id is required', + self.run_command, cmd) + + def test_datastore_list(self): + self.run_command('datastore-list') + self.assert_called('GET', '/datastores') + + def test_datastore_show(self): + self.run_command('datastore-show d-123') + self.assert_called('GET', '/datastores/d-123') + + def test_datastore_version_list(self): + self.run_command('datastore-version-list d-123') + self.assert_called('GET', '/datastores/d-123/versions') + + def test_datastore_version_show(self): + self.run_command('datastore-version-show v-56 --datastore d-123') + self.assert_called('GET', '/datastores/d-123/versions/v-56') + + def test_datastore_version_show_error(self): + expected_error_msg = ('The datastore name or id is required to ' + 'retrieve a datastore version by name.') + self.assertRaisesRegexp(exceptions.NoUniqueMatch, expected_error_msg, + self.run_command, + 'datastore-version-show v-56') + + def test_configuration_list(self): + self.run_command('configuration-list') + self.assert_called('GET', '/configurations') + + def test_configuration_show(self): + self.run_command('configuration-show c-123') + self.assert_called('GET', '/configurations/c-123') + + def test_configuration_create(self): + cmd = "configuration-create c-123 some-thing" + self.assertRaises(ValueError, self.run_command, cmd) + + def test_configuration_update(self): + cmd = "configuration-update c-123 some-thing" + self.assertRaises(ValueError, self.run_command, cmd) + + def test_configuration_patch(self): + cmd = "configuration-patch c-123 some-thing" + self.assertRaises(ValueError, self.run_command, cmd) + + def test_configuration_parameter_list(self): + cmd = 'configuration-parameter-list v-156 --datastore d-123' + self.run_command(cmd) + self.assert_called('GET', + '/datastores/d-123/versions/v-156/parameters') + + def test_configuration_parameter_list_error(self): + expected_error_msg = ('The datastore name or id is required to ' + 'retrieve the parameters for the configuration ' + 'group by name') + self.assertRaisesRegexp( + exceptions.NoUniqueMatch, expected_error_msg, + self.run_command, 'configuration-parameter-list v-156') + + def test_configuration_parameter_show(self): + cmd = ('configuration-parameter-show v_56 ' + 'max_connections --datastore d_123') + self.run_command(cmd) + self.assert_called( + 'GET', + '/datastores/d_123/versions/v_56/parameters/max_connections') + + def test_configuration_instances(self): + cmd = 'configuration-instances c-123' + self.run_command(cmd) + self.assert_called('GET', '/configurations/c-123/instances') + + def test_configuration_delete(self): + self.run_command('configuration-delete c-123') + self.assert_called('DELETE', '/configurations/c-123') + + def test_configuration_default(self): + self.run_command('configuration-default 1234') + self.assert_called('GET', '/instances/1234/configuration') + + def test_configuration_attach(self): + self.run_command('configuration-attach 1234 c-123') + self.assert_called('PUT', '/instances/1234') + + def test_configuration_detach(self): + self.run_command('configuration-detach 1234') + self.assert_called('PUT', '/instances/1234') + + def test_metadata_edit(self): + self.run_command('metadata-edit 1234 key-123 value-123') + self.assert_called('PATCH', '/instances/1234/metadata/key-123') + + def test_metadata_update(self): + self.run_command('metadata-update 1234 key-123 key-456 value-123') + self.assert_called('PUT', '/instances/1234/metadata/key-123') + + def test_metadata_delete(self): + self.run_command('metadata-delete 1234 key-123') + self.assert_called('DELETE', '/instances/1234/metadata/key-123') + + def test_metadata_create(self): + self.run_command('metadata-create 1234 key123 value123') + self.assert_called_anytime( + 'POST', '/instances/1234/metadata/key123', + {'metadata': {'value': 'value123'}}) + + def test_metadata_list(self): + self.run_command('metadata-list 1234') + self.assert_called('GET', '/instances/1234/metadata') + + def test_metadata_show(self): + self.run_command('metadata-show 1234 key123') + self.assert_called('GET', '/instances/1234/metadata/key123') + + def test_limit_list(self): + self.run_command('limit-list') + self.assert_called('GET', '/limits') + + def test_backup_list(self): + self.run_command('backup-list') + self.assert_called('GET', '/backups') + + def test_backup_show(self): + self.run_command('backup-show bk-1234') + self.assert_called('GET', '/backups/bk-1234') + + def test_backup_list_instance(self): + self.run_command('backup-list-instance 1234') + self.assert_called('GET', '/instances/1234/backups') + + def test_backup_delete(self): + self.run_command('backup-delete bk-1234') + self.assert_called('DELETE', '/backups/bk-1234') + + def test_backup_create(self): + self.run_command('backup-create 1234 bkp_1') + self.assert_called_anytime( + 'POST', '/backups', + {'backup': { + 'instance': '1234', + 'name': 'bkp_1' + }}) + + def test_backup_copy(self): + self.run_command('backup-copy new_bkp bk-1234') + self.assert_called_anytime( + 'POST', '/backups', + {'backup': { + 'name': 'new_bkp', + 'backup': {'region': None, 'id': 'bk-1234'} + }}) + + def test_database_list(self): + self.run_command('database-list 1234') + self.assert_called('GET', '/instances/1234/databases') + + def test_database_delete(self): + self.run_command('database-delete 1234 db_1') + self.assert_called('DELETE', '/instances/1234/databases/db_1') + + def test_database_create(self): + cmd = ('database-create 1234 db_1 --character_set utf8 ' + '--collate utf8_general_ci') + self.run_command(cmd) + self.assert_called_anytime( + 'POST', '/instances/1234/databases', + {'databases': [{'character_set': 'utf8', + 'name': 'db_1', + 'collate': 'utf8_general_ci'}]}) + + def test_user_list(self): + self.run_command('user-list 1234') + self.assert_called('GET', '/instances/1234/users') + + def test_user_show(self): + self.run_command('user-show 1234 jacob') + self.assert_called('GET', '/instances/1234/users/jacob') + + def test_user_delete(self): + self.run_command('user-delete 1234 jacob') + self.assert_called('DELETE', '/instances/1234/users/jacob') + + def test_user_create(self): + self.run_command('user-create 1234 jacob password') + self.assert_called_anytime( + 'POST', '/instances/1234/users', + {'users': [{ + 'password': 'password', + 'name': 'jacob', + 'databases': []}]}) + + def test_user_show_access(self): + self.run_command('user-show-access 1234 jacob') + self.assert_called('GET', '/instances/1234/users/jacob/databases') + + def test_user_update_host(self): + cmd = 'user-update-attributes 1234 jacob --new_host 10.0.0.1' + self.run_command(cmd) + self.assert_called('PUT', '/instances/1234/users/jacob') + + def test_user_update_name(self): + self.run_command('user-update-attributes 1234 jacob --new_name sam') + self.assert_called('PUT', '/instances/1234/users/jacob') + + def test_user_update_password(self): + cmd = 'user-update-attributes 1234 jacob --new_password new_pwd' + self.run_command(cmd) + self.assert_called('PUT', '/instances/1234/users/jacob') + + def test_user_grant_access(self): + self.run_command('user-grant-access 1234 jacob db1 db2') + self.assert_called('PUT', '/instances/1234/users/jacob/databases') + + def test_user_revoke_access(self): + self.run_command('user-revoke-access 1234 jacob db1') + self.assert_called('DELETE', + '/instances/1234/users/jacob/databases/db1') + + def test_root_enable(self): + self.run_command('root-enable 1234') + self.assert_called_anytime('POST', '/instances/1234/root') + + def test_root_show(self): + self.run_command('root-show 1234') + self.assert_called('GET', '/instances/1234/root') + + def test_secgroup_list(self): + self.run_command('secgroup-list') + self.assert_called('GET', '/security-groups') + + def test_secgroup_show(self): + self.run_command('secgroup-show 2') + self.assert_called('GET', '/security-groups/2') + + def test_secgroup_list_rules(self): + self.run_command('secgroup-list-rules 2') + self.assert_called('GET', '/security-groups/2') + + def test_secgroup_delete_rule(self): + self.run_command('secgroup-delete-rule 2') + self.assert_called('DELETE', '/security-group-rules/2') + + def test_secgroup_add_rule(self): + self.run_command('secgroup-add-rule 2 15.0.0.0/24') + self.assert_called_anytime( + 'POST', '/security-group-rules', + {'security_group_rule': { + 'cidr': '15.0.0.0/24', + 'group_id': '2', + }}) diff --git a/troveclient/tests/utils.py b/troveclient/tests/utils.py new file mode 100644 index 00000000..c3e81ac8 --- /dev/null +++ b/troveclient/tests/utils.py @@ -0,0 +1,81 @@ +# Copyright [2015] Hewlett-Packard Development Company, L.P. +# +# 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 os + +import fixtures +import mock +import requests +import testtools + +AUTH_URL = "http://localhost:5002/auth_url" +AUTH_URL_V1 = "http://localhost:5002/auth_url/v1.0" +AUTH_URL_V2 = "http://localhost:5002/auth_url/v2.0" + + +def _patch_mock_to_raise_for_invalid_assert_calls(): + def raise_for_invalid_assert_calls(wrapped): + def wrapper(_self, name): + valid_asserts = [ + 'assert_called_with', + 'assert_called_once_with', + 'assert_has_calls', + 'assert_any_calls'] + + if name.startswith('assert') and name not in valid_asserts: + raise AttributeError('%s is not a valid mock assert method' + % name) + + return wrapped(_self, name) + return wrapper + mock.Mock.__getattr__ = raise_for_invalid_assert_calls( + mock.Mock.__getattr__) + +# NOTE(gibi): needs to be called only once at import time +# to patch the mock lib +_patch_mock_to_raise_for_invalid_assert_calls() + + +class TestCase(testtools.TestCase): + TEST_REQUEST_BASE = { + 'verify': True, + } + + def setUp(self): + super(TestCase, self).setUp() + if (os.environ.get('OS_STDOUT_CAPTURE') == 'True' or + os.environ.get('OS_STDOUT_CAPTURE') == '1'): + stdout = self.useFixture(fixtures.StringStream('stdout')).stream + self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout)) + if (os.environ.get('OS_STDERR_CAPTURE') == 'True' or + os.environ.get('OS_STDERR_CAPTURE') == '1'): + stderr = self.useFixture(fixtures.StringStream('stderr')).stream + self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr)) + + +class TestResponse(requests.Response): + """Class used to wrap requests.Response and provide some + convenience to initialize with a dict + """ + + def __init__(self, data): + super(TestResponse, self).__init__() + self._text = None + if isinstance(data, dict): + self.status_code = data.get('status_code') + self.headers = data.get('headers') + # Fake the text attribute to streamline Response creation + self._text = data.get('text') + else: + self.status_code = data