diff --git a/keystoneclient/tests/fakes.py b/keystoneclient/tests/fakes.py deleted file mode 100644 index 70abcc4f0..000000000 --- a/keystoneclient/tests/fakes.py +++ /dev/null @@ -1,120 +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. - -""" -A fake server that "responds" to API methods with pre-canned responses. - -All of these responses come from the spec, so if for some reason the spec's -wrong the tests might raise AssertionError. I've indicated in comments the -places where actual behavior differs from the spec. -""" - -from keystoneclient import access - - -def assert_has_keys(dict, required=None, optional=None): - required = required or [] - optional = optional or [] - keys = dict.keys() - for k in required: - try: - assert k in keys - except AssertionError: - extra_keys = set(keys).difference(set(required + optional)) - raise AssertionError("found unexpected keys: %s" % - list(extra_keys)) - - -class FakeClient(object): - - def assert_called(self, method, url, body=None, pos=-1): - """Assert than an API method was just called.""" - expected = (method, url) - called = self.callstack[pos][0:2] - - assert self.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: - assert self.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.callstack, ("Expected %s %s but no calls were made." % - expected) - - found = False - for entry in self.callstack: - if expected == entry[0:2]: - found = True - break - - assert found, ('Expected %s; got %s' % - (expected, self.callstack)) - if body is not None: - if entry[2] != body: - raise AssertionError('%s != %s' % (entry[2], body)) - self.callstack = [] - - def clear_callstack(self): - self.callstack = [] - - def authenticate(self, cl_obj): - cl_obj.user_id = '1' - cl_obj.auth_user_id = '1' - cl_obj.project_id = '1' - cl_obj.auth_tenant_id = '1' - cl_obj.auth_ref = access.AccessInfo.factory(None, { - "access": { - "token": { - "expires": "2012-02-05T00:00:00", - "id": "887665443383838", - "tenant": { - "id": "1", - "name": "customer-x" - } - }, - "serviceCatalog": [{ - "endpoints": [{ - "adminURL": "http://swift.admin-nets.local:8080/", - "region": "RegionOne", - "internalURL": "http://127.0.0.1:8080/v1/AUTH_1", - "publicURL": - "http://swift.publicinternets.com/v1/AUTH_1" - }], - "type": "object-store", - "name": "swift" - }, { - "endpoints": [{ - "adminURL": "http://cdn.admin-nets.local/v1.1/1", - "region": "RegionOne", - "internalURL": "http://127.0.0.1:7777/v1.1/1", - "publicURL": "http://cdn.publicinternets.com/v1.1/1" - }], - "type": "object-store", - "name": "cdn" - }], - "user": { - "id": "1", - "roles": [{ - "tenantId": "1", - "id": "3", - "name": "Member" - }], - "name": "joeuser" - } - } - }) diff --git a/keystoneclient/tests/v2_0/fakes.py b/keystoneclient/tests/v2_0/fakes.py deleted file mode 100644 index 9b9ef6f9d..000000000 --- a/keystoneclient/tests/v2_0/fakes.py +++ /dev/null @@ -1,495 +0,0 @@ -# Copyright (c) 2011 X.commerce, a business unit of eBay Inc. -# Copyright 2011 OpenStack Foundation -# -# 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 as urlparse - -from keystoneclient.tests import fakes -from keystoneclient.tests.v2_0 import utils - - -class FakeHTTPClient(fakes.FakeClient): - def __init__(self, **kwargs): - self.username = 'username' - self.password = 'password' - self.auth_url = 'auth_url' - self.callstack = [] - - 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': - kwargs.setdefault('body', None) - - # Call the method - args = urlparse.parse_qsl(urlparse.urlparse(url)[4]) - kwargs.update(args) - munged_url = url.rsplit('?', 1)[0] - munged_url = munged_url.strip('/').replace('/', '_').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'))) - - 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, body = getattr(self, callback)(**kwargs) - r = utils.TestResponse({ - "status_code": status, - "text": body}) - return r, body - - # - # List all extensions - # - def post_tokens(self, **kw): - body = [ - {"access": - {"token": - {"expires": "2012-02-05T00:00:00", - "id": "887665443383838", - "tenant": - {"id": "1", - "name": "customer-x"}}, - "serviceCatalog": [ - {"endpoints": [ - {"adminURL": "http://swift.admin-nets.local:8080/", - "region": "RegionOne", - "internalURL": "http://127.0.0.1:8080/v1/AUTH_1", - "publicURL": - "http://swift.publicinternets.com/v1/AUTH_1"}], - "type": "object-store", - "name": "swift"}, - {"endpoints": [ - {"adminURL": "http://cdn.admin-nets.local/v1.1/1", - "region": "RegionOne", - "internalURL": "http://127.0.0.1:7777/v1.1/1", - "publicURL": - "http://cdn.publicinternets.com/v1.1/1"}], - "type": "object-store", - "name": "cdn"}], - "user": - {"id": "1", - "roles": [ - {"tenantId": "1", - "id": "3", - "name": "Member"}], - "name": "joeuser"}} - } - ] - return (200, body) - - def get_tokens_887665443383838(self, **kw): - body = [ - {"access": - {"token": - {"expires": "2012-02-05T00:00:00", - "id": "887665443383838", - "tenant": {"id": "1", - "name": "customer-x"}}, - "user": - {"name": "joeuser", - "tenantName": "customer-x", - "id": "1", - "roles": [{"serviceId": "1", - "id": "3", - "name": "Member"}], - "tenantId": "1"}} - } - ] - return (200, body) - - def get_tokens_887665443383838_endpoints(self, **kw): - body = [ - {"endpoints_links": [ - {"href": - "http://127.0.0.1:35357/tokens/887665443383838" - "/endpoints?'marker=5&limit=10'", - "rel": "next"}], - "endpoints": [ - {"internalURL": "http://127.0.0.1:8080/v1/AUTH_1", - "name": "swift", - "adminURL": "http://swift.admin-nets.local:8080/", - "region": "RegionOne", - "tenantId": 1, - "type": "object-store", - "id": 1, - "publicURL": "http://swift.publicinternets.com/v1/AUTH_1"}, - {"internalURL": "http://localhost:8774/v1.0", - "name": "nova_compat", - "adminURL": "http://127.0.0.1:8774/v1.0", - "region": "RegionOne", - "tenantId": 1, - "type": "compute", - "id": 2, - "publicURL": "http://nova.publicinternets.com/v1.0/"}, - {"internalURL": "http://localhost:8774/v1.1", - "name": "nova", - "adminURL": "http://127.0.0.1:8774/v1.1", - "region": "RegionOne", - "tenantId": 1, - "type": "compute", - "id": 3, - "publicURL": "http://nova.publicinternets.com/v1.1/"}, - {"internalURL": "http://127.0.0.1:9292/v1.1/", - "name": "glance", - "adminURL": "http://nova.admin-nets.local/v1.1/", - "region": "RegionOne", - "tenantId": 1, - "type": "image", - "id": 4, - "publicURL": "http://glance.publicinternets.com/v1.1/"}, - {"internalURL": "http://127.0.0.1:7777/v1.1/1", - "name": "cdn", - "adminURL": "http://cdn.admin-nets.local/v1.1/1", - "region": "RegionOne", - "tenantId": 1, - "versionId": "1.1", - "versionList": "http://127.0.0.1:7777/", - "versionInfo": "http://127.0.0.1:7777/v1.1", - "type": "object-store", - "id": 5, - "publicURL": "http://cdn.publicinternets.com/v1.1/1"}] - } - ] - return (200, body) - - def get(self, **kw): - body = { - "version": { - "id": "v2.0", - "status": "beta", - "updated": "2011-11-19T00:00:00Z", - "links": [ - {"rel": "self", - "href": "http://127.0.0.1:35357/v2.0/"}, - {"rel": "describedby", - "type": "text/html", - "href": "http://docs.openstack.org/" - "api/openstack-identity-service/2.0/content/"}, - {"rel": "describedby", - "type": "application/pdf", - "href": "http://docs.openstack.org/api/" - "openstack-identity-service/2.0/" - "identity-dev-guide-2.0.pdf"}, - {"rel": "describedby", - "type": "application/vnd.sun.wadl+xml", - "href": - "http://127.0.0.1:35357/v2.0/identity-admin.wadl"}], - "media-types": [ - {"base": "application/xml", - "type": "application/vnd.openstack.identity-v2.0+xml"}, - {"base": "application/json", - "type": "application/vnd.openstack.identity-v2.0+json"}] - } - } - return (200, body) - - def get_extensions(self, **kw): - body = { - "extensions": {"values": []} - } - return (200, body) - - def post_tenants(self, **kw): - body = {"tenant": - {"enabled": True, - "description": None, - "name": "new-tenant", - "id": "1"}} - return (200, body) - - def post_tenants_2(self, **kw): - body = {"tenant": - {"enabled": False, - "description": "desc", - "name": "new-tenant1", - "id": "2"}} - return (200, body) - - def get_tenants(self, **kw): - body = { - "tenants_links": [], - "tenants": [ - {"enabled": False, - "description": None, - "name": "project-y", - "id": "1"}, - {"enabled": True, - "description": None, - "name": "new-tenant", - "id": "2"}, - {"enabled": True, - "description": None, - "name": "customer-x", - "id": "1"}] - } - return (200, body) - - def get_tenants_1(self, **kw): - body = {"tenant": - {"enabled": True, - "description": None, - "name": "new-tenant", - "id": "1"}} - return (200, body) - - def get_tenants_2(self, **kw): - body = {"tenant": - {"enabled": True, - "description": None, - "name": "new-tenant", - "id": "2"}} - return (200, body) - - def delete_tenants_2(self, **kw): - body = {} - return (200, body) - - def get_tenants_1_users_1_roles(self, **kw): - body = { - "roles": [ - {"id": "1", - "name": "Admin"}, - {"id": "2", - "name": "Member"}, - {"id": "3", - "name": "new-role"}] - } - return (200, body) - - def put_users_1_roles_OS_KSADM_1(self, **kw): - body = { - "roles": - {"id": "1", - "name": "Admin"}} - return (200, body) - - def delete_users_1_roles_OS_KSADM_1(self, **kw): - body = {} - return (200, body) - - def put_tenants_1_users_1_roles_OS_KSADM_1(self, **kw): - body = { - "role": - {"id": "1", - "name": "Admin"}} - return (200, body) - - def get_users(self, **kw): - body = { - "users": [ - {"name": self.username, - "enabled": "true", - "email": "sdfsdf@sdfsd.sdf", - "id": "1", - "tenantId": "1"}, - {"name": "user2", - "enabled": "true", - "email": "sdfsdf@sdfsd.sdf", - "id": "2", - "tenantId": "1"}] - } - return (200, body) - - def get_users_1(self, **kw): - body = { - "user": { - "tenantId": "1", - "enabled": "true", - "id": "1", - "name": self.username} - } - return (200, body) - - def put_users_1(self, **kw): - body = { - "user": { - "tenantId": "1", - "enabled": "true", - "id": "1", - "name": "new-user1", - "email": "user@email.com"} - } - return (200, body) - - def put_users_1_OS_KSADM_password(self, **kw): - body = { - "user": { - "tenantId": "1", - "enabled": "true", - "id": "1", - "name": "new-user1", - "email": "user@email.com"} - } - return (200, body) - - def post_users(self, **kw): - body = { - "user": { - "tenantId": "1", - "enabled": "true", - "id": "1", - "name": self.username} - } - return (200, body) - - def delete_users_1(self, **kw): - body = [] - return (200, body) - - def get_users_1_roles(self, **kw): - body = [ - {"roles_links": [], - "roles":[ - {"id": "2", - "name": "KeystoneServiceAdmin"}] - } - ] - return (200, body) - - def post_OS_KSADM_roles(self, **kw): - body = {"role": - {"name": "new-role", - "id": "1"}} - return (200, body) - - def get_OS_KSADM_roles(self, **kw): - body = {"roles": [ - {"id": "10", "name": "admin"}, - {"id": "20", "name": "member"}, - {"id": "1", "name": "new-role"}] - } - return (200, body) - - def get_OS_KSADM_roles_1(self, **kw): - body = {"role": - {"name": "new-role", - "id": "1"} - } - return (200, body) - - def delete_OS_KSADM_roles_1(self, **kw): - body = {} - return (200, body) - - def post_OS_KSADM_services(self, **kw): - body = {"OS-KSADM:service": - {"id": "1", - "type": "compute", - "name": "service1", - "description": None} - } - return (200, body) - - def get_OS_KSADM_services_1(self, **kw): - body = {"OS-KSADM:service": - {"description": None, - "type": "compute", - "id": "1", - "name": "service1"} - } - return (200, body) - - def get_OS_KSADM_services(self, **kw): - body = { - "OS-KSADM:services": [ - {"description": None, - "type": "compute", - "id": "1", - "name": "service1"}, - {"description": None, - "type": "identity", - "id": "2", - "name": "service2"}] - } - return (200, body) - - def delete_OS_KSADM_services_1(self, **kw): - body = {} - return (200, body) - - def post_users_1_credentials_OS_EC2(self, **kw): - body = {"credential": - {"access": "1", - "tenant_id": "1", - "secret": "1", - "user_id": "1"} - } - return (200, body) - - def get_users_1_credentials_OS_EC2(self, **kw): - body = {"credentials": [ - {"access": "1", - "tenant_id": "1", - "secret": "1", - "user_id": "1"}] - } - return (200, body) - - def get_users_1_credentials_OS_EC2_2(self, **kw): - body = { - "credential": - {"access": "2", - "tenant_id": "1", - "secret": "1", - "user_id": "1"} - } - return (200, body) - - def delete_users_1_credentials_OS_EC2_2(self, **kw): - body = {} - return (200, body) - - def patch_OS_KSCRUD_users_1(self, **kw): - body = {} - return (200, body) - - def get_endpoints(self, **kw): - body = { - 'endpoints': [ - {'adminURL': 'http://cdn.admin-nets.local/v1.1/1', - 'region': 'RegionOne', - 'internalURL': 'http://127.0.0.1:7777/v1.1/1', - 'publicURL': 'http://cdn.publicinternets.com/v1.1/1'}], - 'type': 'compute', - 'name': 'nova-compute' - } - return (200, body) - - def post_endpoints(self, **kw): - body = { - "endpoint": - {"adminURL": "http://swift.admin-nets.local:8080/", - "region": "RegionOne", - "internalURL": "http://127.0.0.1:8080/v1/AUTH_1", - "publicURL": "http://swift.publicinternets.com/v1/AUTH_1"}, - "type": "compute", - "name": "nova-compute" - } - return (200, body) diff --git a/keystoneclient/tests/v2_0/test_shell.py b/keystoneclient/tests/v2_0/test_shell.py index 54ca0d433..5f80fc1da 100644 --- a/keystoneclient/tests/v2_0/test_shell.py +++ b/keystoneclient/tests/v2_0/test_shell.py @@ -14,12 +14,10 @@ import os import sys import mock -from mox3 import stubout import six from testtools import matchers -from keystoneclient import httpclient -from keystoneclient.tests.v2_0 import fakes +from keystoneclient import fixture from keystoneclient.tests.v2_0 import utils @@ -28,25 +26,18 @@ DEFAULT_PASSWORD = 'password' DEFAULT_TENANT_ID = 'tenant_id' DEFAULT_TENANT_NAME = 'tenant_name' DEFAULT_AUTH_URL = 'http://127.0.0.1:5000/v2.0/' +DEFAULT_ADMIN_URL = 'http://127.0.0.1:35357/v2.0/' class ShellTests(utils.TestCase): + TEST_URL = DEFAULT_ADMIN_URL + def setUp(self): """Patch os.environ to avoid required auth info.""" super(ShellTests, self).setUp() - self.stubs = stubout.StubOutForTesting() - self.fake_client = fakes.FakeHTTPClient() - self.stubs.Set( - httpclient.HTTPClient, "_cs_request", - lambda ign_self, *args, **kwargs: - self.fake_client._cs_request(*args, **kwargs)) - self.stubs.Set( - httpclient.HTTPClient, "authenticate", - lambda cl_obj: - self.fake_client.authenticate(cl_obj)) self.old_environment = os.environ.copy() os.environ = { 'OS_USERNAME': DEFAULT_USERNAME, @@ -58,11 +49,16 @@ class ShellTests(utils.TestCase): import keystoneclient.shell self.shell = keystoneclient.shell.OpenStackIdentityShell() + self.token = fixture.V2Token() + self.token.set_scope() + svc = self.token.add_service('identity') + svc.add_endpoint(public=DEFAULT_AUTH_URL, + admin=DEFAULT_ADMIN_URL) + + self.stub_auth(json=self.token, base_url=DEFAULT_AUTH_URL) + def tearDown(self): - self.stubs.UnsetAll() - self.stubs.SmartUnsetAll() os.environ = self.old_environment - self.fake_client.clear_callstack() super(ShellTests, self).tearDown() def run_command(self, cmd): @@ -82,250 +78,339 @@ class ShellTests(utils.TestCase): sys.stdout = orig return out - def assert_called(self, method, url, body=None, **kwargs): - return self.fake_client.assert_called(method, url, body, **kwargs) - - def assert_called_anytime(self, method, url, body=None): - return self.fake_client.assert_called_anytime(method, url, body) + def assert_called(self, method, path, base_url=TEST_URL): + self.assertEqual(method, self.requests.last_request.method) + self.assertEqual(base_url + path.lstrip('/'), + self.requests.last_request.url) def test_user_list(self): + self.stub_url('GET', ['users'], json={'users': []}) self.run_command('user-list') - self.fake_client.assert_called_anytime('GET', '/users') + self.assert_called('GET', '/users') def test_user_create(self): + self.stub_url('POST', ['users'], json={'user': {}}) self.run_command('user-create --name new-user') - self.fake_client.assert_called_anytime( - 'POST', '/users', - {'user': - {'email': None, - 'password': None, - 'enabled': True, - 'name': 'new-user', - 'tenantId': None}}) + + self.assert_called('POST', '/users') + self.assertRequestBodyIs(json={'user': {'email': None, + 'password': None, + 'enabled': True, + 'name': 'new-user', + 'tenantId': None}}) @mock.patch('sys.stdin', autospec=True) def test_user_create_password_prompt(self, mock_stdin): + self.stub_url('POST', ['users'], json={'user': {}}) + with mock.patch('getpass.getpass') as mock_getpass: mock_getpass.return_value = 'newpass' self.run_command('user-create --name new-user --pass') - self.fake_client.assert_called_anytime( - 'POST', '/users', - {'user': - {'email': None, - 'password': 'newpass', - 'enabled': True, - 'name': 'new-user', - 'tenantId': None}}) + + self.assert_called('POST', '/users') + self.assertRequestBodyIs(json={'user': {'email': None, + 'password': 'newpass', + 'enabled': True, + 'name': 'new-user', + 'tenantId': None}}) def test_user_get(self): + self.stub_url('GET', ['users', '1'], + json={'user': {'id': '1'}}) self.run_command('user-get 1') - self.fake_client.assert_called_anytime('GET', '/users/1') + self.assert_called('GET', '/users/1') def test_user_delete(self): + self.stub_url('GET', ['users', '1'], + json={'user': {'id': '1'}}) + self.stub_url('DELETE', ['users', '1']) self.run_command('user-delete 1') - self.fake_client.assert_called_anytime('DELETE', '/users/1') + self.assert_called('DELETE', '/users/1') def test_user_password_update(self): + self.stub_url('GET', ['users', '1'], + json={'user': {'id': '1'}}) + self.stub_url('PUT', ['users', '1', 'OS-KSADM', 'password']) self.run_command('user-password-update --pass newpass 1') - self.fake_client.assert_called_anytime( - 'PUT', '/users/1/OS-KSADM/password') + self.assert_called('PUT', '/users/1/OS-KSADM/password') def test_user_update(self): + self.stub_url('PUT', ['users', '1']) + self.stub_url('GET', ['users', '1'], + json={"user": {"tenantId": "1", + "enabled": "true", + "id": "1", + "name": "username"}}) + self.run_command('user-update --name new-user1' ' --email user@email.com --enabled true 1') - self.fake_client.assert_called_anytime( - 'PUT', '/users/1', - {'user': - {'id': '1', - 'email': 'user@email.com', - 'enabled': True, - 'name': 'new-user1'} - }) + self.assert_called('PUT', '/users/1') + self.assertRequestBodyIs(json={'user': {'id': '1', + 'email': 'user@email.com', + 'enabled': True, + 'name': 'new-user1'}}) + required = 'User not updated, no arguments present.' out = self.run_command('user-update 1') self.assertThat(out, matchers.MatchesRegex(required)) self.run_command(['user-update', '--email', '', '1']) - self.fake_client.assert_called_anytime( - 'PUT', '/users/1', - {'user': - {'id': '1', - 'email': ''} - }) + self.assert_called('PUT', '/users/1') + self.assertRequestBodyIs(json={'user': {'id': '1', 'email': ''}}) def test_role_create(self): + self.stub_url('POST', ['OS-KSADM', 'roles'], json={'role': {}}) self.run_command('role-create --name new-role') - self.fake_client.assert_called_anytime( - 'POST', '/OS-KSADM/roles', - {"role": {"name": "new-role"}}) + self.assert_called('POST', '/OS-KSADM/roles') + self.assertRequestBodyIs(json={"role": {"name": "new-role"}}) def test_role_get(self): + self.stub_url('GET', ['OS-KSADM', 'roles', '1'], + json={'role': {'id': '1'}}) self.run_command('role-get 1') - self.fake_client.assert_called_anytime('GET', '/OS-KSADM/roles/1') + self.assert_called('GET', '/OS-KSADM/roles/1') def test_role_list(self): + self.stub_url('GET', ['OS-KSADM', 'roles'], json={'roles': []}) self.run_command('role-list') - self.fake_client.assert_called_anytime('GET', '/OS-KSADM/roles') + self.assert_called('GET', '/OS-KSADM/roles') def test_role_delete(self): + self.stub_url('GET', ['OS-KSADM', 'roles', '1'], + json={'role': {'id': '1'}}) + self.stub_url('DELETE', ['OS-KSADM', 'roles', '1']) self.run_command('role-delete 1') - self.fake_client.assert_called_anytime('DELETE', '/OS-KSADM/roles/1') + self.assert_called('DELETE', '/OS-KSADM/roles/1') def test_user_role_add(self): + self.stub_url('GET', ['users', '1'], + json={'user': {'id': '1'}}) + self.stub_url('GET', ['OS-KSADM', 'roles', '1'], + json={'role': {'id': '1'}}) + + self.stub_url('PUT', ['users', '1', 'roles', 'OS-KSADM', '1']) self.run_command('user-role-add --user_id 1 --role_id 1') - self.fake_client.assert_called_anytime( - 'PUT', '/users/1/roles/OS-KSADM/1') + self.assert_called('PUT', '/users/1/roles/OS-KSADM/1') def test_user_role_list(self): - self.run_command('user-role-list --user_id 1 --tenant-id 1') - self.fake_client.assert_called_anytime( - 'GET', '/tenants/1/users/1/roles') - self.run_command('user-role-list --user_id 1') - self.fake_client.assert_called_anytime( - 'GET', '/tenants/1/users/1/roles') + self.stub_url('GET', ['tenants', self.token.tenant_id], + json={'tenant': {'id': self.token.tenant_id}}) + self.stub_url('GET', ['tenants', self.token.tenant_id, + 'users', self.token.user_id, 'roles'], + json={'roles': []}) + + url = '/tenants/%s/users/%s/roles' % (self.token.tenant_id, + self.token.user_id) + + self.run_command('user-role-list --user_id %s --tenant-id %s' % + (self.token.user_id, self.token.tenant_id)) + self.assert_called('GET', url) + + self.run_command('user-role-list --user_id %s' % self.token.user_id) + self.assert_called('GET', url) + self.run_command('user-role-list') - self.fake_client.assert_called_anytime( - 'GET', '/tenants/1/users/1/roles') + self.assert_called('GET', url) def test_user_role_remove(self): + self.stub_url('GET', ['users', '1'], + json={'user': {'id': 1}}) + self.stub_url('GET', ['OS-KSADM', 'roles', '1'], + json={'role': {'id': 1}}) + self.stub_url('DELETE', + ['users', '1', 'roles', 'OS-KSADM', '1']) + self.run_command('user-role-remove --user_id 1 --role_id 1') - self.fake_client.assert_called_anytime( - 'DELETE', '/users/1/roles/OS-KSADM/1') + self.assert_called('DELETE', '/users/1/roles/OS-KSADM/1') def test_tenant_create(self): + self.stub_url('POST', ['tenants'], json={'tenant': {}}) self.run_command('tenant-create --name new-tenant') - self.fake_client.assert_called_anytime( - 'POST', '/tenants', - {"tenant": {"enabled": True, - "name": "new-tenant", - "description": None}}) + self.assertRequestBodyIs(json={"tenant": {"enabled": True, + "name": "new-tenant", + "description": None}}) def test_tenant_get(self): + self.stub_url('GET', ['tenants', '2'], json={'tenant': {}}) self.run_command('tenant-get 2') - self.fake_client.assert_called_anytime('GET', '/tenants/2') + self.assert_called('GET', '/tenants/2') def test_tenant_list(self): + self.stub_url('GET', ['tenants'], json={'tenants': []}) self.run_command('tenant-list') - self.fake_client.assert_called_anytime('GET', '/tenants') + self.assert_called('GET', '/tenants') def test_tenant_update(self): + self.stub_url('GET', ['tenants', '1'], + json={'tenant': {'id': '1'}}) + self.stub_url('GET', ['tenants', '2'], + json={'tenant': {'id': '2'}}) + self.stub_url('POST', ['tenants', '2'], + json={'tenant': {'id': '2'}}) self.run_command('tenant-update' ' --name new-tenant1 --enabled false' ' --description desc 2') - self.fake_client.assert_called_anytime( - 'POST', '/tenants/2', - {"tenant": - {"enabled": False, - "id": "2", - "description": "desc", - "name": "new-tenant1"}}) + self.assert_called('POST', '/tenants/2') + self.assertRequestBodyIs(json={"tenant": {"enabled": False, + "id": "2", + "description": "desc", + "name": "new-tenant1"}}) required = 'Tenant not updated, no arguments present.' out = self.run_command('tenant-update 1') self.assertThat(out, matchers.MatchesRegex(required)) def test_tenant_delete(self): + self.stub_url('GET', ['tenants', '2'], + json={'tenant': {'id': '2'}}) + self.stub_url('DELETE', ['tenants', '2']) self.run_command('tenant-delete 2') - self.fake_client.assert_called_anytime('DELETE', '/tenants/2') + self.assert_called('DELETE', '/tenants/2') def test_service_create(self): + self.stub_url('POST', ['OS-KSADM', 'services'], + json={'OS-KSADM:service': {}}) self.run_command('service-create --name service1 --type compute') - self.fake_client.assert_called_anytime( - 'POST', '/OS-KSADM/services', - {"OS-KSADM:service": - {"type": "compute", - "name": "service1", - "description": None}}) + self.assert_called('POST', '/OS-KSADM/services') + json = {"OS-KSADM:service": {"type": "compute", + "name": "service1", + "description": None}} + self.assertRequestBodyIs(json=json) def test_service_get(self): + self.stub_url('GET', ['OS-KSADM', 'services', '1'], + json={'OS-KSADM:service': {'id': '1'}}) self.run_command('service-get 1') - self.fake_client.assert_called_anytime('GET', '/OS-KSADM/services/1') + self.assert_called('GET', '/OS-KSADM/services/1') def test_service_list(self): + self.stub_url('GET', ['OS-KSADM', 'services'], + json={'OS-KSADM:services': []}) self.run_command('service-list') - self.fake_client.assert_called_anytime('GET', '/OS-KSADM/services') + self.assert_called('GET', '/OS-KSADM/services') def test_service_delete(self): + self.stub_url('GET', ['OS-KSADM', 'services', '1'], + json={'OS-KSADM:service': {'id': 1}}) + self.stub_url('DELETE', ['OS-KSADM', 'services', '1']) self.run_command('service-delete 1') - self.fake_client.assert_called_anytime( - 'DELETE', '/OS-KSADM/services/1') + self.assert_called('DELETE', '/OS-KSADM/services/1') def test_catalog(self): self.run_command('catalog') self.run_command('catalog --service compute') def test_ec2_credentials_create(self): - self.run_command('ec2-credentials-create' - ' --tenant-id 1 --user-id 1') - self.fake_client.assert_called_anytime( - 'POST', '/users/1/credentials/OS-EC2', - {'tenant_id': '1'}) + self.stub_url('POST', + ['users', self.token.user_id, 'credentials', 'OS-EC2'], + json={'credential': {}}) + + url = '/users/%s/credentials/OS-EC2' % self.token.user_id + self.run_command('ec2-credentials-create --tenant-id 1 ' + '--user-id %s' % self.token.user_id) + self.assert_called('POST', url) + self.assertRequestBodyIs(json={'tenant_id': '1'}) self.run_command('ec2-credentials-create --tenant-id 1') - self.fake_client.assert_called_anytime( - 'POST', '/users/1/credentials/OS-EC2', - {'tenant_id': '1'}) + self.assert_called('POST', url) + self.assertRequestBodyIs(json={'tenant_id': '1'}) self.run_command('ec2-credentials-create') - self.fake_client.assert_called_anytime( - 'POST', '/users/1/credentials/OS-EC2', - {'tenant_id': '1'}) + self.assert_called('POST', url) + self.assertRequestBodyIs(json={'tenant_id': self.token.tenant_id}) def test_ec2_credentials_delete(self): - self.run_command('ec2-credentials-delete --access 2 --user-id 1') - self.fake_client.assert_called_anytime( - 'DELETE', '/users/1/credentials/OS-EC2/2') + self.stub_url('DELETE', + ['users', self.token.user_id, + 'credentials', 'OS-EC2', '2']) + self.run_command('ec2-credentials-delete --access 2 --user-id %s' % + self.token.user_id) + + url = '/users/%s/credentials/OS-EC2/2' % self.token.user_id + self.assert_called('DELETE', url) self.run_command('ec2-credentials-delete --access 2') - self.fake_client.assert_called_anytime( - 'DELETE', '/users/1/credentials/OS-EC2/2') + self.assert_called('DELETE', url) def test_ec2_credentials_list(self): - self.run_command('ec2-credentials-list --user-id 1') - self.fake_client.assert_called_anytime( - 'GET', '/users/1/credentials/OS-EC2') + self.stub_url('GET', + ['users', self.token.user_id, 'credentials', 'OS-EC2'], + json={'credentials': []}) + self.run_command('ec2-credentials-list --user-id %s' + % self.token.user_id) + + url = '/users/%s/credentials/OS-EC2' % self.token.user_id + self.assert_called('GET', url) self.run_command('ec2-credentials-list') - self.fake_client.assert_called_anytime( - 'GET', '/users/1/credentials/OS-EC2') + self.assert_called('GET', url) def test_ec2_credentials_get(self): + self.stub_url('GET', + ['users', '1', 'credentials', 'OS-EC2', '2'], + json={'credential': {}}) self.run_command('ec2-credentials-get --access 2 --user-id 1') - self.fake_client.assert_called_anytime( - 'GET', '/users/1/credentials/OS-EC2/2') + self.assert_called('GET', '/users/1/credentials/OS-EC2/2') def test_bootstrap(self): - self.run_command('bootstrap --user-name new-user' - ' --pass 1 --role-name admin' - ' --tenant-name new-tenant') - self.fake_client.assert_called_anytime( - 'POST', '/users', - {'user': - {'email': None, - 'password': '1', - 'enabled': True, - 'name': 'new-user', - 'tenantId': None}}) - self.run_command('bootstrap --user-name new-user' - ' --pass 1 --role-name admin' - ' --tenant-name new-tenant') - self.fake_client.assert_called_anytime( - 'POST', '/tenants', - {"tenant": {"enabled": True, - "name": "new-tenant", - "description": None}}) - self.run_command('bootstrap --user-name new-user' - ' --pass 1 --role-name new-role' - ' --tenant-name new-tenant') - self.fake_client.assert_called_anytime( - 'POST', '/OS-KSADM/roles', - {"role": {"name": "new-role"}}) + user = {'user': {'id': '1'}} + role = {'role': {'id': '1'}} + tenant = {'tenant': {'id': '1'}} - self.run_command('bootstrap --user-name' - ' new-user --pass 1 --role-name admin' + token = fixture.V2Token(user_id=1, tenant_id=1) + token.add_role(id=1) + svc = token.add_service('identity') + svc.add_endpoint(public=DEFAULT_AUTH_URL, + admin=DEFAULT_ADMIN_URL) + + self.stub_auth(json=token) + + self.stub_url('POST', ['OS-KSADM', 'roles'], json=role) + self.stub_url('GET', ['OS-KSADM', 'roles', '1'], json=role) + self.stub_url('POST', ['tenants'], json=tenant) + self.stub_url('GET', ['tenants', '1'], json=tenant) + self.stub_url('POST', ['users'], json=user) + self.stub_url('GET', ['users', '1'], json=user) + self.stub_url('PUT', + ['tenants', '1', 'users', '1', 'roles', 'OS-KSADM', '1'], + json=role) + + self.run_command('bootstrap --user-name new-user' + ' --pass 1 --role-name admin' ' --tenant-name new-tenant') - self.fake_client.assert_called_anytime( - 'PUT', '/tenants/1/users/1/roles/OS-KSADM/1') + + def called_anytime(method, path, json=None): + for r in self.requests.request_history: + if not r.method == method: + continue + if not r.url == self.TEST_URL + path: + continue + + if json: + last_request_body = r.body.decode('utf-8') + json_body = jsonutils.loads(last_request_body) + + if not json_body == json: + continue + + return True + + return False + + called_anytime('POST', '/users', {'user': {'email': None, + 'password': '1', + 'enabled': True, + 'name': 'new-user', + 'tenantId': None}}) + + called_anytime('POST', '/tenants', {"tenant": {"enabled": True, + "name": "new-tenant", + "description": None}}) + + called_anytime('POST', '/OS-KSADM/roles', + {"role": {"name": "new-role"}}) + + called_anytime('PUT', '/tenants/1/users/1/roles/OS-KSADM/1') def test_bash_completion(self): self.run_command('bash-completion') @@ -336,26 +421,32 @@ class ShellTests(utils.TestCase): self.assertThat(out, matchers.MatchesRegex(required)) def test_password_update(self): + self.stub_url('PATCH', + ['OS-KSCRUD', 'users', self.token.user_id], + base_url=DEFAULT_AUTH_URL) self.run_command('password-update --current-password oldpass' ' --new-password newpass') - self.fake_client.assert_called_anytime( - 'PATCH', '/OS-KSCRUD/users/1', - {'user': - {'original_password': 'oldpass', - 'password': 'newpass'}}) + self.assert_called('PATCH', + '/OS-KSCRUD/users/%s' % self.token.user_id, + base_url=DEFAULT_AUTH_URL) + self.assertRequestBodyIs(json={'user': {'original_password': 'oldpass', + 'password': 'newpass'}}) def test_endpoint_create(self): + self.stub_url('GET', ['OS-KSADM', 'services', '1'], + json={'OS-KSADM:service': {'id': '1'}}) + self.stub_url('POST', ['endpoints'], json={'endpoint': {}}) self.run_command('endpoint-create --service-id 1 ' '--publicurl=http://example.com:1234/go') - self.fake_client.assert_called_anytime( - 'POST', '/endpoints', - {'endpoint': - {'adminurl': None, - 'service_id': '1', - 'region': 'regionOne', - 'internalurl': None, - 'publicurl': "http://example.com:1234/go"}}) + self.assert_called('POST', '/endpoints') + json = {'endpoint': {'adminurl': None, + 'service_id': '1', + 'region': 'regionOne', + 'internalurl': None, + 'publicurl': "http://example.com:1234/go"}} + self.assertRequestBodyIs(json=json) def test_endpoint_list(self): + self.stub_url('GET', ['endpoints'], json={'endpoints': []}) self.run_command('endpoint-list') - self.fake_client.assert_called_anytime('GET', '/endpoints') + self.assert_called('GET', '/endpoints')