Convert use of .register_uri to .register_uris

This patch converts all use of .register_uri to .register_uris for
the following files: test_caching, test_create_server, test_domains,
test_flavors, test_floating_ip_neutron, test_users.

Change-Id: I48d90465f22bfa27e9de552e135f42967a9e21fa
This commit is contained in:
Morgan Fainberg 2017-02-13 14:59:49 -08:00
parent ea83e73be0
commit 820885c1a1
7 changed files with 660 additions and 694 deletions

View File

@ -364,12 +364,14 @@ class RequestsMockTestCase(BaseTestCase):
method = to_mock.pop('method')
uri = to_mock.pop('uri')
key = '{method}:{uri}'.format(method=method, uri=uri)
validate = to_mock.pop('validate', {})
headers = structures.CaseInsensitiveDict(to_mock.pop('headers',
{}))
validate = to_mock.pop('validate', {})
if 'content-type' not in headers:
headers[u'content-type'] = 'application/json'
to_mock['headers'] = headers
self.calls += [
dict(
method=method,

View File

@ -116,24 +116,16 @@ class TestMemoryCache(base.RequestsMockTestCase):
second_response = {'projects': [p.json_response['project']
for p in project_list]}
self.register_uri(
'GET',
self.get_mock_url(
service_type='identity',
interface='admin',
resource='projects',
base_url_append='v3'),
status_code=200,
json=first_response)
self.register_uri(
'GET',
self.get_mock_url(
service_type='identity',
interface='admin',
resource='projects',
base_url_append='v3'),
status_code=200,
json=second_response)
mock_uri = self.get_mock_url(
service_type='identity', interface='admin', resource='projects',
base_url_append='v3')
self.register_uris([
dict(method='GET', uri=mock_uri, status_code=200,
json=first_response),
dict(method='GET', uri=mock_uri, status_code=200,
json=second_response)])
self.assertEqual(
self.cloud._normalize_projects(
meta.obj_list_to_dict(first_response['projects'])),
@ -161,22 +153,15 @@ class TestMemoryCache(base.RequestsMockTestCase):
second_response = {'tenants': [p.json_response['tenant']
for p in project_list]}
self.register_uri(
'GET',
self.get_mock_url(
service_type='identity',
interface='admin',
resource='tenants'),
status_code=200,
json=first_response)
self.register_uri(
'GET',
self.get_mock_url(
service_type='identity',
interface='admin',
resource='tenants'),
status_code=200,
json=second_response)
mock_uri = self.get_mock_url(
service_type='identity', interface='admin', resource='tenants')
self.register_uris([
dict(method='GET', uri=mock_uri, status_code=200,
json=first_response),
dict(method='GET', uri=mock_uri, status_code=200,
json=second_response)])
self.assertEqual(
self.cloud._normalize_projects(
meta.obj_list_to_dict(first_response['tenants'])),
@ -410,20 +395,22 @@ class TestMemoryCache(base.RequestsMockTestCase):
self.assert_calls()
def test_list_flavors(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': []})
mock_uri = '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT)
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})
for flavor in fakes.FAKE_FLAVOR_LIST:
self.register_uri(
'GET', '{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=flavor['id']),
json={'extra_specs': {}})
uris_to_mock = [
dict(method='GET', uri=mock_uri, json={'flavors': []}),
dict(method='GET', uri=mock_uri,
json={'flavors': fakes.FAKE_FLAVOR_LIST})
]
uris_to_mock.extend([
dict(method='GET',
uri='{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=flavor['id']),
json={'extra_specs': {}})
for flavor in fakes.FAKE_FLAVOR_LIST])
self.register_uris(uris_to_mock)
self.assertEqual([], self.cloud.list_flavors())

View File

@ -313,10 +313,13 @@ class TestCreateServer(base.RequestsMockTestCase):
@mock.patch.object(OpenStackCloud, 'nova_client')
def test_create_server_get_flavor_image(
self, mock_nova, mock_image, mock_get_server_by_id):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})
self.register_uris([
dict(method='GET',
uri='{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})])
self.cloud.create_server(
'server-name', 'image-id', 'vanilla',
nics=[{'net-id': 'some-network'}])

View File

@ -66,9 +66,9 @@ class TestDomains(base.RequestsMockTestCase):
def test_list_domains(self):
self._add_discovery_uri_call()
domain_data = self._get_domain_data()
self.register_uri(
'GET', self.get_mock_url(), status_code=200,
json={'domains': [domain_data.json_response['domain']]})
self.register_uris([
dict(method='GET', uri=self.get_mock_url(), status_code=200,
json={'domains': [domain_data.json_response['domain']]})])
domains = self.op_cloud.list_domains()
self.assertThat(len(domains), matchers.Equals(1))
self.assertThat(domains[0].name,
@ -80,10 +80,11 @@ class TestDomains(base.RequestsMockTestCase):
def test_get_domain(self):
self._add_discovery_uri_call()
domain_data = self._get_domain_data()
self.register_uri(
'GET', self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=domain_data.json_response)
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=domain_data.json_response)])
domain = self.op_cloud.get_domain(domain_id=domain_data.domain_id)
self.assertThat(domain.id, matchers.Equals(domain_data.domain_id))
self.assertThat(domain.name, matchers.Equals(domain_data.domain_name))
@ -92,12 +93,12 @@ class TestDomains(base.RequestsMockTestCase):
def test_get_domain_with_name_or_id(self):
self._add_discovery_uri_call()
domain_data = self._get_domain_data()
self.register_uri(
'GET', self.get_mock_url(), status_code=200,
json={'domains': [domain_data.json_response['domain']]})
self.register_uri(
'GET', self.get_mock_url(), status_code=200,
json={'domains': [domain_data.json_response['domain']]})
response = {'domains': [domain_data.json_response['domain']]}
self.register_uris([
dict(method='GET', uri=self.get_mock_url(), status_code=200,
json=response),
dict(method='GET', uri=self.get_mock_url(), status_code=200,
json=response)])
domain = self.op_cloud.get_domain(name_or_id=domain_data.domain_id)
domain_by_name = self.op_cloud.get_domain(
name_or_id=domain_data.domain_name)
@ -113,17 +114,14 @@ class TestDomains(base.RequestsMockTestCase):
self._add_discovery_uri_call()
domain_data = self._get_domain_data(description=uuid.uuid4().hex,
enabled=True)
self.register_uri(
'POST',
self.get_mock_url(),
status_code=200,
json=domain_data.json_response,
validate=dict(json=domain_data.json_request))
self.register_uri(
'GET',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=domain_data.json_response)
self.register_uris([
dict(method='POST', uri=self.get_mock_url(), status_code=200,
json=domain_data.json_response,
validate=dict(json=domain_data.json_request)),
dict(method='GET',
uri=self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=domain_data.json_request)])
domain = self.op_cloud.create_domain(
domain_data.domain_name, domain_data.description)
self.assertThat(domain.id, matchers.Equals(domain_data.domain_id))
@ -138,10 +136,8 @@ class TestDomains(base.RequestsMockTestCase):
shade.OpenStackCloudException,
"Failed to create domain domain_name"
):
self.register_uri(
'POST',
self.get_mock_url(),
status_code=409)
self.register_uris([
dict(method='POST', uri=self.get_mock_url(), status_code=409)])
self.op_cloud.create_domain('domain_name')
self.assert_calls()
@ -150,21 +146,13 @@ class TestDomains(base.RequestsMockTestCase):
domain_data = self._get_domain_data()
new_resp = domain_data.json_response.copy()
new_resp['domain']['enabled'] = False
self.register_uri(
'PATCH',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=new_resp,
validate={'domain': {'enabled': False}})
self.register_uri(
'GET',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=new_resp)
self.register_uri(
'DELETE',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=204)
domain_resource_uri = self.get_mock_url(append=[domain_data.domain_id])
self.register_uris([
dict(method='PATCH', uri=domain_resource_uri, status_code=200,
json=new_resp, validate={'domain': {'enabled': False}}),
dict(method='GET', uri=domain_resource_uri, status_code=200,
json=new_resp),
dict(method='DELETE', uri=domain_resource_uri, status_code=204)])
self.op_cloud.delete_domain(domain_data.domain_id)
self.assert_calls()
@ -173,26 +161,16 @@ class TestDomains(base.RequestsMockTestCase):
domain_data = self._get_domain_data()
new_resp = domain_data.json_response.copy()
new_resp['domain']['enabled'] = False
self.register_uri(
'GET',
self.get_mock_url(),
status_code=200,
json={'domains': [domain_data.json_response['domain']]})
self.register_uri(
'PATCH',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=new_resp,
validate={'domain': {'enabled': False}})
self.register_uri(
'GET',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=new_resp)
self.register_uri(
'DELETE',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=204)
domain_resource_uri = self.get_mock_url(append=[domain_data.domain_id])
self.register_uris([
dict(method='GET', uri=self.get_mock_url(), status_code=200,
json={'domains': [domain_data.json_response['domain']]}),
dict(method='PATCH', uri=domain_resource_uri, status_code=200,
json=new_resp, validate={'domain': {'enabled': False}}),
dict(method='GET', uri=domain_resource_uri, status_code=200,
json=new_resp),
dict(method='DELETE', uri=domain_resource_uri, status_code=204)])
self.op_cloud.delete_domain(name_or_id=domain_data.domain_id)
self.assert_calls()
@ -206,21 +184,13 @@ class TestDomains(base.RequestsMockTestCase):
domain_data = self._get_domain_data()
new_resp = domain_data.json_response.copy()
new_resp['domain']['enabled'] = False
self.register_uri(
'PATCH',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=new_resp,
validate={'domain': {'enabled': False}})
self.register_uri(
'GET',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=new_resp)
self.register_uri(
'DELETE',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=404)
domain_resource_uri = self.get_mock_url(append=[domain_data.domain_id])
self.register_uris([
dict(method='PATCH', uri=domain_resource_uri, status_code=200,
json=new_resp, validate={'domain': {'enabled': False}}),
dict(method='GET', uri=domain_resource_uri, status_code=200,
json=new_resp),
dict(method='DELETE', uri=domain_resource_uri, status_code=404)])
with testtools.ExpectedException(
shade.OpenStackCloudException,
"Failed to delete domain %s" % domain_data.domain_id
@ -232,17 +202,13 @@ class TestDomains(base.RequestsMockTestCase):
self._add_discovery_uri_call()
domain_data = self._get_domain_data(
description=self.getUniqueString('domainDesc'))
self.register_uri(
'PATCH',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=domain_data.json_response,
validate=dict(json=domain_data.json_request))
self.register_uri(
'GET',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=domain_data.json_response)
domain_resource_uri = self.get_mock_url(append=[domain_data.domain_id])
self.register_uris([
dict(method='PATCH', uri=domain_resource_uri, status_code=200,
json=domain_data.json_response,
validate=dict(json=domain_data.json_request)),
dict(method='GET', uri=domain_resource_uri, status_code=200,
json=domain_data.json_response)])
domain = self.op_cloud.update_domain(
domain_data.domain_id,
name=domain_data.domain_name,
@ -257,21 +223,15 @@ class TestDomains(base.RequestsMockTestCase):
self._add_discovery_uri_call()
domain_data = self._get_domain_data(
description=self.getUniqueString('domainDesc'))
self.register_uri(
'GET',
self.get_mock_url(), status_code=200,
json={'domains': [domain_data.json_response['domain']]})
self.register_uri(
'PATCH',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=domain_data.json_response,
validate=dict(json=domain_data.json_request))
self.register_uri(
'GET',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=200,
json=domain_data.json_response)
domain_resource_uri = self.get_mock_url(append=[domain_data.domain_id])
self.register_uris([
dict(method='GET', uri=self.get_mock_url(), status_code=200,
json={'domains': [domain_data.json_response['domain']]}),
dict(method='PATCH', uri=domain_resource_uri, status_code=200,
json=domain_data.json_response,
validate=dict(json=domain_data.json_request)),
dict(method='GET', uri=domain_resource_uri, status_code=200,
json=domain_data.json_response)])
domain = self.op_cloud.update_domain(
name_or_id=domain_data.domain_id,
name=domain_data.domain_name,
@ -286,10 +246,10 @@ class TestDomains(base.RequestsMockTestCase):
self._add_discovery_uri_call()
domain_data = self._get_domain_data(
description=self.getUniqueString('domainDesc'))
self.register_uri(
'PATCH',
self.get_mock_url(append=[domain_data.domain_id]),
status_code=409)
self.register_uris([
dict(method='PATCH',
uri=self.get_mock_url(append=[domain_data.domain_id]),
status_code=409)])
with testtools.ExpectedException(
shade.OpenStackCloudException,
"Error in updating domain %s" % domain_data.domain_id

View File

@ -20,22 +20,23 @@ class TestFlavors(base.RequestsMockTestCase):
def test_create_flavor(self):
self.register_uri(
'POST', '{endpoint}/flavors'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavor': fakes.FAKE_FLAVOR},
validate=dict(
json={'flavor': {
"name": "vanilla",
"ram": 65536,
"vcpus": 24,
"swap": 0,
"os-flavor-access:is_public": True,
"rxtx_factor": 1.0,
"OS-FLV-EXT-DATA:ephemeral": 0,
"disk": 1600,
"id": None
}}))
self.register_uris([
dict(method='POST',
uri='{endpoint}/flavors'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavor': fakes.FAKE_FLAVOR},
validate=dict(
json={
'flavor': {
"name": "vanilla",
"ram": 65536,
"vcpus": 24,
"swap": 0,
"os-flavor-access:is_public": True,
"rxtx_factor": 1.0,
"OS-FLV-EXT-DATA:ephemeral": 0,
"disk": 1600,
"id": None}}))])
self.op_cloud.create_flavor(
'vanilla', ram=65536, disk=1600, vcpus=24,
@ -43,49 +44,57 @@ class TestFlavors(base.RequestsMockTestCase):
self.assert_calls()
def test_delete_flavor(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})
self.register_uri(
'DELETE', '{endpoint}/flavors/{id}'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=fakes.FLAVOR_ID))
self.register_uris([
dict(method='GET',
uri='{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST}),
dict(method='DELETE',
uri='{endpoint}/flavors/{id}'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=fakes.FLAVOR_ID))])
self.assertTrue(self.op_cloud.delete_flavor('vanilla'))
self.assert_calls()
def test_delete_flavor_not_found(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})
self.register_uris([
dict(method='GET',
uri='{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})])
self.assertFalse(self.op_cloud.delete_flavor('invalid'))
self.assert_calls()
def test_delete_flavor_exception(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})
self.register_uri(
'DELETE', '{endpoint}/flavors/{id}'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=fakes.FLAVOR_ID),
status_code=503)
self.register_uris([
dict(method='GET',
uri='{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST}),
dict(method='DELETE',
uri='{endpoint}/flavors/{id}'.format(
endpoint=fakes.FAKE_FLAVOR_LIST, id=fakes.FLAVOR_ID),
status_code=503)])
self.assertRaises(shade.OpenStackCloudException,
self.op_cloud.delete_flavor, 'vanilla')
def test_list_flavors(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})
for flavor in fakes.FAKE_FLAVOR_LIST:
self.register_uri(
'GET', '{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=flavor['id']),
json={'extra_specs': {}})
uris_to_mock = [
dict(method='GET',
uri='{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST}),
]
uris_to_mock.extend([
dict(method='GET',
uri='{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=flavor['id']),
json={'extra_specs': {}})
for flavor in fakes.FAKE_FLAVOR_LIST])
self.register_uris(uris_to_mock)
flavors = self.cloud.list_flavors()
@ -103,59 +112,65 @@ class TestFlavors(base.RequestsMockTestCase):
self.assert_calls()
def test_get_flavor_by_ram(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})
for flavor in fakes.FAKE_FLAVOR_LIST:
self.register_uri(
'GET', '{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=flavor['id']),
json={'extra_specs': {}})
uris_to_mock = [
dict(method='GET',
uri='{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST}),
]
uris_to_mock.extend([
dict(method='GET',
uri='{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=flavor['id']),
json={'extra_specs': {}})
for flavor in fakes.FAKE_FLAVOR_LIST])
self.register_uris(uris_to_mock)
flavor = self.cloud.get_flavor_by_ram(ram=250)
self.assertEqual(fakes.STRAWBERRY_FLAVOR_ID, flavor['id'])
def test_get_flavor_by_ram_and_include(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST})
for flavor in fakes.FAKE_FLAVOR_LIST:
self.register_uri(
'GET', '{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=flavor['id']),
json={'extra_specs': {}})
uris_to_mock = [
dict(method='GET',
uri='{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': fakes.FAKE_FLAVOR_LIST}),
]
uris_to_mock.extend([
dict(method='GET',
uri='{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=flavor['id']),
json={'extra_specs': {}})
for flavor in fakes.FAKE_FLAVOR_LIST])
self.register_uris(uris_to_mock)
flavor = self.cloud.get_flavor_by_ram(ram=150, include='strawberry')
self.assertEqual(fakes.STRAWBERRY_FLAVOR_ID, flavor['id'])
def test_get_flavor_by_ram_not_found(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': []})
self.register_uris([
dict(method='GET',
uri='{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': []})])
self.assertRaises(
shade.OpenStackCloudException,
self.cloud.get_flavor_by_ram,
ram=100)
def test_get_flavor_string_and_int(self):
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': [fakes.make_fake_flavor('1', 'vanilla')]})
self.register_uri(
'GET', '{endpoint}/flavors/1/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'extra_specs': {}})
self.register_uri(
'GET', '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'flavors': [fakes.make_fake_flavor('1', 'vanilla')]})
self.register_uri(
'GET', '{endpoint}/flavors/1/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={'extra_specs': {}})
flavor_list_uri = '{endpoint}/flavors/detail?is_public=None'.format(
endpoint=fakes.COMPUTE_ENDPOINT)
flavor_resource_uri = '{endpoint}/flavors/1/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT)
flavor_list_json = {'flavors': [fakes.make_fake_flavor(
'1', 'vanilla')]}
flavor_json = {'extra_specs': {}}
self.register_uris([
dict(method='GET', uri=flavor_list_uri, json=flavor_list_json),
dict(method='GET', uri=flavor_resource_uri, json=flavor_json),
dict(method='GET', uri=flavor_list_uri, json=flavor_list_json),
dict(method='GET', uri=flavor_resource_uri, json=flavor_json)])
flavor1 = self.cloud.get_flavor('1')
self.assertEqual('1', flavor1['id'])
@ -164,65 +179,60 @@ class TestFlavors(base.RequestsMockTestCase):
def test_set_flavor_specs(self):
extra_specs = dict(key1='value1')
self.register_uri(
'POST', '{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=1),
json=dict(extra_specs=extra_specs))
self.register_uris([
dict(method='POST',
uri='{endpoint}/flavors/{id}/os-extra_specs'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=1),
json=dict(extra_specs=extra_specs))])
self.op_cloud.set_flavor_specs(1, extra_specs)
self.assert_calls()
def test_unset_flavor_specs(self):
keys = ['key1', 'key2']
for key in keys:
self.register_uri(
'DELETE',
'{endpoint}/flavors/{id}/os-extra_specs/{key}'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=1, key=key))
self.register_uris([
dict(method='DELETE',
uri='{endpoint}/flavors/{id}/os-extra_specs/{key}'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id=1, key=key))
for key in keys])
self.op_cloud.unset_flavor_specs(1, keys)
self.assert_calls()
def test_add_flavor_access(self):
self.register_uri(
'POST', '{endpoint}/flavors/{id}/action'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id='flavor_id'),
json={
'flavor_access': [{
'flavor_id': 'flavor_id',
'tenant_id': 'tenant_id',
}]},
validate=dict(
json={
'addTenantAccess': {
'tenant': 'tenant_id',
}}))
self.register_uris([
dict(method='POST',
uri='{endpoint}/flavors/{id}/action'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id='flavor_id'),
json={
'flavor_access': [{
'flavor_id': 'flavor_id', 'tenant_id': 'tenant_id'}]},
validate=dict(
json={'addTenantAccess': {'tenant': 'tenant_id'}}))])
self.op_cloud.add_flavor_access('flavor_id', 'tenant_id')
self.assert_calls()
def test_remove_flavor_access(self):
self.register_uri(
'POST', '{endpoint}/flavors/{id}/action'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id='flavor_id'),
json={'flavor_access': []},
validate=dict(
json={
'removeTenantAccess': {
'tenant': 'tenant_id',
}}))
self.register_uris([
dict(method='POST',
uri='{endpoint}/flavors/{id}/action'.format(
endpoint=fakes.COMPUTE_ENDPOINT, id='flavor_id'),
json={'flavor_access': []},
validate=dict(
json={'removeTenantAccess': {'tenant': 'tenant_id'}}))])
self.op_cloud.remove_flavor_access('flavor_id', 'tenant_id')
self.assert_calls()
def test_list_flavor_access(self):
self.register_uri(
'GET', '{endpoint}/flavors/vanilla/os-flavor-access'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={
'flavor_access': [{
'flavor_id': 'vanilla',
'tenant_id': 'tenant_id',
}]})
self.register_uris([
dict(method='GET',
uri='{endpoint}/flavors/vanilla/os-flavor-access'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={
'flavor_access': [
{'flavor_id': 'vanilla', 'tenant_id': 'tenant_id'}]})
])
self.op_cloud.list_flavor_access('vanilla')
self.assert_calls()

View File

@ -163,9 +163,10 @@ class TestFloatingIP(base.RequestsMockTestCase):
self.assertEqual('UNKNOWN', normalized[0]['status'])
def test_list_floating_ips(self):
self.register_uri(
'GET', 'https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_list_rep)
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_list_rep)])
floating_ips = self.cloud.list_floating_ips()
@ -177,19 +178,22 @@ class TestFloatingIP(base.RequestsMockTestCase):
def test_list_floating_ips_with_filters(self):
self.register_uri(
'GET', 'https://network.example.com/v2.0/floatingips.json?Foo=42',
json={'floatingips': []})
self.register_uris([
dict(method='GET',
uri=('https://network.example.com/v2.0/floatingips.json?'
'Foo=42'),
json={'floatingips': []})])
self.cloud.list_floating_ips(filters={'Foo': 42})
self.assert_calls()
def test_search_floating_ips(self):
self.register_uri(
'GET',
'https://network.example.com/v2.0/floatingips.json?attached=False',
json=self.mock_floating_ip_list_rep)
self.register_uris([
dict(method='GET',
uri=('https://network.example.com/v2.0/floatingips.json'
'?attached=False'),
json=self.mock_floating_ip_list_rep)])
floating_ips = self.cloud.search_floating_ips(
filters={'attached': False})
@ -200,9 +204,10 @@ class TestFloatingIP(base.RequestsMockTestCase):
self.assert_calls()
def test_get_floating_ip(self):
self.register_uri(
'GET', 'https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_list_rep)
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_list_rep)])
floating_ip = self.cloud.get_floating_ip(
id='2f245a7b-796b-4f26-9cf9-9e82d248fda7')
@ -221,9 +226,10 @@ class TestFloatingIP(base.RequestsMockTestCase):
self.assert_calls()
def test_get_floating_ip_not_found(self):
self.register_uri(
'GET', 'https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_list_rep)
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_list_rep)])
floating_ip = self.cloud.get_floating_ip(id='non-existent')
@ -231,14 +237,17 @@ class TestFloatingIP(base.RequestsMockTestCase):
self.assert_calls()
def test_create_floating_ip(self):
self.register_uri(
'GET', 'https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]})
self.register_uri(
'POST', 'https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_new_rep,
validate=dict(
json={'floatingip': {'floating_network_id': 'my-network-id'}}))
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]}),
dict(method='POST',
uri='https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_new_rep,
validate=dict(
json={'floatingip': {
'floating_network_id': 'my-network-id'}}))
])
ip = self.cloud.create_floating_ip(network='my-network')
self.assertEqual(
@ -247,17 +256,18 @@ class TestFloatingIP(base.RequestsMockTestCase):
self.assert_calls()
def test_create_floating_ip_port_bad_response(self):
self.register_uri(
'GET', 'https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]})
self.register_uri(
'POST', 'https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_new_rep,
validate=dict(
json={'floatingip': {
'floating_network_id': 'my-network-id',
'port_id': u'ce705c24-c1ef-408a-bda3-7bbd946164ab',
}}))
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]}),
dict(method='POST',
uri='https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_new_rep,
validate=dict(
json={'floatingip': {
'floating_network_id': 'my-network-id',
'port_id': u'ce705c24-c1ef-408a-bda3-7bbd946164ab'}}))
])
# Fails because we requested a port and the returned FIP has no port
self.assertRaises(
@ -267,17 +277,18 @@ class TestFloatingIP(base.RequestsMockTestCase):
self.assert_calls()
def test_create_floating_ip_port(self):
self.register_uri(
'GET', 'https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]})
self.register_uri(
'POST', 'https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_port_rep,
validate=dict(
json={'floatingip': {
'floating_network_id': 'my-network-id',
'port_id': u'ce705c24-c1ef-408a-bda3-7bbd946164ac',
}}))
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]}),
dict(method='POST',
uri='https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_port_rep,
validate=dict(
json={'floatingip': {
'floating_network_id': 'my-network-id',
'port_id': u'ce705c24-c1ef-408a-bda3-7bbd946164ac'}}))
])
ip = self.cloud.create_floating_ip(
network='my-network', port='ce705c24-c1ef-408a-bda3-7bbd946164ac')
@ -287,51 +298,51 @@ class TestFloatingIP(base.RequestsMockTestCase):
ip['floating_ip_address'])
self.assert_calls()
def test__neutron_available_floating_ips(self):
def test_neutron_available_floating_ips(self):
"""
Test without specifying a network name.
"""
self.register_uri(
'GET', 'https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]})
self.register_uri(
'GET', 'https://network.example.com/v2.0/subnets.json',
json={'subnets': []})
self.register_uri(
'GET', 'https://network.example.com/v2.0/floatingips.json',
json={'floatingips': []})
self.register_uri(
'POST', 'https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_new_rep,
validate=dict(json={
'floatingip': {
'floating_network_id': self.mock_get_network_rep['id'],
}}))
fips_mock_uri = 'https://network.example.com/v2.0/floatingips.json'
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]}),
dict(method='GET',
uri='https://network.example.com/v2.0/subnets.json',
json={'subnets': []}),
dict(method='GET', uri=fips_mock_uri, json={'floatingips': []}),
dict(method='POST', uri=fips_mock_uri,
json=self.mock_floating_ip_new_rep,
validate=dict(json={
'floatingip': {
'floating_network_id': self.mock_get_network_rep['id']
}}))
])
# Test if first network is selected if no network is given
self.cloud._neutron_available_floating_ips()
self.assert_calls()
def test__neutron_available_floating_ips_network(self):
def test_neutron_available_floating_ips_network(self):
"""
Test with specifying a network name.
"""
self.register_uri(
'GET', 'https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]})
self.register_uri(
'GET', 'https://network.example.com/v2.0/subnets.json',
json={'subnets': []})
self.register_uri(
'GET', 'https://network.example.com/v2.0/floatingips.json',
json={'floatingips': []})
self.register_uri(
'POST', 'https://network.example.com/v2.0/floatingips.json',
json=self.mock_floating_ip_new_rep,
validate=dict(json={
'floatingip': {
'floating_network_id': self.mock_get_network_rep['id'],
}}))
fips_mock_uri = 'https://network.example.com/v2.0/floatingips.json'
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]}),
dict(method='GET',
uri='https://network.example.com/v2.0/subnets.json',
json={'subnets': []}),
dict(method='GET', uri=fips_mock_uri, json={'floatingips': []}),
dict(method='POST', uri=fips_mock_uri,
json=self.mock_floating_ip_new_rep,
validate=dict(json={
'floatingip': {
'floating_network_id': self.mock_get_network_rep['id']
}}))
])
# Test if first network is selected if no network is given
self.cloud._neutron_available_floating_ips(
@ -339,16 +350,18 @@ class TestFloatingIP(base.RequestsMockTestCase):
)
self.assert_calls()
def test__neutron_available_floating_ips_invalid_network(self):
def test_neutron_available_floating_ips_invalid_network(self):
"""
Test with an invalid network name.
"""
self.register_uri(
'GET', 'https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]})
self.register_uri(
'GET', 'https://network.example.com/v2.0/subnets.json',
json={'subnets': []})
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/networks.json',
json={'networks': [self.mock_get_network_rep]}),
dict(method='GET',
uri='https://network.example.com/v2.0/subnets.json',
json={'subnets': []})
])
self.assertRaises(
exc.OpenStackCloudException,
@ -359,205 +372,202 @@ class TestFloatingIP(base.RequestsMockTestCase):
def test_auto_ip_pool_no_reuse(self):
# payloads taken from citycloud
self.register_uri(
'GET', 'https://network.example.com/v2.0/networks.json',
json={"networks": [{
"status": "ACTIVE",
"subnets": [
"df3e17fa-a4b2-47ae-9015-bc93eb076ba2",
"6b0c3dc9-b0b8-4d87-976a-7f2ebf13e7ec",
"fc541f48-fc7f-48c0-a063-18de6ee7bdd7"],
"availability_zone_hints": [],
"availability_zones": ["nova"],
"name": "ext-net",
"admin_state_up": True,
"tenant_id": "a564613210ee43708b8a7fc6274ebd63",
"tags": [],
"ipv6_address_scope": "9f03124f-89af-483a-b6fd-10f08079db4d",
"mtu": 0,
"is_default": False,
"router:external": True,
"ipv4_address_scope": None,
"shared": False,
"id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf",
"description": None
}, {
"status": "ACTIVE",
"subnets": ["f0ad1df5-53ee-473f-b86b-3604ea5591e9"],
"availability_zone_hints": [],
"availability_zones": ["nova"],
"name": "private",
"admin_state_up": True,
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"created_at": "2016-10-22T13:46:26",
"tags": [],
"updated_at": "2016-10-22T13:46:26",
"ipv6_address_scope": None,
"router:external": False,
"ipv4_address_scope": None,
"shared": False,
"mtu": 1450,
"id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
"description": ""
}]})
self.register_uri(
'GET',
'https://network.example.com/v2.0/ports.json'
'?device_id=f80e3ad0-e13e-41d4-8e9c-be79bccdb8f7',
json={"ports": [{
"status": "ACTIVE",
"created_at": "2017-02-06T20:59:45",
"description": "",
"allowed_address_pairs": [],
"admin_state_up": True,
"network_id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
"dns_name": None,
"extra_dhcp_opts": [],
"mac_address": "fa:16:3e:e8:7f:03",
"updated_at": "2017-02-06T20:59:49",
"name": "",
"device_owner": "compute:None",
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"binding:vnic_type": "normal",
"fixed_ips": [{
"subnet_id": "f0ad1df5-53ee-473f-b86b-3604ea5591e9",
"ip_address": "10.4.0.16"}],
"id": "a767944e-057a-47d1-a669-824a21b8fb7b",
"security_groups": ["9fb5ba44-5c46-4357-8e60-8b55526cab54"],
"device_id": "f80e3ad0-e13e-41d4-8e9c-be79bccdb8f7",
}]})
self.register_uris([
dict(method='GET',
uri='https://network.example.com/v2.0/networks.json',
json={"networks": [{
"status": "ACTIVE",
"subnets": [
"df3e17fa-a4b2-47ae-9015-bc93eb076ba2",
"6b0c3dc9-b0b8-4d87-976a-7f2ebf13e7ec",
"fc541f48-fc7f-48c0-a063-18de6ee7bdd7"],
"availability_zone_hints": [],
"availability_zones": ["nova"],
"name": "ext-net",
"admin_state_up": True,
"tenant_id": "a564613210ee43708b8a7fc6274ebd63",
"tags": [],
"ipv6_address_scope": "9f03124f-89af-483a-b6fd-10f08079db4d", # noqa
"mtu": 0,
"is_default": False,
"router:external": True,
"ipv4_address_scope": None,
"shared": False,
"id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf",
"description": None
}, {
"status": "ACTIVE",
"subnets": ["f0ad1df5-53ee-473f-b86b-3604ea5591e9"],
"availability_zone_hints": [],
"availability_zones": ["nova"],
"name": "private",
"admin_state_up": True,
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"created_at": "2016-10-22T13:46:26",
"tags": [],
"updated_at": "2016-10-22T13:46:26",
"ipv6_address_scope": None,
"router:external": False,
"ipv4_address_scope": None,
"shared": False,
"mtu": 1450,
"id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
"description": ""
}]}),
dict(method='GET',
uri='https://network.example.com/v2.0/ports.json'
'?device_id=f80e3ad0-e13e-41d4-8e9c-be79bccdb8f7',
json={"ports": [{
"status": "ACTIVE",
"created_at": "2017-02-06T20:59:45",
"description": "",
"allowed_address_pairs": [],
"admin_state_up": True,
"network_id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
"dns_name": None,
"extra_dhcp_opts": [],
"mac_address": "fa:16:3e:e8:7f:03",
"updated_at": "2017-02-06T20:59:49",
"name": "",
"device_owner": "compute:None",
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"binding:vnic_type": "normal",
"fixed_ips": [{
"subnet_id": "f0ad1df5-53ee-473f-b86b-3604ea5591e9",
"ip_address": "10.4.0.16"}],
"id": "a767944e-057a-47d1-a669-824a21b8fb7b",
"security_groups": [
"9fb5ba44-5c46-4357-8e60-8b55526cab54"],
"device_id": "f80e3ad0-e13e-41d4-8e9c-be79bccdb8f7",
}]}),
self.register_uri(
'POST',
'https://network.example.com/v2.0/floatingips.json',
json={"floatingip": {
"router_id": "9de9c787-8f89-4a53-8468-a5533d6d7fd1",
"status": "DOWN",
"description": "",
"dns_domain": "",
"floating_network_id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf",
"fixed_ip_address": "10.4.0.16",
"floating_ip_address": "89.40.216.153",
"port_id": "a767944e-057a-47d1-a669-824a21b8fb7b",
"id": "e69179dc-a904-4c9a-a4c9-891e2ecb984c",
"dns_name": "",
"tenant_id": "65222a4d09ea4c68934fa1028c77f394"
}},
validate=dict(json={"floatingip": {
"floating_network_id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf",
"fixed_ip_address": "10.4.0.16",
"port_id": "a767944e-057a-47d1-a669-824a21b8fb7b",
}}))
self.register_uri(
'GET',
'{endpoint}/servers/detail'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={"servers": [{
"status": "ACTIVE",
"updated": "2017-02-06T20:59:49Z",
"addresses": {
"private": [{
"OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:e8:7f:03",
"version": 4,
"addr": "10.4.0.16",
"OS-EXT-IPS:type": "fixed"
}, {
"OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:e8:7f:03",
"version": 4,
"addr": "89.40.216.153",
"OS-EXT-IPS:type": "floating"
}]},
"key_name": None,
"image": {"id": "95e4c449-8abf-486e-97d9-dc3f82417d2d"},
"OS-EXT-STS:task_state": None,
"OS-EXT-STS:vm_state": "active",
"OS-SRV-USG:launched_at": "2017-02-06T20:59:48.000000",
"flavor": {"id": "2186bd79-a05e-4953-9dde-ddefb63c88d4"},
"id": "f80e3ad0-e13e-41d4-8e9c-be79bccdb8f7",
"security_groups": [{"name": "default"}],
"OS-SRV-USG:terminated_at": None,
"OS-EXT-AZ:availability_zone": "nova",
"user_id": "c17534835f8f42bf98fc367e0bf35e09",
"name": "testmt",
"created": "2017-02-06T20:59:44Z",
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"OS-DCF:diskConfig": "MANUAL",
"os-extended-volumes:volumes_attached": [],
"accessIPv4": "",
"accessIPv6": "",
"progress": 0,
"OS-EXT-STS:power_state": 1,
"config_drive": "",
"metadata": {}
}]})
self.register_uri(
'GET', 'https://network.example.com/v2.0/networks.json',
json={"networks": [{
"status": "ACTIVE",
"subnets": [
"df3e17fa-a4b2-47ae-9015-bc93eb076ba2",
"6b0c3dc9-b0b8-4d87-976a-7f2ebf13e7ec",
"fc541f48-fc7f-48c0-a063-18de6ee7bdd7"],
"availability_zone_hints": [],
"availability_zones": ["nova"],
"name": "ext-net",
"admin_state_up": True,
"tenant_id": "a564613210ee43708b8a7fc6274ebd63",
"tags": [],
"ipv6_address_scope": "9f03124f-89af-483a-b6fd-10f08079db4d",
"mtu": 0,
"is_default": False,
"router:external": True,
"ipv4_address_scope": None,
"shared": False,
"id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf",
"description": None
}, {
"status": "ACTIVE",
"subnets": ["f0ad1df5-53ee-473f-b86b-3604ea5591e9"],
"availability_zone_hints": [],
"availability_zones": ["nova"],
"name": "private",
"admin_state_up": True,
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"created_at": "2016-10-22T13:46:26",
"tags": [],
"updated_at": "2016-10-22T13:46:26",
"ipv6_address_scope": None,
"router:external": False,
"ipv4_address_scope": None,
"shared": False,
"mtu": 1450,
"id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
"description": ""
}]})
self.register_uri(
'GET', 'https://network.example.com/v2.0/subnets.json',
json={"subnets": [{
"description": "",
"enable_dhcp": True,
"network_id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"created_at": "2016-10-22T13:46:26",
"dns_nameservers": [
"89.36.90.101",
"89.36.90.102"],
"updated_at": "2016-10-22T13:46:26",
"gateway_ip": "10.4.0.1",
"ipv6_ra_mode": None,
"allocation_pools": [{
"start": "10.4.0.2",
"end": "10.4.0.200"}],
"host_routes": [],
"ip_version": 4,
"ipv6_address_mode": None,
"cidr": "10.4.0.0/24",
"id": "f0ad1df5-53ee-473f-b86b-3604ea5591e9",
"subnetpool_id": None,
"name": "private-subnet-ipv4",
}]})
dict(method='POST',
uri='https://network.example.com/v2.0/floatingips.json',
json={"floatingip": {
"router_id": "9de9c787-8f89-4a53-8468-a5533d6d7fd1",
"status": "DOWN",
"description": "",
"dns_domain": "",
"floating_network_id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf", # noqa
"fixed_ip_address": "10.4.0.16",
"floating_ip_address": "89.40.216.153",
"port_id": "a767944e-057a-47d1-a669-824a21b8fb7b",
"id": "e69179dc-a904-4c9a-a4c9-891e2ecb984c",
"dns_name": "",
"tenant_id": "65222a4d09ea4c68934fa1028c77f394"
}},
validate=dict(json={"floatingip": {
"floating_network_id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf", # noqa
"fixed_ip_address": "10.4.0.16",
"port_id": "a767944e-057a-47d1-a669-824a21b8fb7b",
}})),
dict(method='GET',
uri='{endpoint}/servers/detail'.format(
endpoint=fakes.COMPUTE_ENDPOINT),
json={"servers": [{
"status": "ACTIVE",
"updated": "2017-02-06T20:59:49Z",
"addresses": {
"private": [{
"OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:e8:7f:03",
"version": 4,
"addr": "10.4.0.16",
"OS-EXT-IPS:type": "fixed"
}, {
"OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:e8:7f:03",
"version": 4,
"addr": "89.40.216.153",
"OS-EXT-IPS:type": "floating"
}]},
"key_name": None,
"image": {"id": "95e4c449-8abf-486e-97d9-dc3f82417d2d"},
"OS-EXT-STS:task_state": None,
"OS-EXT-STS:vm_state": "active",
"OS-SRV-USG:launched_at": "2017-02-06T20:59:48.000000",
"flavor": {"id": "2186bd79-a05e-4953-9dde-ddefb63c88d4"},
"id": "f80e3ad0-e13e-41d4-8e9c-be79bccdb8f7",
"security_groups": [{"name": "default"}],
"OS-SRV-USG:terminated_at": None,
"OS-EXT-AZ:availability_zone": "nova",
"user_id": "c17534835f8f42bf98fc367e0bf35e09",
"name": "testmt",
"created": "2017-02-06T20:59:44Z",
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"OS-DCF:diskConfig": "MANUAL",
"os-extended-volumes:volumes_attached": [],
"accessIPv4": "",
"accessIPv6": "",
"progress": 0,
"OS-EXT-STS:power_state": 1,
"config_drive": "",
"metadata": {}
}]}),
dict(method='GET',
uri='https://network.example.com/v2.0/networks.json',
json={"networks": [{
"status": "ACTIVE",
"subnets": [
"df3e17fa-a4b2-47ae-9015-bc93eb076ba2",
"6b0c3dc9-b0b8-4d87-976a-7f2ebf13e7ec",
"fc541f48-fc7f-48c0-a063-18de6ee7bdd7"],
"availability_zone_hints": [],
"availability_zones": ["nova"],
"name": "ext-net",
"admin_state_up": True,
"tenant_id": "a564613210ee43708b8a7fc6274ebd63",
"tags": [],
"ipv6_address_scope": "9f03124f-89af-483a-b6fd-10f08079db4d", # noqa
"mtu": 0,
"is_default": False,
"router:external": True,
"ipv4_address_scope": None,
"shared": False,
"id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf",
"description": None
}, {
"status": "ACTIVE",
"subnets": ["f0ad1df5-53ee-473f-b86b-3604ea5591e9"],
"availability_zone_hints": [],
"availability_zones": ["nova"],
"name": "private",
"admin_state_up": True,
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"created_at": "2016-10-22T13:46:26",
"tags": [],
"updated_at": "2016-10-22T13:46:26",
"ipv6_address_scope": None,
"router:external": False,
"ipv4_address_scope": None,
"shared": False,
"mtu": 1450,
"id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
"description": ""
}]}),
dict(method='GET',
uri='https://network.example.com/v2.0/subnets.json',
json={"subnets": [{
"description": "",
"enable_dhcp": True,
"network_id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f",
"tenant_id": "65222a4d09ea4c68934fa1028c77f394",
"created_at": "2016-10-22T13:46:26",
"dns_nameservers": [
"89.36.90.101",
"89.36.90.102"],
"updated_at": "2016-10-22T13:46:26",
"gateway_ip": "10.4.0.1",
"ipv6_ra_mode": None,
"allocation_pools": [{
"start": "10.4.0.2",
"end": "10.4.0.200"}],
"host_routes": [],
"ip_version": 4,
"ipv6_address_mode": None,
"cidr": "10.4.0.0/24",
"id": "f0ad1df5-53ee-473f-b86b-3604ea5591e9",
"subnetpool_id": None,
"name": "private-subnet-ipv4",
}]})])
self.cloud.add_ips_to_server(
munch.Munch(
@ -678,11 +688,11 @@ class TestFloatingIP(base.RequestsMockTestCase):
self.assertEqual(mock_get_floating_ip.call_count, 3)
def test_delete_floating_ip_not_found(self):
self.register_uri(
'DELETE',
'https://network.example.com/v2.0/floatingips'
'/a-wild-id-appears.json',
status_code=404)
self.register_uris([
dict(method='DELETE',
uri=('https://network.example.com/v2.0/floatingips/'
'a-wild-id-appears.json'),
status_code=404)])
ret = self.cloud.delete_floating_ip(
floating_ip_id='a-wild-id-appears')

View File

@ -33,18 +33,17 @@ class TestUsers(base.RequestsMockTestCase):
user_data = self._get_user_data()
self.register_uri('POST',
self._get_keystone_mock_url(resource='users',
v3=False),
status_code=204,
json=user_data.json_response,
validate=dict(json=user_data.json_request))
self.register_uri('GET',
self._get_keystone_mock_url(
resource='users',
append=[user_data.user_id],
v3=False),
status_code=200, json=user_data.json_response)
self.register_uris([
dict(method='POST',
uri=self._get_keystone_mock_url(resource='users', v3=False),
status_code=200,
json=user_data.json_response,
validate=user_data.json_request),
dict(method='GET',
uri=self._get_keystone_mock_url(resource='users', v3=False,
append=[user_data.user_id]),
status_code=200, json=user_data.json_response)])
user = self.op_cloud.create_user(
name=user_data.name, email=user_data.email,
password=user_data.password)
@ -59,17 +58,17 @@ class TestUsers(base.RequestsMockTestCase):
user_data = self._get_user_data(
domain_id=uuid.uuid4().hex,
description=self.getUniqueString('description'))
self.register_uri(
'POST',
self._get_keystone_mock_url(resource='users'),
status_code=204,
json=user_data.json_response,
validate=user_data.json_request)
self.register_uri(
'GET',
self._get_keystone_mock_url(resource='users',
append=[user_data.user_id]),
status_code=200, json=user_data.json_response)
self.register_uris([
dict(method='POST',
uri=self._get_keystone_mock_url(resource='users'),
status_code=200, json=user_data.json_response,
validate=user_data.json_request),
dict(method='GET',
uri=self._get_keystone_mock_url(
resource='users', append=[user_data.user_id]),
status_code=200, json=user_data.json_response)])
user = self.op_cloud.create_user(
name=user_data.name, email=user_data.email,
password=user_data.password,
@ -84,47 +83,40 @@ class TestUsers(base.RequestsMockTestCase):
def test_update_user_password_v2(self):
self.use_keystone_v2()
user_data = self._get_user_data(email='test@example.com')
self.register_uri('GET',
self._get_keystone_mock_url(resource='users',
v3=False),
status_code=200,
json={'users': [user_data.json_response['user']]})
self.register_uri('GET',
self._get_keystone_mock_url(
resource='users',
v3=False,
append=[user_data.user_id]),
json=user_data.json_response)
self.register_uri(
'PUT',
self._get_keystone_mock_url(
resource='users', v3=False,
append=[user_data.user_id, 'OS-KSADM', 'password']),
status_code=204, json=user_data.json_response,
validate=dict(json={'user': {'password': user_data.password}}))
self.register_uri('GET',
self._get_keystone_mock_url(
resource='users', v3=False,
append=[user_data.user_id]),
json=user_data.json_response)
# NOTE(notmorgan): when keystoneclient is dropped, the extra call is
# not needed as it is a blank put. Keystoneclient has very limited
# logic and does odd things when updates inclue passwords in v2
# keystone.
self.register_uri(
'PUT',
self._get_keystone_mock_url(resource='users',
append=[user_data.user_id],
v3=False),
status_code=204, json=user_data.json_response,
validate=dict(json={'user': {}}))
self.register_uri('GET',
self._get_keystone_mock_url(
resource='users',
v3=False,
append=[user_data.user_id]),
json=user_data.json_response)
mock_user_resource_uri = self._get_keystone_mock_url(
resource='users', append=[user_data.user_id], v3=False)
mock_users_uri = self._get_keystone_mock_url(
resource='users', v3=False)
self.register_uris([
# GET list to find user id
# GET user info with user_id from list
# PUT user with password update
# GET user info with id after update
# PUT empty update (password change is different than update)
# but is always chained together [keystoneclient oddity]
# GET user info after user update
dict(method='GET', uri=mock_users_uri, status_code=200,
json={'users': [user_data.json_response['user']]}),
dict(method='GET', uri=mock_user_resource_uri, status_code=200,
json=user_data.json_response),
dict(method='PUT',
uri=self._get_keystone_mock_url(
resource='users', v3=False,
append=[user_data.user_id, 'OS-KSADM', 'password']),
status_code=200, json=user_data.json_response,
validate=dict(
json={'user': {'password': user_data.password}})),
dict(method='GET', uri=mock_user_resource_uri, status_code=200,
json=user_data.json_response),
dict(method='PUT', uri=mock_user_resource_uri, status_code=200,
json=user_data.json_response,
validate=dict(json={'user': {}})),
dict(method='GET', uri=mock_user_resource_uri, status_code=200,
json=user_data.json_response)])
user = self.op_cloud.update_user(
user_data.user_id, password=user_data.password)
self.assertEqual(user_data.name, user.name)
@ -146,47 +138,48 @@ class TestUsers(base.RequestsMockTestCase):
def test_delete_user(self):
self._add_discovery_uri_call()
user_data = self._get_user_data(domain_id=uuid.uuid4().hex)
self.register_uri('GET', self._get_keystone_mock_url(resource='users'),
status_code=200,
json={'users': [user_data.json_response['user']]})
self.register_uri('GET',
self._get_keystone_mock_url(
resource='users', append=[user_data.user_id]),
status_code=200, json=user_data.json_response)
self.register_uri('DELETE',
self._get_keystone_mock_url(
resource='users', append=[user_data.user_id]),
status_code=204)
user_resource_uri = self._get_keystone_mock_url(
resource='users', append=[user_data.user_id])
self.register_uris([
dict(method='GET',
uri=self._get_keystone_mock_url(resource='users'),
status_code=200,
json={'users': [user_data.json_response['user']]}),
dict(method='GET', uri=user_resource_uri, status_code=200,
json=user_data.json_response),
dict(method='DELETE', uri=user_resource_uri, status_code=204)])
self.op_cloud.delete_user(user_data.name)
self.assert_calls()
def test_delete_user_not_found(self):
self._add_discovery_uri_call()
self.register_uri('GET',
self._get_keystone_mock_url(resource='users'),
status_code=200,
json={'users': []})
self.register_uris([
dict(method='GET',
uri=self._get_keystone_mock_url(resource='users'),
status_code=200, json={'users': []})])
self.assertFalse(self.op_cloud.delete_user(self.getUniqueString()))
def test_add_user_to_group(self):
self._add_discovery_uri_call()
user_data = self._get_user_data()
group_data = self._get_group_data()
self.register_uri('GET',
self._get_keystone_mock_url(resource='users'),
status_code=200,
json={'users': [user_data.json_response['user']]})
self.register_uri(
'GET',
self._get_keystone_mock_url(resource='groups'),
status_code=200,
json={'groups': [group_data.json_response['group']]})
self.register_uri(
'PUT',
self._get_keystone_mock_url(
resource='groups',
append=[group_data.group_id, 'users', user_data.user_id]),
status_code=200)
self.register_uris([
dict(method='GET',
uri=self._get_keystone_mock_url(resource='users'),
status_code=200,
json={'users': [user_data.json_response['user']]}),
dict(method='GET',
uri=self._get_keystone_mock_url(resource='groups'),
status_code=200,
json={'groups': [group_data.json_response['group']]}),
dict(method='PUT',
uri=self._get_keystone_mock_url(
resource='groups',
append=[group_data.group_id, 'users', user_data.user_id]),
status_code=200)])
self.op_cloud.add_user_to_group(user_data.user_id, group_data.group_id)
self.assert_calls()
@ -194,21 +187,22 @@ class TestUsers(base.RequestsMockTestCase):
self._add_discovery_uri_call()
user_data = self._get_user_data()
group_data = self._get_group_data()
self.register_uri('GET',
self._get_keystone_mock_url(resource='users'),
status_code=200,
json={'users': [user_data.json_response['user']]})
self.register_uri(
'GET',
self._get_keystone_mock_url(resource='groups'),
status_code=200,
json={'groups': [group_data.json_response['group']]})
self.register_uri(
'HEAD',
self._get_keystone_mock_url(
resource='groups',
append=[group_data.group_id, 'users', user_data.user_id]),
status_code=204)
self.register_uris([
dict(method='GET',
uri=self._get_keystone_mock_url(resource='users'),
status_code=200,
json={'users': [user_data.json_response['user']]}),
dict(method='GET',
uri=self._get_keystone_mock_url(resource='groups'),
status_code=200,
json={'groups': [group_data.json_response['group']]}),
dict(method='HEAD',
uri=self._get_keystone_mock_url(
resource='groups',
append=[group_data.group_id, 'users', user_data.user_id]),
status_code=204)])
self.assertTrue(self.op_cloud.is_user_in_group(
user_data.user_id, group_data.group_id))
self.assert_calls()
@ -217,21 +211,21 @@ class TestUsers(base.RequestsMockTestCase):
self._add_discovery_uri_call()
user_data = self._get_user_data()
group_data = self._get_group_data()
self.register_uri('GET',
self._get_keystone_mock_url(resource='users'),
status_code=200,
json={'users': [user_data.json_response['user']]})
self.register_uri(
'GET',
self._get_keystone_mock_url(resource='groups'),
status_code=200,
json={'groups': [group_data.json_response['group']]})
self.register_uri(
'DELETE',
self._get_keystone_mock_url(
resource='groups',
append=[group_data.group_id, 'users', user_data.user_id]),
status_code=204)
self.register_uris([
dict(method='GET',
uri=self._get_keystone_mock_url(resource='users'),
json={'users': [user_data.json_response['user']]}),
dict(method='GET',
uri=self._get_keystone_mock_url(resource='groups'),
status_code=200,
json={'groups': [group_data.json_response['group']]}),
dict(method='DELETE',
uri=self._get_keystone_mock_url(
resource='groups',
append=[group_data.group_id, 'users', user_data.user_id]),
status_code=204)])
self.op_cloud.remove_user_from_group(user_data.user_id,
group_data.group_id)
self.assert_calls()