Merge "Convert use of .register_uri to .register_uris"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
844ec53ad0
@ -128,7 +128,13 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
# it on cleanup). Subclassing here could be 100% eliminated in the
|
||||
# future allowing any class to simply
|
||||
# self.useFixture(shade.RequestsMockFixture) and get all the benefits.
|
||||
self._uri_registry = {}
|
||||
|
||||
# NOTE(notmorgan): use an ordered dict here to ensure we preserve the
|
||||
# order in which items are added to the uri_registry. This makes
|
||||
# the behavior more consistent when dealing with ensuring the
|
||||
# requests_mock uri/query_string matchers are ordered and parse the
|
||||
# request in the correct orders.
|
||||
self._uri_registry = collections.OrderedDict()
|
||||
self.discovery_json = os.path.join(
|
||||
self.fixtures_directory, 'discovery.json')
|
||||
self.use_keystone_v3()
|
||||
@ -167,29 +173,33 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
# Generate multiple projects
|
||||
project_list = [self._get_project_data(v3=v3)
|
||||
for c in range(0, project_count)]
|
||||
uri_mock_list = []
|
||||
if list_get:
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(
|
||||
service_type='identity',
|
||||
interface='admin',
|
||||
resource='projects',
|
||||
base_url_append=base_url_append),
|
||||
status_code=200,
|
||||
json={'projects': [p.json_response['project']
|
||||
for p in project_list]})
|
||||
uri_mock_list.append(
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
service_type='identity',
|
||||
interface='admin',
|
||||
resource='projects',
|
||||
base_url_append=base_url_append),
|
||||
status_code=200,
|
||||
json={'projects': [p.json_response['project']
|
||||
for p in project_list]})
|
||||
)
|
||||
if id_get:
|
||||
for p in project_list:
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(
|
||||
service_type='identity',
|
||||
interface='admin',
|
||||
resource='projects',
|
||||
append=[p.project_id],
|
||||
base_url_append=base_url_append),
|
||||
status_code=200,
|
||||
json=p.json_response)
|
||||
uri_mock_list.append(
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
service_type='identity',
|
||||
interface='admin',
|
||||
resource='projects',
|
||||
append=[p.project_id],
|
||||
base_url_append=base_url_append),
|
||||
status_code=200,
|
||||
json=p.json_response)
|
||||
)
|
||||
self.__do_register_uris(uri_mock_list)
|
||||
return project_list
|
||||
|
||||
def _get_project_data(self, project_name=None, enabled=None,
|
||||
@ -264,36 +274,35 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
self.adapter = self.useFixture(rm_fixture.Fixture())
|
||||
self.calls = []
|
||||
self._uri_registry.clear()
|
||||
self.register_uri('GET', 'https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read())
|
||||
self.register_uri(
|
||||
'POST', 'https://identity.example.com/v3/auth/tokens',
|
||||
headers={
|
||||
'X-Subject-Token': self.getUniqueString()},
|
||||
text=open(
|
||||
os.path.join(
|
||||
self.fixtures_directory,
|
||||
'catalog-v3.json'),
|
||||
'r').read())
|
||||
self.__do_register_uris([
|
||||
dict(method='GET', uri='https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read()),
|
||||
dict(method='POST',
|
||||
uri='https://identity.example.com/v3/auth/tokens',
|
||||
headers={
|
||||
'X-Subject-Token': self.getUniqueString('KeystoneToken')},
|
||||
text=open(os.path.join(
|
||||
self.fixtures_directory, 'catalog-v3.json'), 'r').read()
|
||||
)
|
||||
])
|
||||
self._make_test_cloud(identity_api_version='3')
|
||||
|
||||
def use_keystone_v2(self):
|
||||
self.adapter = self.useFixture(rm_fixture.Fixture())
|
||||
self.calls = []
|
||||
self._uri_registry.clear()
|
||||
self.register_uri('GET', 'https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read())
|
||||
self.register_uri(
|
||||
'POST', 'https://identity.example.com/v2.0/tokens',
|
||||
text=open(
|
||||
os.path.join(
|
||||
self.fixtures_directory,
|
||||
'catalog-v2.json'),
|
||||
'r').read())
|
||||
self.register_uri('GET', 'https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read())
|
||||
self.register_uri('GET', 'https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read())
|
||||
self.__do_register_uris([
|
||||
dict(method='GET', uri='https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read()),
|
||||
dict(method='POST', uri='https://identity.example.com/v2.0/tokens',
|
||||
text=open(os.path.join(
|
||||
self.fixtures_directory, 'catalog-v2.json'), 'r').read()
|
||||
),
|
||||
dict(method='GET', uri='https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read()),
|
||||
dict(method='GET', uri='https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read())
|
||||
])
|
||||
|
||||
self._make_test_cloud(cloud_name='_test_cloud_v2_',
|
||||
identity_api_version='2.0')
|
||||
@ -304,8 +313,9 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
# us to inject another call to discovery where needed in a test that
|
||||
# no longer mocks out kyestoneclient and performs the extra round
|
||||
# trips.
|
||||
self.register_uri('GET', 'https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read())
|
||||
self.__do_register_uris([
|
||||
dict(method='GET', uri='https://identity.example.com/',
|
||||
text=open(self.discovery_json, 'r').read())])
|
||||
|
||||
def _make_test_cloud(self, cloud_name='_test_cloud_', **kwargs):
|
||||
test_cloud = os.environ.get('SHADE_OS_CLOUD', cloud_name)
|
||||
@ -318,14 +328,23 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
cloud_config=self.cloud_config,
|
||||
log_inner_exceptions=True)
|
||||
|
||||
def use_glance(self, image_version_json='image-version.json'):
|
||||
def get_glance_discovery_mock_dict(
|
||||
self, image_version_json='image-version.json'):
|
||||
discovery_fixture = os.path.join(
|
||||
self.fixtures_directory, image_version_json)
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/',
|
||||
text=open(discovery_fixture, 'r').read())
|
||||
return dict(method='GET', uri='https://image.example.com/',
|
||||
text=open(discovery_fixture, 'r').read())
|
||||
|
||||
def register_uris(self, uri_mock_list):
|
||||
def use_glance(self, image_version_json='image-version.json'):
|
||||
# NOTE(notmorgan): This method is only meant to be used in "setUp"
|
||||
# where the ordering of the url being registered is tightly controlled
|
||||
# if the functionality of .use_glance is meant to be used during an
|
||||
# actual test case, use .get_glance_discovery_mock and apply to the
|
||||
# right location in the mock_uris when calling .register_uris
|
||||
self.__do_register_uris([
|
||||
self.get_glance_discovery_mock_dict(image_version_json)])
|
||||
|
||||
def register_uris(self, uri_mock_list=None):
|
||||
"""Mock a list of URIs and responses via requests mock.
|
||||
|
||||
This method may be called only once per test-case to avoid odd
|
||||
@ -354,13 +373,27 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
Methods are allowed and will be collapsed into a
|
||||
single matcher. Each response will be returned
|
||||
in order as the URI+Method is hit.
|
||||
:type uri_mock_list: list
|
||||
:return: None
|
||||
"""
|
||||
assert not self.__register_uris_called
|
||||
self.__do_register_uris(uri_mock_list or [])
|
||||
self.__register_uris_called = True
|
||||
|
||||
def __do_register_uris(self, uri_mock_list=None):
|
||||
for to_mock in uri_mock_list:
|
||||
kw_params = {k: to_mock.pop(k)
|
||||
for k in ('request_headers', 'complete_qs',
|
||||
'_real_http')
|
||||
if k in to_mock}
|
||||
|
||||
method = to_mock.pop('method')
|
||||
uri = to_mock.pop('uri')
|
||||
key = '{method}:{uri}'.format(method=method, uri=uri)
|
||||
# NOTE(notmorgan): make sure the delimiter is non-url-safe, in this
|
||||
# case "|" is used so that the split can be a bit easier on
|
||||
# maintainers of this code.
|
||||
key = '{method}|{uri}|{params}'.format(
|
||||
method=method, uri=uri, params=kw_params)
|
||||
validate = to_mock.pop('validate', {})
|
||||
headers = structures.CaseInsensitiveDict(to_mock.pop('headers',
|
||||
{}))
|
||||
@ -374,33 +407,27 @@ class RequestsMockTestCase(BaseTestCase):
|
||||
method=method,
|
||||
url=uri, **validate)
|
||||
]
|
||||
self._uri_registry.setdefault(key, []).append(to_mock)
|
||||
self._uri_registry.setdefault(
|
||||
key, {'response_list': [], 'kw_params': kw_params})
|
||||
if self._uri_registry[key]['kw_params'] != kw_params:
|
||||
raise AssertionError(
|
||||
'PROGRAMMING ERROR: key-word-params '
|
||||
'should be part of the uri_key and cannot change, '
|
||||
'it will affect the matcher in requests_mock. '
|
||||
'%(old)r != %(new)r' %
|
||||
{'old': self._uri_registry[key]['kw_params'],
|
||||
'new': kw_params})
|
||||
self._uri_registry[key]['response_list'].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
|
||||
for mocked, params in self._uri_registry.items():
|
||||
mock_method, mock_uri, _ignored = mocked.split('|', 2)
|
||||
self.adapter.register_uri(
|
||||
mock_method, mock_uri, params['response_list'],
|
||||
**params['kw_params'])
|
||||
|
||||
def register_uri(self, method, uri, **kwargs):
|
||||
validate = kwargs.pop('validate', {})
|
||||
key = '{method}:{uri}'.format(method=method, uri=uri)
|
||||
headers = structures.CaseInsensitiveDict(kwargs.pop('headers', {}))
|
||||
if 'content-type' not in headers:
|
||||
headers[u'content-type'] = 'application/json'
|
||||
kwargs['headers'] = headers
|
||||
|
||||
if key in self._uri_registry:
|
||||
self._uri_registry[key].append(kwargs)
|
||||
self.adapter.register_uri(method, uri, self._uri_registry[key])
|
||||
else:
|
||||
self._uri_registry[key] = [kwargs]
|
||||
self.adapter.register_uri(method, uri, **kwargs)
|
||||
|
||||
self.calls += [
|
||||
dict(
|
||||
method=method,
|
||||
url=uri, **validate)
|
||||
]
|
||||
self.__do_register_uris([
|
||||
dict(method=method, uri=uri, **kwargs)])
|
||||
|
||||
def assert_calls(self, stop_after=None):
|
||||
for (x, (call, history)) in enumerate(
|
||||
|
||||
@ -108,25 +108,26 @@ class TestImage(BaseTestImage):
|
||||
output_path='fake_path', output_file=fake_fd)
|
||||
|
||||
def test_download_image_no_images_found(self):
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json=dict(images=[]))
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://image.example.com/v2/images',
|
||||
json=dict(images=[]))])
|
||||
self.assertRaises(exc.OpenStackCloudResourceNotFound,
|
||||
self.cloud.download_image, 'fake_image',
|
||||
output_path='fake_path')
|
||||
self.assert_calls()
|
||||
|
||||
def _register_image_mocks(self):
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images/{id}/file'.format(
|
||||
id=self.image_id),
|
||||
content=self.output,
|
||||
headers={
|
||||
'Content-Type': 'application/octet-stream'
|
||||
})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://image.example.com/v2/images',
|
||||
json=self.fake_search_return),
|
||||
dict(method='GET',
|
||||
uri='https://image.example.com/v2/images/{id}/file'.format(
|
||||
id=self.image_id),
|
||||
content=self.output,
|
||||
headers={'Content-Type': 'application/octet-stream'})
|
||||
])
|
||||
|
||||
def test_download_image_with_fd(self):
|
||||
self._register_image_mocks()
|
||||
@ -145,15 +146,18 @@ class TestImage(BaseTestImage):
|
||||
self.assert_calls()
|
||||
|
||||
def test_empty_list_images(self):
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images', json={'images': []})
|
||||
self.register_uris([
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json={'images': []})
|
||||
])
|
||||
self.assertEqual([], self.cloud.list_images())
|
||||
self.assert_calls()
|
||||
|
||||
def test_list_images(self):
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
self.register_uris([
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
])
|
||||
self.assertEqual(
|
||||
self.cloud._normalize_images([self.fake_image_dict]),
|
||||
self.cloud.list_images())
|
||||
@ -161,17 +165,16 @@ class TestImage(BaseTestImage):
|
||||
|
||||
def test_list_images_paginated(self):
|
||||
marker = str(uuid.uuid4())
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json={
|
||||
'images': [self.fake_image_dict],
|
||||
'next': '/v2/images?marker={marker}'.format(marker=marker),
|
||||
})
|
||||
self.register_uri(
|
||||
'GET',
|
||||
'https://image.example.com/v2/images?marker={marker}'.format(
|
||||
marker=marker),
|
||||
json=self.fake_search_return)
|
||||
self.register_uris([
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json={'images': [self.fake_image_dict],
|
||||
'next': '/v2/images?marker={marker}'.format(
|
||||
marker=marker)}),
|
||||
dict(method='GET',
|
||||
uri=('https://image.example.com/v2/images?'
|
||||
'marker={marker}'.format(marker=marker)),
|
||||
json=self.fake_search_return)
|
||||
])
|
||||
self.assertEqual(
|
||||
self.cloud._normalize_images([
|
||||
self.fake_image_dict, self.fake_image_dict]),
|
||||
@ -181,32 +184,27 @@ class TestImage(BaseTestImage):
|
||||
def test_create_image_put_v2(self):
|
||||
self.cloud.image_api_use_tasks = False
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json={'images': []})
|
||||
|
||||
self.register_uri(
|
||||
'POST', 'https://image.example.com/v2/images',
|
||||
json=self.fake_image_dict,
|
||||
validate=dict(
|
||||
json={
|
||||
u'container_format': u'bare',
|
||||
u'disk_format': u'qcow2',
|
||||
u'name': u'fake_image',
|
||||
u'owner_specified.shade.md5': NO_MD5,
|
||||
u'owner_specified.shade.object': u'images/fake_image',
|
||||
u'owner_specified.shade.sha256': NO_SHA256,
|
||||
u'visibility': u'private'
|
||||
}),
|
||||
)
|
||||
self.register_uri(
|
||||
'PUT', 'https://image.example.com/v2/images/{id}/file'.format(
|
||||
id=self.image_id),
|
||||
request_headers={'Content-Type': 'application/octet-stream'})
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
self.register_uris([
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json={'images': []}),
|
||||
dict(method='POST', uri='https://image.example.com/v2/images',
|
||||
json=self.fake_image_dict,
|
||||
validate=dict(
|
||||
json={u'container_format': u'bare',
|
||||
u'disk_format': u'qcow2',
|
||||
u'name': u'fake_image',
|
||||
u'owner_specified.shade.md5': NO_MD5,
|
||||
u'owner_specified.shade.object': u'images/fake_image', # noqa
|
||||
u'owner_specified.shade.sha256': NO_SHA256,
|
||||
u'visibility': u'private'})
|
||||
),
|
||||
dict(method='PUT',
|
||||
uri='https://image.example.com/v2/images/{id}/file'.format(
|
||||
id=self.image_id),
|
||||
request_headers={'Content-Type': 'application/octet-stream'}),
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
])
|
||||
|
||||
self.cloud.create_image(
|
||||
'fake_image', self.imagefile.name, wait=True, timeout=1,
|
||||
@ -221,64 +219,6 @@ class TestImage(BaseTestImage):
|
||||
container_name = 'image_upload_v2_test_container'
|
||||
endpoint = self.cloud._object_store_client.get_endpoint()
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images', json={'images': []})
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://object-store.example.com/info',
|
||||
json=dict(
|
||||
swift={'max_file_size': 1000},
|
||||
slo={'min_segment_size': 500}))
|
||||
|
||||
self.register_uri(
|
||||
'HEAD', '{endpoint}/{container}'.format(
|
||||
endpoint=endpoint,
|
||||
container=container_name),
|
||||
status_code=404)
|
||||
|
||||
self.register_uri(
|
||||
'PUT', '{endpoint}/{container}'.format(
|
||||
endpoint=endpoint,
|
||||
container=container_name,),
|
||||
status_code=201,
|
||||
headers={
|
||||
'Date': 'Fri, 16 Dec 2016 18:21:20 GMT',
|
||||
'Content-Length': '0',
|
||||
'Content-Type': 'text/html; charset=UTF-8',
|
||||
})
|
||||
|
||||
self.register_uri(
|
||||
'HEAD', '{endpoint}/{container}'.format(
|
||||
endpoint=endpoint,
|
||||
container=container_name),
|
||||
headers={
|
||||
'Content-Length': '0',
|
||||
'X-Container-Object-Count': '0',
|
||||
'Accept-Ranges': 'bytes',
|
||||
'X-Storage-Policy': 'Policy-0',
|
||||
'Date': 'Fri, 16 Dec 2016 18:29:05 GMT',
|
||||
'X-Timestamp': '1481912480.41664',
|
||||
'X-Trans-Id': 'tx60ec128d9dbf44b9add68-0058543271dfw1',
|
||||
'X-Container-Bytes-Used': '0',
|
||||
'Content-Type': 'text/plain; charset=utf-8'})
|
||||
|
||||
self.register_uri(
|
||||
'HEAD', '{endpoint}/{container}/{object}'.format(
|
||||
endpoint=endpoint,
|
||||
container=container_name, object=image_name),
|
||||
status_code=404)
|
||||
|
||||
self.register_uri(
|
||||
'PUT', '{endpoint}/{container}/{object}'.format(
|
||||
endpoint=endpoint,
|
||||
container=container_name, object=image_name),
|
||||
status_code=201,
|
||||
validate=dict(
|
||||
headers={
|
||||
'x-object-meta-x-shade-md5': NO_MD5,
|
||||
'x-object-meta-x-shade-sha256': NO_SHA256,
|
||||
}))
|
||||
|
||||
task_id = str(uuid.uuid4())
|
||||
args = dict(
|
||||
id=task_id,
|
||||
@ -294,61 +234,92 @@ class TestImage(BaseTestImage):
|
||||
del(image_no_checksums['owner_specified.shade.sha256'])
|
||||
del(image_no_checksums['owner_specified.shade.object'])
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json={'images': []})
|
||||
|
||||
self.register_uri(
|
||||
'POST', 'https://image.example.com/v2/tasks',
|
||||
json=args,
|
||||
validate=dict(
|
||||
json=dict(
|
||||
type='import', input={
|
||||
'import_from': '{container}/{object}'.format(
|
||||
container=container_name, object=image_name),
|
||||
'image_properties': {'name': image_name}})))
|
||||
|
||||
self.register_uri(
|
||||
'GET',
|
||||
'https://image.example.com/v2/tasks/{id}'.format(id=task_id),
|
||||
status_code=503, text='Random error')
|
||||
|
||||
self.register_uri(
|
||||
'GET',
|
||||
'https://image.example.com/v2/tasks/{id}'.format(id=task_id),
|
||||
json={'images': args})
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json={'images': [image_no_checksums]})
|
||||
|
||||
self.register_uri(
|
||||
'PATCH',
|
||||
'https://image.example.com/v2/images/{id}'.format(
|
||||
id=self.image_id),
|
||||
validate=dict(
|
||||
json=sorted([
|
||||
{
|
||||
u'op': u'add',
|
||||
u'value': '{container}/{object}'.format(
|
||||
container=container_name, object=image_name),
|
||||
u'path': u'/owner_specified.shade.object'
|
||||
}, {
|
||||
u'op': u'add',
|
||||
u'value': NO_MD5,
|
||||
u'path': u'/owner_specified.shade.md5'
|
||||
}, {
|
||||
u'op': u'add', u'value': NO_SHA256,
|
||||
u'path': u'/owner_specified.shade.sha256'
|
||||
}], key=operator.itemgetter('value')),
|
||||
headers={
|
||||
'Content-Type':
|
||||
'application/openstack-images-v2.1-json-patch'
|
||||
}))
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
self.register_uris([
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json={'images': []}),
|
||||
dict(method='GET', uri='https://object-store.example.com/info',
|
||||
json=dict(
|
||||
swift={'max_file_size': 1000},
|
||||
slo={'min_segment_size': 500})),
|
||||
dict(method='HEAD',
|
||||
uri='{endpoint}/{container}'.format(
|
||||
endpoint=endpoint, container=container_name),
|
||||
status_code=404),
|
||||
dict(method='PUT',
|
||||
uri='{endpoint}/{container}'.format(
|
||||
endpoint=endpoint, container=container_name),
|
||||
status_code=201,
|
||||
headers={'Date': 'Fri, 16 Dec 2016 18:21:20 GMT',
|
||||
'Content-Length': '0',
|
||||
'Content-Type': 'text/html; charset=UTF-8'}),
|
||||
dict(method='HEAD',
|
||||
uri='{endpoint}/{container}'.format(
|
||||
endpoint=endpoint, container=container_name),
|
||||
headers={'Content-Length': '0',
|
||||
'X-Container-Object-Count': '0',
|
||||
'Accept-Ranges': 'bytes',
|
||||
'X-Storage-Policy': 'Policy-0',
|
||||
'Date': 'Fri, 16 Dec 2016 18:29:05 GMT',
|
||||
'X-Timestamp': '1481912480.41664',
|
||||
'X-Trans-Id': 'tx60ec128d9dbf44b9add68-0058543271dfw1', # noqa
|
||||
'X-Container-Bytes-Used': '0',
|
||||
'Content-Type': 'text/plain; charset=utf-8'}),
|
||||
dict(method='HEAD',
|
||||
uri='{endpoint}/{container}/{object}'.format(
|
||||
endpoint=endpoint, container=container_name,
|
||||
object=image_name),
|
||||
status_code=404),
|
||||
dict(method='PUT',
|
||||
uri='{endpoint}/{container}/{object}'.format(
|
||||
endpoint=endpoint, container=container_name,
|
||||
object=image_name),
|
||||
status_code=201,
|
||||
validate=dict(
|
||||
headers={'x-object-meta-x-shade-md5': NO_MD5,
|
||||
'x-object-meta-x-shade-sha256': NO_SHA256})
|
||||
),
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json={'images': []}),
|
||||
dict(method='POST', uri='https://image.example.com/v2/tasks',
|
||||
json=args,
|
||||
validate=dict(
|
||||
json=dict(
|
||||
type='import', input={
|
||||
'import_from': '{container}/{object}'.format(
|
||||
container=container_name, object=image_name),
|
||||
'image_properties': {'name': image_name}}))
|
||||
),
|
||||
dict(method='GET',
|
||||
uri='https://image.example.com/v2/tasks/{id}'.format(
|
||||
id=task_id),
|
||||
status_code=503, text='Random error'),
|
||||
dict(method='GET',
|
||||
uri='https://image.example.com/v2/tasks/{id}'.format(
|
||||
id=task_id),
|
||||
json={'images': args}),
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json={'images': [image_no_checksums]}),
|
||||
dict(method='PATCH',
|
||||
uri='https://image.example.com/v2/images/{id}'.format(
|
||||
id=self.image_id),
|
||||
validate=dict(
|
||||
json=sorted([{u'op': u'add',
|
||||
u'value': '{container}/{object}'.format(
|
||||
container=container_name,
|
||||
object=image_name),
|
||||
u'path': u'/owner_specified.shade.object'},
|
||||
{u'op': u'add', u'value': NO_MD5,
|
||||
u'path': u'/owner_specified.shade.md5'},
|
||||
{u'op': u'add', u'value': NO_SHA256,
|
||||
u'path': u'/owner_specified.shade.sha256'}],
|
||||
key=operator.itemgetter('value')),
|
||||
headers={
|
||||
'Content-Type':
|
||||
'application/openstack-images-v2.1-json-patch'})
|
||||
),
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
])
|
||||
|
||||
self.cloud.create_image(
|
||||
image_name, self.imagefile.name, wait=True, timeout=1,
|
||||
@ -719,9 +690,11 @@ class TestImageVersionDiscovery(BaseTestImage):
|
||||
self.cloud.cloud_config.config['image_endpoint_override'] = \
|
||||
'https://image.example.com/v2/override'
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/override/images',
|
||||
json={'images': []})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://image.example.com/v2/override/images',
|
||||
json={'images': []})
|
||||
])
|
||||
self.assertEqual([], self.cloud.list_images())
|
||||
self.assertEqual(
|
||||
self.cloud._image_client.endpoint_override,
|
||||
@ -735,22 +708,26 @@ class TestImageVolume(BaseTestImage):
|
||||
|
||||
volume_id = 'some-volume'
|
||||
|
||||
self.register_uri(
|
||||
'POST', '{endpoint}/volumes/{id}/action'.format(
|
||||
endpoint=CINDER_URL, id=volume_id),
|
||||
json={'os-volume_upload_image': {'image_id': self.image_id}},
|
||||
validate=dict(json={
|
||||
u'os-volume_upload_image': {
|
||||
u'container_format': u'bare',
|
||||
u'disk_format': u'qcow2',
|
||||
u'force': False,
|
||||
u'image_name': u'fake_image'}}))
|
||||
|
||||
self.use_glance()
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
self.register_uris([
|
||||
dict(method='POST',
|
||||
uri='{endpoint}/volumes/{id}/action'.format(
|
||||
endpoint=CINDER_URL, id=volume_id),
|
||||
json={'os-volume_upload_image': {'image_id': self.image_id}},
|
||||
validate=dict(json={
|
||||
u'os-volume_upload_image': {
|
||||
u'container_format': u'bare',
|
||||
u'disk_format': u'qcow2',
|
||||
u'force': False,
|
||||
u'image_name': u'fake_image'}})
|
||||
),
|
||||
# NOTE(notmorgan): Glance discovery happens here, insert the
|
||||
# glance discovery mock at this point, DO NOT use the
|
||||
# .use_glance() method, that is intended only for use in
|
||||
# .setUp
|
||||
self.get_glance_discovery_mock_dict(),
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
])
|
||||
|
||||
self.cloud.create_image(
|
||||
'fake_image', self.imagefile.name, wait=True, timeout=1,
|
||||
@ -762,22 +739,26 @@ class TestImageVolume(BaseTestImage):
|
||||
|
||||
volume_id = 'some-volume'
|
||||
|
||||
self.register_uri(
|
||||
'POST', '{endpoint}/volumes/{id}/action'.format(
|
||||
endpoint=CINDER_URL, id=volume_id),
|
||||
json={'os-volume_upload_image': {'image_id': self.image_id}},
|
||||
validate=dict(json={
|
||||
u'os-volume_upload_image': {
|
||||
u'container_format': u'bare',
|
||||
u'disk_format': u'qcow2',
|
||||
u'force': True,
|
||||
u'image_name': u'fake_image'}}))
|
||||
|
||||
self.use_glance()
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
self.register_uris([
|
||||
dict(method='POST',
|
||||
uri='{endpoint}/volumes/{id}/action'.format(
|
||||
endpoint=CINDER_URL, id=volume_id),
|
||||
json={'os-volume_upload_image': {'image_id': self.image_id}},
|
||||
validate=dict(json={
|
||||
u'os-volume_upload_image': {
|
||||
u'container_format': u'bare',
|
||||
u'disk_format': u'qcow2',
|
||||
u'force': True,
|
||||
u'image_name': u'fake_image'}})
|
||||
),
|
||||
# NOTE(notmorgan): Glance discovery happens here, insert the
|
||||
# glance discovery mock at this point, DO NOT use the
|
||||
# .use_glance() method, that is intended only for use in
|
||||
# .setUp
|
||||
self.get_glance_discovery_mock_dict(),
|
||||
dict(method='GET', uri='https://image.example.com/v2/images',
|
||||
json=self.fake_search_return)
|
||||
])
|
||||
|
||||
self.cloud.create_image(
|
||||
'fake_image', self.imagefile.name, wait=True, timeout=1,
|
||||
|
||||
@ -274,15 +274,17 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
PUBLIC_V4, meta.get_server_ip(srv, ext_tag='floating'))
|
||||
|
||||
def test_get_server_private_ip(self):
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net-name'
|
||||
}]})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net-name'}]}
|
||||
),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
])
|
||||
|
||||
srv = meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -310,43 +312,39 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
mock_get_flavor_name.return_value = 'm1.tiny'
|
||||
mock_get_volumes.return_value = []
|
||||
|
||||
self.register_uri(
|
||||
'GET',
|
||||
'https://network.example.com/v2.0/ports.json?device_id=test-id',
|
||||
json={'ports': [{
|
||||
'id': 'test_port_id',
|
||||
'mac_address': 'fa:16:3e:ae:7d:42',
|
||||
'device_id': 'test-id',
|
||||
}]})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=('https://network.example.com/v2.0/ports.json?'
|
||||
'device_id=test-id'),
|
||||
json={'ports': [{
|
||||
'id': 'test_port_id',
|
||||
'mac_address': 'fa:16:3e:ae:7d:42',
|
||||
'device_id': 'test-id'}]}
|
||||
),
|
||||
dict(method='GET',
|
||||
uri=('https://network.example.com/v2.0/'
|
||||
'floatingips.json?port_id=test_port_id'),
|
||||
json={'floatingips': []}),
|
||||
|
||||
self.register_uri(
|
||||
'GET',
|
||||
'https://network.example.com/v2.0/floatingips.json'
|
||||
'?port_id=test_port_id',
|
||||
json={'floatingips': []})
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [
|
||||
{'id': 'test_pnztt_net',
|
||||
'name': 'test_pnztt_net',
|
||||
'router:external': False
|
||||
},
|
||||
{'id': 'private',
|
||||
'name': 'private'}]}
|
||||
),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT}),
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [
|
||||
{
|
||||
'id': 'test_pnztt_net',
|
||||
'name': 'test_pnztt_net',
|
||||
'router:external': False,
|
||||
},
|
||||
{
|
||||
'id': 'private',
|
||||
'name': 'private',
|
||||
}
|
||||
]})
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
|
||||
self.register_uri(
|
||||
'GET', '{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
dict(method='GET',
|
||||
uri='{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
])
|
||||
|
||||
srv = self.cloud.get_openstack_vars(meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -378,26 +376,25 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
mock_get_flavor_name.return_value = 'm1.tiny'
|
||||
mock_get_volumes.return_value = []
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [
|
||||
{
|
||||
'id': 'test_pnztt_net',
|
||||
'name': 'test_pnztt_net',
|
||||
'router:external': False,
|
||||
},
|
||||
{
|
||||
'id': 'private',
|
||||
'name': 'private',
|
||||
}
|
||||
]})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
self.register_uri(
|
||||
'GET', '{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [
|
||||
{'id': 'test_pnztt_net',
|
||||
'name': 'test_pnztt_net',
|
||||
'router:external': False,
|
||||
},
|
||||
{'id': 'private',
|
||||
'name': 'private'}]}
|
||||
),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT}),
|
||||
dict(method='GET',
|
||||
uri='{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
])
|
||||
|
||||
srv = self.cloud.get_openstack_vars(meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -427,26 +424,27 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
mock_get_image_name.return_value = 'cirros-0.3.4-x86_64-uec'
|
||||
mock_get_flavor_name.return_value = 'm1.tiny'
|
||||
mock_get_volumes.return_value = []
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [
|
||||
{
|
||||
'id': 'test_pnztt_net',
|
||||
'name': 'test_pnztt_net',
|
||||
'router:external': False,
|
||||
},
|
||||
{
|
||||
'id': 'private',
|
||||
'name': 'private',
|
||||
}
|
||||
]})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
self.register_uri(
|
||||
'GET', '{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [
|
||||
{
|
||||
'id': 'test_pnztt_net',
|
||||
'name': 'test_pnztt_net',
|
||||
'router:external': False,
|
||||
},
|
||||
{
|
||||
'id': 'private',
|
||||
'name': 'private'}]}
|
||||
),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT}),
|
||||
dict(method='GET',
|
||||
uri='{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
])
|
||||
|
||||
srv = self.cloud.get_openstack_vars(meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -474,47 +472,45 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
mock_get_flavor_name.return_value = 'm1.tiny'
|
||||
mock_get_volumes.return_value = []
|
||||
|
||||
self.register_uri(
|
||||
'GET',
|
||||
'https://network.example.com/v2.0/ports.json?device_id=test-id',
|
||||
json={'ports': [{
|
||||
'id': 'test_port_id',
|
||||
'mac_address': 'fa:16:3e:ae:7d:42',
|
||||
'device_id': 'test-id',
|
||||
}]})
|
||||
|
||||
self.register_uri(
|
||||
'GET',
|
||||
'https://network.example.com/v2.0/floatingips.json'
|
||||
'?port_id=test_port_id',
|
||||
json={'floatingips': [{
|
||||
'id': 'floating-ip-id',
|
||||
'port_id': 'test_port_id',
|
||||
'fixed_ip_address': PRIVATE_V4,
|
||||
'floating_ip_address': PUBLIC_V4,
|
||||
}]})
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [
|
||||
{
|
||||
'id': 'test_pnztt_net',
|
||||
'name': 'test_pnztt_net',
|
||||
'router:external': False,
|
||||
},
|
||||
{
|
||||
'id': 'private',
|
||||
'name': 'private',
|
||||
}
|
||||
]})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
|
||||
self.register_uri(
|
||||
'GET', '{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=('https://network.example.com/v2.0/ports.json?'
|
||||
'device_id=test-id'),
|
||||
json={'ports': [{
|
||||
'id': 'test_port_id',
|
||||
'mac_address': 'fa:16:3e:ae:7d:42',
|
||||
'device_id': 'test-id'}]}
|
||||
),
|
||||
dict(method='GET',
|
||||
uri=('https://network.example.com/v2.0/floatingips.json'
|
||||
'?port_id=test_port_id'),
|
||||
json={'floatingips': [{
|
||||
'id': 'floating-ip-id',
|
||||
'port_id': 'test_port_id',
|
||||
'fixed_ip_address': PRIVATE_V4,
|
||||
'floating_ip_address': PUBLIC_V4,
|
||||
}]}),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [
|
||||
{
|
||||
'id': 'test_pnztt_net',
|
||||
'name': 'test_pnztt_net',
|
||||
'router:external': False,
|
||||
},
|
||||
{
|
||||
'id': 'private',
|
||||
'name': 'private',
|
||||
}
|
||||
]}),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT}),
|
||||
dict(method='GET',
|
||||
uri='{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
])
|
||||
|
||||
srv = self.cloud.get_openstack_vars(meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -546,10 +542,12 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
mock_get_flavor_name.return_value = 'm1.tiny'
|
||||
mock_get_volumes.return_value = []
|
||||
|
||||
self.register_uri(
|
||||
'GET', '{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
])
|
||||
|
||||
srv = self.cloud.get_openstack_vars(meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -597,16 +595,18 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
mock_get_flavor_name.return_value = 'm1.tiny'
|
||||
mock_get_volumes.return_value = []
|
||||
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': OSIC_NETWORKS})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': OSIC_SUBNETS})
|
||||
self.register_uri(
|
||||
'GET', '{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': OSIC_NETWORKS}),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': OSIC_SUBNETS}),
|
||||
dict(method='GET',
|
||||
uri='{endpoint}/servers/test-id/os-security-groups'.format(
|
||||
endpoint=fakes.COMPUTE_ENDPOINT),
|
||||
json={'security_groups': []})
|
||||
])
|
||||
|
||||
srv = self.cloud.get_openstack_vars(meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -639,17 +639,18 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
|
||||
def test_get_server_external_ipv4_neutron(self):
|
||||
# Testing Clouds with Neutron
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net',
|
||||
'router:external': True,
|
||||
}]})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net',
|
||||
'router:external': True
|
||||
}]}),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
])
|
||||
srv = meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
addresses={'test-net': [{
|
||||
@ -663,17 +664,19 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
|
||||
def test_get_server_external_provider_ipv4_neutron(self):
|
||||
# Testing Clouds with Neutron
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net',
|
||||
'provider:network_type': 'vlan',
|
||||
'provider:physical_network': 'vlan',
|
||||
}]})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net',
|
||||
'provider:network_type': 'vlan',
|
||||
'provider:physical_network': 'vlan',
|
||||
}]}),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
])
|
||||
|
||||
srv = meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -688,19 +691,20 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
|
||||
def test_get_server_internal_provider_ipv4_neutron(self):
|
||||
# Testing Clouds with Neutron
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net',
|
||||
'router:external': False,
|
||||
'provider:network_type': 'vxlan',
|
||||
'provider:physical_network': None,
|
||||
}]})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net',
|
||||
'router:external': False,
|
||||
'provider:network_type': 'vxlan',
|
||||
'provider:physical_network': None,
|
||||
}]}),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
])
|
||||
srv = meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
addresses={'test-net': [{
|
||||
@ -716,16 +720,18 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
|
||||
def test_get_server_external_none_ipv4_neutron(self):
|
||||
# Testing Clouds with Neutron
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net',
|
||||
'router:external': False,
|
||||
}]})
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
json={'networks': [{
|
||||
'id': 'test-net-id',
|
||||
'name': 'test-net',
|
||||
'router:external': False,
|
||||
}]}),
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/subnets.json',
|
||||
json={'subnets': SUBNETS_WITH_NAT})
|
||||
])
|
||||
|
||||
srv = meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
@ -756,9 +762,10 @@ class TestMeta(base.RequestsMockTestCase):
|
||||
|
||||
def test_get_server_external_ipv4_neutron_exception(self):
|
||||
# Testing Clouds with a non working Neutron
|
||||
self.register_uri(
|
||||
'GET', 'https://network.example.com/v2.0/networks.json',
|
||||
status_code=404)
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri='https://network.example.com/v2.0/networks.json',
|
||||
status_code=404)])
|
||||
|
||||
srv = meta.obj_to_dict(fakes.FakeServer(
|
||||
id='test-id', name='test-name', status='ACTIVE',
|
||||
|
||||
@ -36,17 +36,16 @@ class TestProject(base.RequestsMockTestCase):
|
||||
def test_create_project_v2(self):
|
||||
self.use_keystone_v2()
|
||||
project_data = self._get_project_data(v3=False)
|
||||
self.register_uri(
|
||||
'POST',
|
||||
self.get_mock_url(v3=False),
|
||||
status_code=200,
|
||||
json=project_data.json_response,
|
||||
validate=project_data.json_request)
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(v3=False, append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response)
|
||||
self.register_uris([
|
||||
dict(method='POST', uri=self.get_mock_url(v3=False),
|
||||
status_code=200, json=project_data.json_response,
|
||||
validate=project_data.json_request),
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
v3=False, append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response)
|
||||
])
|
||||
project = self.op_cloud.create_project(
|
||||
name=project_data.project_name,
|
||||
description=project_data.description)
|
||||
@ -59,17 +58,17 @@ class TestProject(base.RequestsMockTestCase):
|
||||
self._add_discovery_uri_call()
|
||||
project_data = self._get_project_data(
|
||||
description=self.getUniqueString('projectDesc'))
|
||||
self.register_uri(
|
||||
'POST',
|
||||
self.get_mock_url(),
|
||||
status_code=200,
|
||||
json=project_data.json_response,
|
||||
validate=project_data.json_request)
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response)
|
||||
self.register_uris([
|
||||
dict(method='POST',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json=project_data.json_response,
|
||||
validate=project_data.json_request),
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response)
|
||||
])
|
||||
project = self.op_cloud.create_project(
|
||||
name=project_data.project_name,
|
||||
description=project_data.description,
|
||||
@ -94,41 +93,43 @@ class TestProject(base.RequestsMockTestCase):
|
||||
def test_delete_project_v2(self):
|
||||
self.use_keystone_v2()
|
||||
project_data = self._get_project_data(v3=False)
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(v3=False),
|
||||
status_code=200,
|
||||
json={'tenants': [project_data.json_response['tenant']]})
|
||||
self.register_uri(
|
||||
'DELETE',
|
||||
self.get_mock_url(v3=False, append=[project_data.project_id]),
|
||||
status_code=204)
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(v3=False),
|
||||
status_code=200,
|
||||
json={'tenants': [project_data.json_response['tenant']]}),
|
||||
dict(method='DELETE',
|
||||
uri=self.get_mock_url(
|
||||
v3=False, append=[project_data.project_id]),
|
||||
status_code=204)
|
||||
])
|
||||
self.op_cloud.delete_project(project_data.project_id)
|
||||
self.assert_calls()
|
||||
|
||||
def test_delete_project_v3(self):
|
||||
self._add_discovery_uri_call()
|
||||
project_data = self._get_project_data(v3=False)
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['tenant']]})
|
||||
self.register_uri(
|
||||
'DELETE',
|
||||
self.get_mock_url(append=[project_data.project_id]),
|
||||
status_code=204)
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['tenant']]}),
|
||||
dict(method='DELETE',
|
||||
uri=self.get_mock_url(append=[project_data.project_id]),
|
||||
status_code=204)
|
||||
])
|
||||
self.op_cloud.delete_project(project_data.project_id)
|
||||
self.assert_calls()
|
||||
|
||||
def test_update_project_not_found(self):
|
||||
self._add_discovery_uri_call()
|
||||
project_data = self._get_project_data()
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'projects': []})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'projects': []})
|
||||
])
|
||||
# NOTE(notmorgan): This test (and shade) does not represent a case
|
||||
# where the project is in the project list but a 404 is raised when
|
||||
# the PATCH is issued. This is a bug in shade and should be fixed,
|
||||
@ -146,22 +147,23 @@ class TestProject(base.RequestsMockTestCase):
|
||||
project_data = self._get_project_data(
|
||||
v3=False,
|
||||
description=self.getUniqueString('projectDesc'))
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(v3=False),
|
||||
status_code=200,
|
||||
json={'tenants': [project_data.json_response['tenant']]})
|
||||
self.register_uri(
|
||||
'POST',
|
||||
self.get_mock_url(v3=False, append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response,
|
||||
validate=project_data.json_request)
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(v3=False, append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response)
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(v3=False),
|
||||
status_code=200,
|
||||
json={'tenants': [project_data.json_response['tenant']]}),
|
||||
dict(method='POST',
|
||||
uri=self.get_mock_url(
|
||||
v3=False, append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response,
|
||||
validate=project_data.json_request),
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
v3=False, append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response)
|
||||
])
|
||||
project = self.op_cloud.update_project(
|
||||
project_data.project_id,
|
||||
description=project_data.description)
|
||||
@ -176,23 +178,21 @@ class TestProject(base.RequestsMockTestCase):
|
||||
self._add_discovery_uri_call()
|
||||
project_data = self._get_project_data(
|
||||
description=self.getUniqueString('projectDesc'))
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(
|
||||
resource='projects?domain_id=%s' % project_data.domain_id),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
self.register_uri(
|
||||
'PATCH',
|
||||
self.get_mock_url(append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response,
|
||||
validate=project_data.json_request)
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(append=[project_data.project_id]),
|
||||
status_code=200,
|
||||
json=project_data.json_response)
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
resource=('projects?domain_id=%s' %
|
||||
project_data.domain_id)),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]}),
|
||||
dict(method='PATCH',
|
||||
uri=self.get_mock_url(append=[project_data.project_id]),
|
||||
status_code=200, json=project_data.json_response,
|
||||
validate=project_data.json_request),
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(append=[project_data.project_id]),
|
||||
status_code=200, json=project_data.json_response)
|
||||
])
|
||||
project = self.op_cloud.update_project(
|
||||
project_data.project_id,
|
||||
description=project_data.description,
|
||||
@ -208,12 +208,14 @@ class TestProject(base.RequestsMockTestCase):
|
||||
self._add_discovery_uri_call()
|
||||
project_data = self._get_project_data(
|
||||
description=self.getUniqueString('projectDesc'))
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(
|
||||
resource='projects?domain_id=%s' % project_data.domain_id),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
resource=('projects?domain_id=%s' %
|
||||
project_data.domain_id)),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
])
|
||||
projects = self.op_cloud.list_projects(project_data.domain_id)
|
||||
self.assertThat(len(projects), matchers.Equals(1))
|
||||
self.assertThat(
|
||||
@ -224,12 +226,14 @@ class TestProject(base.RequestsMockTestCase):
|
||||
self._add_discovery_uri_call()
|
||||
project_data = self._get_project_data(
|
||||
description=self.getUniqueString('projectDesc'))
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(
|
||||
resource='projects?domain_id=%s' % project_data.domain_id),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
resource=('projects?domain_id=%s' %
|
||||
project_data.domain_id)),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
])
|
||||
projects = self.op_cloud.list_projects(
|
||||
domain_id=project_data.domain_id)
|
||||
self.assertThat(len(projects), matchers.Equals(1))
|
||||
@ -241,11 +245,12 @@ class TestProject(base.RequestsMockTestCase):
|
||||
self._add_discovery_uri_call()
|
||||
project_data = self._get_project_data(
|
||||
description=self.getUniqueString('projectDesc'))
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
])
|
||||
projects = self.op_cloud.search_projects(project_data.project_id)
|
||||
self.assertThat(len(projects), matchers.Equals(1))
|
||||
self.assertThat(
|
||||
@ -256,12 +261,14 @@ class TestProject(base.RequestsMockTestCase):
|
||||
self._add_discovery_uri_call()
|
||||
project_data = self._get_project_data(
|
||||
description=self.getUniqueString('projectDesc'))
|
||||
self.register_uri(
|
||||
'GET',
|
||||
self.get_mock_url(
|
||||
resource='projects?domain_id=%s' % project_data.domain_id),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
self.register_uris([
|
||||
dict(method='GET',
|
||||
uri=self.get_mock_url(
|
||||
resource=('projects?domain_id=%s' %
|
||||
project_data.domain_id)),
|
||||
status_code=200,
|
||||
json={'projects': [project_data.json_response['project']]})
|
||||
])
|
||||
projects = self.op_cloud.search_projects(
|
||||
domain_id=project_data.domain_id)
|
||||
self.assertThat(len(projects), matchers.Equals(1))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user