Remove keystoneclient mocks in test_caching for users
Remove the direct keystoneclient mocks in test_caching for user cases. This implements "register_uris" which takes a list of dictionaries describing the mocked URIs for ease of reading when dealing with highly complex request mocking such as the test_modify_user_invalidates_cache test case. Change-Id: I75f62a7b2c6d0838dd4a4bdf10cb74e4baf4aa7e
This commit is contained in:
Morgan Fainberg
parent
aff1f8f8b6
commit
ea83e73be0
@ -132,6 +132,7 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
self.discovery_json = os.path.join(
|
||||
self.fixtures_directory, 'discovery.json')
|
||||
self.use_keystone_v3()
|
||||
self.__register_uris_called = False
|
||||
|
||||
def get_mock_url(self, service_type, interface, resource=None,
|
||||
append=None, base_url_append=None):
|
||||
@ -327,6 +328,60 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
'GET', 'https://image.example.com/',
|
||||
text=open(discovery_fixture, 'r').read())
|
||||
|
||||
def register_uris(self, uri_mock_list):
|
||||
"""Mock a list of URIs and responses via requests mock.
|
||||
|
||||
This method may be called only once per test-case to avoid odd
|
||||
and difficult to debug interactions. Discovery and Auth request mocking
|
||||
happens separately from this method.
|
||||
|
||||
:param uri_mock_list: List of dictionaries that template out what is
|
||||
passed to requests_mock fixture's `register_uri`.
|
||||
Format is:
|
||||
{'method': <HTTP_METHOD>,
|
||||
'uri': <URI to be mocked>,
|
||||
...
|
||||
}
|
||||
|
||||
Common keys to pass in the dictionary:
|
||||
* json: the json response (dict)
|
||||
* status_code: the HTTP status (int)
|
||||
* validate: The request body (dict) to
|
||||
validate with assert_calls
|
||||
all key-word arguments that are valid to send to
|
||||
requests_mock are supported.
|
||||
|
||||
This list should be in the order in which calls
|
||||
are made. When `assert_calls` is executed, order
|
||||
here will be validated. Duplicate URIs and
|
||||
Methods are allowed and will be collapsed into a
|
||||
single matcher. Each response will be returned
|
||||
in order as the URI+Method is hit.
|
||||
:return: None
|
||||
"""
|
||||
assert not self.__register_uris_called
|
||||
for to_mock in uri_mock_list:
|
||||
method = to_mock.pop('method')
|
||||
uri = to_mock.pop('uri')
|
||||
key = '{method}:{uri}'.format(method=method, uri=uri)
|
||||
headers = structures.CaseInsensitiveDict(to_mock.pop('headers',
|
||||
{}))
|
||||
validate = to_mock.pop('validate', {})
|
||||
if 'content-type' not in headers:
|
||||
headers[u'content-type'] = 'application/json'
|
||||
|
||||
self.calls += [
|
||||
dict(
|
||||
method=method,
|
||||
url=uri, **validate)
|
||||
]
|
||||
self._uri_registry.setdefault(key, []).append(to_mock)
|
||||
|
||||
for mock_method_uri, params in self._uri_registry.items():
|
||||
mock_method, mock_uri = mock_method_uri.split(':', 1)
|
||||
self.adapter.register_uri(mock_method, mock_uri, params)
|
||||
self.__register_uris_called = True
|
||||
|
||||
def register_uri(self, method, uri, **kwargs):
|
||||
validate = kwargs.pop('validate', {})
|
||||
key = '{method}:{uri}'.format(method=method, uri=uri)
|
||||
|
||||
@ -293,61 +293,121 @@ class TestMemoryCache(base.RequestsMockTestCase):
|
||||
self.cloud.delete_volume('12345')
|
||||
self.assertEqual([fake_volb4_dict], self.cloud.list_volumes())
|
||||
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_list_users(self, keystone_mock):
|
||||
fake_user = fakes.FakeUser('999', '', '')
|
||||
keystone_mock.users.list.return_value = [fake_user]
|
||||
def test_list_users(self):
|
||||
self._add_discovery_uri_call()
|
||||
user_data = self._get_user_data(email='test@example.com')
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
service_type='identity',
|
||||
interface='admin',
|
||||
resource='users',
|
||||
base_url_append='v3'),
|
||||
status_code=200,
|
||||
json={'users': [user_data.json_response['user']]})])
|
||||
users = self.cloud.list_users()
|
||||
self.assertEqual(1, len(users))
|
||||
self.assertEqual('999', users[0]['id'])
|
||||
self.assertEqual('', users[0]['name'])
|
||||
self.assertEqual('', users[0]['email'])
|
||||
self.assertEqual(user_data.user_id, users[0]['id'])
|
||||
self.assertEqual(user_data.name, users[0]['name'])
|
||||
self.assertEqual(user_data.email, users[0]['email'])
|
||||
self.assert_calls()
|
||||
|
||||
@mock.patch.object(shade.OpenStackCloud, 'keystone_client')
|
||||
def test_modify_user_invalidates_cache(self, keystone_mock):
|
||||
def test_modify_user_invalidates_cache(self):
|
||||
self.use_keystone_v2()
|
||||
fake_user = fakes.FakeUser('abc123', 'abc123@domain.test',
|
||||
'abc123 name')
|
||||
# first cache an empty list
|
||||
keystone_mock.users.list.return_value = []
|
||||
self.assertEqual([], self.cloud.list_users())
|
||||
# now add one
|
||||
keystone_mock.users.list.return_value = [fake_user]
|
||||
keystone_mock.users.create.return_value = fake_user
|
||||
created = self.cloud.create_user(name='abc123 name',
|
||||
email='abc123@domain.test')
|
||||
self.assertEqual('abc123', created['id'])
|
||||
self.assertEqual('abc123 name', created['name'])
|
||||
self.assertEqual('abc123@domain.test', created['email'])
|
||||
|
||||
user_data = self._get_user_data(email='test@example.com')
|
||||
new_resp = {'user': user_data.json_response['user'].copy()}
|
||||
new_resp['user']['email'] = 'Nope@Nope.Nope'
|
||||
new_req = {'user': {'email': new_resp['user']['email']}}
|
||||
|
||||
mock_users_url = self.get_mock_url(
|
||||
service_type='identity',
|
||||
interface='admin',
|
||||
resource='users')
|
||||
mock_user_resource_url = self.get_mock_url(
|
||||
service_type='identity',
|
||||
interface='admin',
|
||||
resource='users',
|
||||
append=[user_data.user_id])
|
||||
|
||||
empty_user_list_resp = {'users': []}
|
||||
users_list_resp = {'users': [user_data.json_response['user']]}
|
||||
updated_users_list_resp = {'users': [new_resp['user']]}
|
||||
|
||||
uris_to_mock = [
|
||||
# Inital User List is Empty
|
||||
dict(method='GET', uri=mock_users_url, status_code=200,
|
||||
json=empty_user_list_resp),
|
||||
# POST to create the user
|
||||
# GET to get the user data after POST
|
||||
dict(method='POST', uri=mock_users_url, status_code=200,
|
||||
json=user_data.json_response,
|
||||
validate=user_data.json_request),
|
||||
dict(method='GET', uri=mock_user_resource_url, status_code=200,
|
||||
json=user_data.json_response),
|
||||
# List Users Call
|
||||
dict(method='GET', uri=mock_users_url, status_code=200,
|
||||
json=users_list_resp),
|
||||
# List users to get ID for update
|
||||
# Get user using user_id from list
|
||||
# Update user
|
||||
# Get updated user
|
||||
dict(method='GET', uri=mock_users_url, status_code=200,
|
||||
json=users_list_resp),
|
||||
dict(method='GET', uri=mock_user_resource_url, status_code=200,
|
||||
json=user_data.json_response),
|
||||
dict(method='PUT', uri=mock_user_resource_url, status_code=200,
|
||||
json=new_resp, validate=new_req),
|
||||
dict(method='GET', uri=mock_user_resource_url, status_code=200,
|
||||
json=new_resp),
|
||||
# List Users Call
|
||||
dict(method='GET', uri=mock_users_url, status_code=200,
|
||||
json=updated_users_list_resp),
|
||||
# List User to get ID for delete
|
||||
# Get user using user_id from list
|
||||
# delete user
|
||||
dict(method='GET', uri=mock_users_url, status_code=200,
|
||||
json=updated_users_list_resp),
|
||||
dict(method='GET', uri=mock_user_resource_url, status_code=200,
|
||||
json=new_resp),
|
||||
dict(method='DELETE', uri=mock_user_resource_url, status_code=204),
|
||||
# List Users Call (empty post delete)
|
||||
dict(method='GET', uri=mock_users_url, status_code=200,
|
||||
json=empty_user_list_resp)
|
||||
]
|
||||
|
||||
self.register_uris(uris_to_mock)
|
||||
|
||||
# first cache an empty list
|
||||
self.assertEqual([], self.cloud.list_users())
|
||||
|
||||
# now add one
|
||||
created = self.cloud.create_user(name=user_data.name,
|
||||
email=user_data.email)
|
||||
self.assertEqual(user_data.user_id, created['id'])
|
||||
self.assertEqual(user_data.name, created['name'])
|
||||
self.assertEqual(user_data.email, created['email'])
|
||||
# Cache should have been invalidated
|
||||
users = self.cloud.list_users()
|
||||
self.assertEqual(1, len(users))
|
||||
self.assertEqual('abc123', users[0]['id'])
|
||||
self.assertEqual('abc123 name', users[0]['name'])
|
||||
self.assertEqual('abc123@domain.test', users[0]['email'])
|
||||
self.assertEqual(user_data.user_id, users[0]['id'])
|
||||
self.assertEqual(user_data.name, users[0]['name'])
|
||||
self.assertEqual(user_data.email, users[0]['email'])
|
||||
|
||||
# Update and check to see if it is updated
|
||||
fake_user2 = fakes.FakeUser('abc123',
|
||||
'abc123-changed@domain.test',
|
||||
'abc123 name')
|
||||
fake_user2_dict = meta.obj_to_dict(fake_user2)
|
||||
keystone_mock.users.update.return_value = fake_user2
|
||||
keystone_mock.users.list.return_value = [fake_user2]
|
||||
keystone_mock.users.get.return_value = fake_user2_dict
|
||||
self.cloud.update_user('abc123', email='abc123-changed@domain.test')
|
||||
keystone_mock.users.update.assert_called_with(
|
||||
user=fake_user2_dict, email='abc123-changed@domain.test')
|
||||
updated = self.cloud.update_user(user_data.user_id,
|
||||
email=new_resp['user']['email'])
|
||||
self.assertEqual(user_data.user_id, updated.id)
|
||||
self.assertEqual(user_data.name, updated.name)
|
||||
self.assertEqual(new_resp['user']['email'], updated.email)
|
||||
users = self.cloud.list_users()
|
||||
self.assertEqual(1, len(users))
|
||||
self.assertEqual('abc123', users[0]['id'])
|
||||
self.assertEqual('abc123 name', users[0]['name'])
|
||||
self.assertEqual('abc123-changed@domain.test', users[0]['email'])
|
||||
self.assertEqual(user_data.user_id, users[0]['id'])
|
||||
self.assertEqual(user_data.name, users[0]['name'])
|
||||
self.assertEqual(new_resp['user']['email'], users[0]['email'])
|
||||
# Now delete and ensure it disappears
|
||||
keystone_mock.users.list.return_value = []
|
||||
self.cloud.delete_user('abc123')
|
||||
self.cloud.delete_user(user_data.user_id)
|
||||
self.assertEqual([], self.cloud.list_users())
|
||||
self.assertTrue(keystone_mock.users.delete.was_called)
|
||||
self.assert_calls()
|
||||
|
||||
def test_list_flavors(self):
|
||||
self.register_uri(
|
||||
|
||||
Loading…
Reference in New Issue
Block a user