Convert shell tests to requests-mock

This finally gets us rid of the test fakes and will remove the
dependency on _cs_request.

Change-Id: Ibd09db9a09c49bfbcf0ffe1f150d7fe730dd51b7
This commit is contained in:
Jamie Lennox
2014-07-29 15:52:00 +10:00
parent 7825e99e36
commit e58b8285bf
3 changed files with 256 additions and 780 deletions

View File

@@ -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"
}
}
})

View File

@@ -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)

View File

@@ -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,22 +78,22 @@ 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,
self.assert_called('POST', '/users')
self.assertRequestBodyIs(json={'user': {'email': None,
'password': None,
'enabled': True,
'name': 'new-user',
@@ -105,117 +101,157 @@ class ShellTests(utils.TestCase):
@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,
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',
self.assert_called('PUT', '/users/1')
self.assertRequestBodyIs(json={'user': {'id': '1',
'email': 'user@email.com',
'enabled': True,
'name': 'new-user1'}
})
'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,
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,
self.assert_called('POST', '/tenants/2')
self.assertRequestBodyIs(json={"tenant": {"enabled": False,
"id": "2",
"description": "desc",
"name": "new-tenant1"}})
@@ -225,107 +261,156 @@ class ShellTests(utils.TestCase):
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",
self.assert_called('POST', '/OS-KSADM/services')
json = {"OS-KSADM:service": {"type": "compute",
"name": "service1",
"description": None}})
"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):
user = {'user': {'id': '1'}}
role = {'role': {'id': '1'}}
tenant = {'tenant': {'id': '1'}}
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(
'POST', '/users',
{'user':
{'email': None,
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}})
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,
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',
called_anytime('POST', '/OS-KSADM/roles',
{"role": {"name": "new-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')
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',
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,
self.assert_called('POST', '/endpoints')
json = {'endpoint': {'adminurl': None,
'service_id': '1',
'region': 'regionOne',
'internalurl': None,
'publicurl': "http://example.com:1234/go"}})
'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')