Merge "Functional tests migration for metadef APIs"
This commit is contained in:
Zuul
@@ -12,12 +12,11 @@
|
||||
# limitations under the License.
|
||||
|
||||
from oslo_utils.fixture import uuidsentinel as uuids
|
||||
import requests
|
||||
|
||||
from glance.tests import functional
|
||||
|
||||
|
||||
class MetadefFunctionalTestBase(functional.FunctionalTest):
|
||||
class MetadefFunctionalTestBase(functional.SynchronousAPIBase):
|
||||
"""A basic set of assertions and utilities for testing the metadef API."""
|
||||
|
||||
def setUp(self):
|
||||
@@ -61,4 +60,5 @@ class MetadefFunctionalTestBase(functional.FunctionalTest):
|
||||
|
||||
:returns: a dictionary of the namespace in the response
|
||||
"""
|
||||
return requests.post(path, headers=headers, json=namespace).json()
|
||||
|
||||
return self.api_post(path, headers, json=namespace)
|
||||
|
||||
@@ -16,7 +16,6 @@
|
||||
import http.client as http
|
||||
|
||||
from oslo_serialization import jsonutils
|
||||
import requests
|
||||
|
||||
from glance.tests.functional.v2 import metadef_base
|
||||
|
||||
@@ -25,27 +24,24 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestNamespaces, self).setUp()
|
||||
self.cleanup()
|
||||
self.api_server.deployment_flavor = 'noauth'
|
||||
self.start_servers(**self.__dict__.copy())
|
||||
self.start_server(enable_cache=False)
|
||||
|
||||
def test_namespace_lifecycle(self):
|
||||
# Namespace should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/MyNamespace'
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create a namespace
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
namespace_name = 'MyNamespace'
|
||||
data = jsonutils.dumps({
|
||||
data = {
|
||||
"namespace": namespace_name,
|
||||
"display_name": "My User Friendly Namespace",
|
||||
"description": "My description"
|
||||
}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
namespace_loc_header = response.headers['Location']
|
||||
|
||||
@@ -80,11 +76,11 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(namespace[key], value, key)
|
||||
|
||||
# Attempt to insert a duplicate
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CONFLICT, response.status_code)
|
||||
|
||||
# Get the namespace using the returned Location header
|
||||
response = requests.get(namespace_loc_header, headers=self._headers())
|
||||
response = self.api_get(namespace_loc_header, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
namespace = jsonutils.loads(response.text)
|
||||
self.assertEqual(namespace_name, namespace['namespace'])
|
||||
@@ -94,21 +90,19 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertFalse(namespace['protected'])
|
||||
|
||||
# The namespace should be mutable
|
||||
path = self._url('/v2/metadefs/namespaces/%s' % namespace_name)
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace_name
|
||||
media_type = 'application/json'
|
||||
headers = self._headers({'content-type': media_type})
|
||||
namespace_name = "MyNamespace-UPDATED"
|
||||
data = jsonutils.dumps(
|
||||
{
|
||||
"namespace": namespace_name,
|
||||
"display_name": "display_name-UPDATED",
|
||||
"description": "description-UPDATED",
|
||||
"visibility": "private", # Not changed
|
||||
"protected": True,
|
||||
"owner": self.tenant2
|
||||
}
|
||||
)
|
||||
response = requests.put(path, headers=headers, data=data)
|
||||
data = {
|
||||
"namespace": namespace_name,
|
||||
"display_name": "display_name-UPDATED",
|
||||
"description": "description-UPDATED",
|
||||
"visibility": "private", # Not changed
|
||||
"protected": True,
|
||||
"owner": self.tenant2
|
||||
}
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
|
||||
# Returned namespace should reflect the changes
|
||||
@@ -121,8 +115,8 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(self.tenant2, namespace['owner'])
|
||||
|
||||
# Updates should persist across requests
|
||||
path = self._url('/v2/metadefs/namespaces/%s' % namespace_name)
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace_name
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
namespace = jsonutils.loads(response.text)
|
||||
self.assertEqual('MyNamespace-UPDATED', namespace['namespace'])
|
||||
@@ -133,12 +127,12 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(self.tenant2, namespace['owner'])
|
||||
|
||||
# Deletion should not work on protected namespaces
|
||||
path = self._url('/v2/metadefs/namespaces/%s' % namespace_name)
|
||||
response = requests.delete(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace_name
|
||||
response = self.api_delete(path, headers=self._headers())
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Unprotect namespace for deletion
|
||||
path = self._url('/v2/metadefs/namespaces/%s' % namespace_name)
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace_name
|
||||
media_type = 'application/json'
|
||||
headers = self._headers({'content-type': media_type})
|
||||
doc = {
|
||||
@@ -149,87 +143,26 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
"protected": False,
|
||||
"owner": self.tenant2
|
||||
}
|
||||
data = jsonutils.dumps(doc)
|
||||
response = requests.put(path, headers=headers, data=data)
|
||||
response = self.api_put(path, headers=headers, json=doc)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
|
||||
# Deletion should work. Deleting namespace MyNamespace
|
||||
path = self._url('/v2/metadefs/namespaces/%s' % namespace_name)
|
||||
response = requests.delete(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace_name
|
||||
response = self.api_delete(path, headers=self._headers())
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# Namespace should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/MyNamespace'
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
def test_metadef_dont_accept_illegal_bodies(self):
|
||||
# Namespace should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/bodytest')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create a namespace
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
namespace_name = 'bodytest'
|
||||
data = jsonutils.dumps({
|
||||
"namespace": namespace_name,
|
||||
"display_name": "My User Friendly Namespace",
|
||||
"description": "My description"
|
||||
}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Test all the urls that supply data
|
||||
data_urls = [
|
||||
'/v2/schemas/metadefs/namespace',
|
||||
'/v2/schemas/metadefs/namespaces',
|
||||
'/v2/schemas/metadefs/resource_type',
|
||||
'/v2/schemas/metadefs/resource_types',
|
||||
'/v2/schemas/metadefs/property',
|
||||
'/v2/schemas/metadefs/properties',
|
||||
'/v2/schemas/metadefs/object',
|
||||
'/v2/schemas/metadefs/objects',
|
||||
'/v2/schemas/metadefs/tag',
|
||||
'/v2/schemas/metadefs/tags',
|
||||
'/v2/metadefs/resource_types',
|
||||
]
|
||||
for value in data_urls:
|
||||
path = self._url(value)
|
||||
data = jsonutils.dumps(["body"])
|
||||
response = requests.get(path, headers=self._headers(), data=data)
|
||||
self.assertEqual(http.BAD_REQUEST, response.status_code)
|
||||
|
||||
# Put the namespace into the url
|
||||
test_urls = [
|
||||
('/v2/metadefs/namespaces/%s/resource_types', 'get'),
|
||||
('/v2/metadefs/namespaces/%s/resource_types/type', 'delete'),
|
||||
('/v2/metadefs/namespaces/%s', 'get'),
|
||||
('/v2/metadefs/namespaces/%s', 'delete'),
|
||||
('/v2/metadefs/namespaces/%s/objects/name', 'get'),
|
||||
('/v2/metadefs/namespaces/%s/objects/name', 'delete'),
|
||||
('/v2/metadefs/namespaces/%s/properties', 'get'),
|
||||
('/v2/metadefs/namespaces/%s/tags/test', 'get'),
|
||||
('/v2/metadefs/namespaces/%s/tags/test', 'post'),
|
||||
('/v2/metadefs/namespaces/%s/tags/test', 'delete'),
|
||||
]
|
||||
|
||||
for link, method in test_urls:
|
||||
path = self._url(link % namespace_name)
|
||||
data = jsonutils.dumps(["body"])
|
||||
response = getattr(requests, method)(
|
||||
path, headers=self._headers(), data=data)
|
||||
self.assertEqual(http.BAD_REQUEST, response.status_code)
|
||||
|
||||
def _update_namespace(self, path, headers, data):
|
||||
# The namespace should be mutable
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
|
||||
# Returned namespace should reflect the changes
|
||||
namespace = response.json()
|
||||
namespace = jsonutils.loads(response.text)
|
||||
expected_namespace = {
|
||||
"namespace": data['namespace'],
|
||||
"display_name": data['display_name'],
|
||||
@@ -245,10 +178,10 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(namespace, expected_namespace)
|
||||
|
||||
# Updates should persist across requests
|
||||
path = self._url('/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
namespace = response.json()
|
||||
namespace = jsonutils.loads(response.text)
|
||||
namespace.pop('created_at')
|
||||
namespace.pop('updated_at')
|
||||
self.assertEqual(namespace, expected_namespace)
|
||||
@@ -257,7 +190,7 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def test_role_based_namespace_lifecycle(self):
|
||||
# Create public and private namespaces for tenant1 and tenant2
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
tenant_namespaces = dict()
|
||||
for tenant in [self.tenant1, self.tenant2]:
|
||||
@@ -272,6 +205,7 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
}
|
||||
namespace = self.create_namespace(path, headers,
|
||||
namespace_data)
|
||||
namespace = jsonutils.loads(namespace.text)
|
||||
self.assertNamespacesEqual(namespace, namespace_data)
|
||||
tenant_namespaces.setdefault(tenant, list())
|
||||
tenant_namespaces[tenant].append(namespace)
|
||||
@@ -294,12 +228,12 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
# Check Tenant 1 and Tenant 2 will be able to see total 3 namespaces
|
||||
# (two of own and 1 public of other tenant)
|
||||
for tenant in [self.tenant1, self.tenant2]:
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'X-Tenant-Id': tenant,
|
||||
'X-Roles': 'reader,member'})
|
||||
response = requests.get(path, headers=headers)
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
namespaces = response.json()['namespaces']
|
||||
namespaces = jsonutils.loads(response.text)['namespaces']
|
||||
expected_namespaces = _get_expected_namespaces(tenant)
|
||||
self.assertEqual(sorted(x['namespace'] for x in namespaces),
|
||||
sorted(expected_namespaces))
|
||||
@@ -308,10 +242,8 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
headers = self._headers({'X-Tenant-Id': tenant,
|
||||
'X-Roles': 'reader,member'})
|
||||
for namespace in namespaces:
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
headers = headers
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
response = self.api_get(path, headers=headers)
|
||||
if namespace['visibility'] == 'public':
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
else:
|
||||
@@ -338,14 +270,13 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
"protected": True, # changed
|
||||
"owner": namespace["owner"] # Not changed
|
||||
}
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
headers = self._headers({
|
||||
'X-Tenant-Id': namespace['owner'],
|
||||
})
|
||||
# Update namespace should fail with non admin role
|
||||
headers['X-Roles'] = "reader,member"
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Should work with admin role
|
||||
@@ -353,16 +284,14 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
namespace = self._update_namespace(path, headers, data)
|
||||
|
||||
# Deletion should fail as namespaces are protected now
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
headers['X-Roles'] = "admin"
|
||||
response = requests.delete(path, headers=headers)
|
||||
response = self.api_delete(path, headers=headers)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Deletion should not be allowed for non admin roles
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
response = requests.delete(
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
response = self.api_delete(
|
||||
path, headers=self._headers({
|
||||
'X-Roles': 'reader,member',
|
||||
'X-Tenant-Id': namespace['owner']
|
||||
@@ -372,29 +301,26 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
# Unprotect the namespaces before deletion
|
||||
headers = self._headers()
|
||||
for namespace in total_ns:
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
headers = headers
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
data = {
|
||||
"namespace": namespace['namespace'],
|
||||
"protected": False,
|
||||
}
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
|
||||
# Get updated namespace set again
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces'
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
self.assertFalse(namespace['protected'])
|
||||
namespaces = response.json()['namespaces']
|
||||
namespaces = jsonutils.loads(response.text)['namespaces']
|
||||
|
||||
# Verify that deletion is not allowed for unprotected namespaces with
|
||||
# non admin role
|
||||
for namespace in namespaces:
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
response = requests.delete(
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
response = self.api_delete(
|
||||
path, headers=self._headers({
|
||||
'X-Roles': 'reader,member',
|
||||
'X-Tenant-Id': namespace['owner']
|
||||
@@ -403,13 +329,11 @@ class TestNamespaces(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
# Delete namespaces of all tenants
|
||||
for namespace in total_ns:
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
response = requests.delete(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
response = self.api_delete(path, headers=headers)
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# Deleted namespace should not be returned
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s' % namespace['namespace'])
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s' % namespace['namespace']
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
@@ -16,7 +16,6 @@
|
||||
import http.client as http
|
||||
|
||||
from oslo_serialization import jsonutils
|
||||
import requests
|
||||
|
||||
from glance.tests.functional.v2 import metadef_base
|
||||
|
||||
@@ -25,21 +24,19 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestMetadefObjects, self).setUp()
|
||||
self.cleanup()
|
||||
self.api_server.deployment_flavor = 'noauth'
|
||||
self.start_servers(**self.__dict__.copy())
|
||||
self.start_server(enable_cache=False)
|
||||
|
||||
def test_metadata_objects_lifecycle(self):
|
||||
# Namespace should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/MyNamespace'
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create a namespace
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
namespace_name = 'MyNamespace'
|
||||
data = jsonutils.dumps({
|
||||
data = {
|
||||
"namespace": namespace_name,
|
||||
"display_name": "My User Friendly Namespace",
|
||||
"description": "My description",
|
||||
@@ -47,58 +44,55 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
"protected": False,
|
||||
"owner": "The Test Owner"
|
||||
}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Metadata objects should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace/objects/object1')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/MyNamespace/objects/object1'
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create a object
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace/objects')
|
||||
path = '/v2/metadefs/namespaces/MyNamespace/objects'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
metadata_object_name = "object1"
|
||||
data = jsonutils.dumps(
|
||||
{
|
||||
"name": metadata_object_name,
|
||||
"description": "object1 description.",
|
||||
"required": [
|
||||
"property1"
|
||||
],
|
||||
"properties": {
|
||||
"property1": {
|
||||
"type": "integer",
|
||||
"title": "property1",
|
||||
"description": "property1 description",
|
||||
"operators": ["<all-in>"],
|
||||
"default": 100,
|
||||
"minimum": 100,
|
||||
"maximum": 30000369
|
||||
},
|
||||
"property2": {
|
||||
"type": "string",
|
||||
"title": "property2",
|
||||
"description": "property2 description ",
|
||||
"default": "value2",
|
||||
"minLength": 2,
|
||||
"maxLength": 50
|
||||
}
|
||||
data = {
|
||||
"name": metadata_object_name,
|
||||
"description": "object1 description.",
|
||||
"required": [
|
||||
"property1"
|
||||
],
|
||||
"properties": {
|
||||
"property1": {
|
||||
"type": "integer",
|
||||
"title": "property1",
|
||||
"description": "property1 description",
|
||||
"operators": ["<all-in>"],
|
||||
"default": 100,
|
||||
"minimum": 100,
|
||||
"maximum": 30000369
|
||||
},
|
||||
"property2": {
|
||||
"type": "string",
|
||||
"title": "property2",
|
||||
"description": "property2 description ",
|
||||
"default": "value2",
|
||||
"minLength": 2,
|
||||
"maxLength": 50
|
||||
}
|
||||
}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Attempt to insert a duplicate
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CONFLICT, response.status_code)
|
||||
|
||||
# Get the metadata object created above
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace_name, metadata_object_name))
|
||||
response = requests.get(path,
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace_name, metadata_object_name)
|
||||
response = self.api_get(path,
|
||||
headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
metadata_object = jsonutils.loads(response.text)
|
||||
@@ -166,40 +160,38 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
)
|
||||
|
||||
# The metadata_object should be mutable
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace_name, metadata_object_name))
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace_name, metadata_object_name)
|
||||
media_type = 'application/json'
|
||||
headers = self._headers({'content-type': media_type})
|
||||
metadata_object_name = "object1-UPDATED"
|
||||
data = jsonutils.dumps(
|
||||
{
|
||||
"name": metadata_object_name,
|
||||
"description": "desc-UPDATED",
|
||||
"required": [
|
||||
"property2"
|
||||
],
|
||||
"properties": {
|
||||
'property1': {
|
||||
'type': 'integer',
|
||||
"title": "property1",
|
||||
'description': 'p1 desc-UPDATED',
|
||||
'default': 500,
|
||||
'minimum': 500,
|
||||
'maximum': 1369
|
||||
},
|
||||
"property2": {
|
||||
"type": "string",
|
||||
"title": "property2",
|
||||
"description": "p2 desc-UPDATED",
|
||||
'operators': ['<or>'],
|
||||
"default": "value2-UPDATED",
|
||||
"minLength": 5,
|
||||
"maxLength": 150
|
||||
}
|
||||
data = {
|
||||
"name": metadata_object_name,
|
||||
"description": "desc-UPDATED",
|
||||
"required": [
|
||||
"property2"
|
||||
],
|
||||
"properties": {
|
||||
'property1': {
|
||||
'type': 'integer',
|
||||
"title": "property1",
|
||||
'description': 'p1 desc-UPDATED',
|
||||
'default': 500,
|
||||
'minimum': 500,
|
||||
'maximum': 1369
|
||||
},
|
||||
"property2": {
|
||||
"type": "string",
|
||||
"title": "property2",
|
||||
"description": "p2 desc-UPDATED",
|
||||
'operators': ['<or>'],
|
||||
"default": "value2-UPDATED",
|
||||
"minLength": 5,
|
||||
"maxLength": 150
|
||||
}
|
||||
}
|
||||
)
|
||||
response = requests.put(path, headers=headers, data=data)
|
||||
}
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
|
||||
# Returned metadata_object should reflect the changes
|
||||
@@ -222,9 +214,9 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(150, updated_property2['maxLength'])
|
||||
|
||||
# Updates should persist across requests
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace_name, metadata_object_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace_name, metadata_object_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(200, response.status_code)
|
||||
self.assertEqual('object1-UPDATED', metadata_object['name'])
|
||||
self.assertEqual('desc-UPDATED', metadata_object['description'])
|
||||
@@ -244,15 +236,15 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(150, updated_property2['maxLength'])
|
||||
|
||||
# Deletion of metadata_object object1
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace_name, metadata_object_name))
|
||||
response = requests.delete(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace_name, metadata_object_name)
|
||||
response = self.api_delete(path, headers=self._headers())
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# metadata_object object1 should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace_name, metadata_object_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace_name, metadata_object_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
def _create_object(self, namespaces):
|
||||
@@ -273,11 +265,11 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
},
|
||||
}
|
||||
}
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects' %
|
||||
namespace['namespace'])
|
||||
response = requests.post(path, headers=headers, json=data)
|
||||
path = ('/v2/metadefs/namespaces/%s/'
|
||||
'objects') % namespace['namespace']
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
obj_metadata = response.json()
|
||||
obj_metadata = jsonutils.loads(response.text)
|
||||
metadef_objects = dict()
|
||||
metadef_objects[namespace['namespace']] = obj_metadata['name']
|
||||
objects.append(metadef_objects)
|
||||
@@ -285,7 +277,7 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
return objects
|
||||
|
||||
def _update_object(self, path, headers, data, namespace):
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
|
||||
expected_object = {
|
||||
@@ -298,21 +290,21 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
data['name'])
|
||||
}
|
||||
# Returned metadata_object should reflect the changes
|
||||
metadata_object = response.json()
|
||||
metadata_object = jsonutils.loads(response.text)
|
||||
metadata_object.pop('created_at')
|
||||
metadata_object.pop('updated_at')
|
||||
self.assertEqual(metadata_object, expected_object)
|
||||
|
||||
# Updates should persist across requests
|
||||
response = requests.get(path, headers=self._headers())
|
||||
metadata_object = response.json()
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
metadata_object = jsonutils.loads(response.text)
|
||||
metadata_object.pop('created_at')
|
||||
metadata_object.pop('updated_at')
|
||||
self.assertEqual(metadata_object, expected_object)
|
||||
|
||||
def test_role_base_metadata_objects_lifecycle(self):
|
||||
# Create public and private namespaces for tenant1 and tenant2
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
tenant1_namespaces = []
|
||||
tenant2_namespaces = []
|
||||
@@ -328,6 +320,7 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
}
|
||||
namespace = self.create_namespace(path, headers,
|
||||
namespace_data)
|
||||
namespace = jsonutils.loads(namespace.text)
|
||||
self.assertNamespacesEqual(namespace, namespace_data)
|
||||
if tenant == self.tenant1:
|
||||
tenant1_namespaces.append(namespace)
|
||||
@@ -344,10 +337,9 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
'X-Roles': 'reader,member'})
|
||||
for obj in objects:
|
||||
for namespace, object_name in obj.items():
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace, object_name))
|
||||
headers = headers
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace, object_name)
|
||||
response = self.api_get(path, headers=headers)
|
||||
if namespace.split('_')[1] == 'public':
|
||||
expected = http.OK
|
||||
else:
|
||||
@@ -355,12 +347,11 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
self.assertEqual(expected, response.status_code)
|
||||
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s/objects' % namespace)
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/objects' % namespace
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(expected, response.status_code)
|
||||
if expected == http.OK:
|
||||
resp_objs = response.json()['objects']
|
||||
resp_objs = jsonutils.loads(response.text)['objects']
|
||||
self.assertEqual(
|
||||
sorted(obj.values()),
|
||||
sorted([x['name'] for x in resp_objs]))
|
||||
@@ -392,10 +383,10 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
}
|
||||
}
|
||||
# Update object should fail with non admin role
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace, object_name))
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace, object_name)
|
||||
headers['X-Roles'] = "reader,member"
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Should work with admin role
|
||||
@@ -406,9 +397,9 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
# Delete object should not be allowed to non admin role
|
||||
for obj in total_objects:
|
||||
for namespace, object_name in obj.items():
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace, object_name))
|
||||
response = requests.delete(
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace, object_name)
|
||||
response = self.api_delete(
|
||||
path, headers=self._headers({
|
||||
'X-Roles': 'reader,member',
|
||||
'X-Tenant-Id': namespace.split('_')[0]
|
||||
@@ -419,11 +410,11 @@ class TestMetadefObjects(metadef_base.MetadefFunctionalTestBase):
|
||||
headers = self._headers()
|
||||
for obj in total_objects:
|
||||
for namespace, object_name in obj.items():
|
||||
path = self._url('/v2/metadefs/namespaces/%s/objects/%s' %
|
||||
(namespace, object_name))
|
||||
response = requests.delete(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/objects/%s' % (
|
||||
namespace, object_name)
|
||||
response = self.api_delete(path, headers=headers)
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# Deleted objects should not be exist
|
||||
response = requests.get(path, headers=headers)
|
||||
# Deleted objects should not exist
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
@@ -16,7 +16,6 @@
|
||||
import http.client as http
|
||||
|
||||
from oslo_serialization import jsonutils
|
||||
import requests
|
||||
|
||||
from glance.tests.functional.v2 import metadef_base
|
||||
|
||||
@@ -25,23 +24,21 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestNamespaceProperties, self).setUp()
|
||||
self.cleanup()
|
||||
self.api_server.deployment_flavor = 'noauth'
|
||||
self.start_servers(**self.__dict__.copy())
|
||||
self.start_server(enable_cache=False)
|
||||
|
||||
def test_properties_lifecycle(self):
|
||||
# Namespace should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/MyNamespace'
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create a namespace
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
namespace_name = 'MyNamespace'
|
||||
resource_type_name = 'MyResourceType'
|
||||
resource_type_prefix = 'MyPrefix'
|
||||
data = jsonutils.dumps({
|
||||
data = {
|
||||
"namespace": namespace_name,
|
||||
"display_name": "My User Friendly Namespace",
|
||||
"description": "My description",
|
||||
@@ -54,43 +51,40 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
"prefix": resource_type_prefix
|
||||
}
|
||||
]
|
||||
})
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Property1 should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace/properties'
|
||||
'/property1')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/MyNamespace/properties/property1'
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create a property
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace/properties')
|
||||
path = '/v2/metadefs/namespaces/MyNamespace/properties'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
property_name = "property1"
|
||||
data = jsonutils.dumps(
|
||||
{
|
||||
"name": property_name,
|
||||
"type": "integer",
|
||||
"title": "property1",
|
||||
"description": "property1 description",
|
||||
"default": 100,
|
||||
"minimum": 100,
|
||||
"maximum": 30000369,
|
||||
"readonly": False,
|
||||
}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
data = {
|
||||
"name": property_name,
|
||||
"type": "integer",
|
||||
"title": "property1",
|
||||
"description": "property1 description",
|
||||
"default": 100,
|
||||
"minimum": 100,
|
||||
"maximum": 30000369,
|
||||
"readonly": False,
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Attempt to insert a duplicate
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CONFLICT, response.status_code)
|
||||
|
||||
# Get the property created above
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace_name, property_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace_name, property_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
property_object = jsonutils.loads(response.text)
|
||||
self.assertEqual("integer", property_object['type'])
|
||||
@@ -102,19 +96,19 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(30000369, property_object['maximum'])
|
||||
|
||||
# Get the property with specific resource type association
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s%s' % (
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s%s' % (
|
||||
namespace_name, property_name, '='.join(['?resource_type',
|
||||
resource_type_name])))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
resource_type_name]))
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Get the property with prefix and specific resource type association
|
||||
property_name_with_prefix = ''.join([resource_type_prefix,
|
||||
property_name])
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s%s' % (
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s%s' % (
|
||||
namespace_name, property_name_with_prefix, '='.join([
|
||||
'?resource_type', resource_type_name])))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
'?resource_type', resource_type_name]))
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
property_object = jsonutils.loads(response.text)
|
||||
self.assertEqual("integer", property_object['type'])
|
||||
@@ -153,25 +147,23 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(property_object[key], value, key)
|
||||
|
||||
# The property should be mutable
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace_name, property_name))
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace_name, property_name)
|
||||
media_type = 'application/json'
|
||||
headers = self._headers({'content-type': media_type})
|
||||
property_name = "property1-UPDATED"
|
||||
data = jsonutils.dumps(
|
||||
{
|
||||
"name": property_name,
|
||||
"type": "string",
|
||||
"title": "string property",
|
||||
"description": "desc-UPDATED",
|
||||
"operators": ["<or>"],
|
||||
"default": "value-UPDATED",
|
||||
"minLength": 5,
|
||||
"maxLength": 10,
|
||||
"readonly": True,
|
||||
}
|
||||
)
|
||||
response = requests.put(path, headers=headers, data=data)
|
||||
data = {
|
||||
"name": property_name,
|
||||
"type": "string",
|
||||
"title": "string property",
|
||||
"description": "desc-UPDATED",
|
||||
"operators": ["<or>"],
|
||||
"default": "value-UPDATED",
|
||||
"minLength": 5,
|
||||
"maxLength": 10,
|
||||
"readonly": True,
|
||||
}
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
|
||||
# Returned property should reflect the changes
|
||||
@@ -185,9 +177,10 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertTrue(property_object['readonly'])
|
||||
|
||||
# Updates should persist across requests
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace_name, property_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace_name, property_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
property_object = jsonutils.loads(response.text)
|
||||
self.assertEqual('string', property_object['type'])
|
||||
self.assertEqual('desc-UPDATED', property_object['description'])
|
||||
self.assertEqual('value-UPDATED', property_object['default'])
|
||||
@@ -196,15 +189,15 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(10, property_object['maxLength'])
|
||||
|
||||
# Deletion of property property1
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace_name, property_name))
|
||||
response = requests.delete(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace_name, property_name)
|
||||
response = self.api_delete(path, headers=self._headers())
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# property1 should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace_name, property_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace_name, property_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
def _create_properties(self, namespaces):
|
||||
@@ -217,11 +210,11 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
"title": "property",
|
||||
"description": "property description",
|
||||
}
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties' %
|
||||
namespace['namespace'])
|
||||
response = requests.post(path, headers=headers, json=data)
|
||||
path = ('/v2/metadefs/namespaces/%s/'
|
||||
'properties') % namespace['namespace']
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
prop_metadata = response.json()
|
||||
prop_metadata = jsonutils.loads(response.text)
|
||||
metadef_property = dict()
|
||||
metadef_property[namespace['namespace']] = prop_metadata['name']
|
||||
properties.append(metadef_property)
|
||||
@@ -230,22 +223,23 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def _update_property(self, path, headers, data):
|
||||
# The property should be mutable
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
|
||||
# Returned property should reflect the changes
|
||||
property_object = response.json()
|
||||
property_object = jsonutils.loads(response.text)
|
||||
self.assertEqual('string', property_object['type'])
|
||||
self.assertEqual(data['description'], property_object['description'])
|
||||
|
||||
# Updates should persist across requests
|
||||
response = requests.get(path, headers=self._headers())
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
property_object = jsonutils.loads(response.text)
|
||||
self.assertEqual('string', property_object['type'])
|
||||
self.assertEqual(data['description'], property_object['description'])
|
||||
|
||||
def test_role_base_metadata_properties_lifecycle(self):
|
||||
# Create public and private namespaces for tenant1 and tenant2
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
tenant1_namespaces = []
|
||||
tenant2_namespaces = []
|
||||
@@ -261,6 +255,7 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
}
|
||||
namespace = self.create_namespace(path, headers,
|
||||
namespace_data)
|
||||
namespace = jsonutils.loads(namespace.text)
|
||||
self.assertNamespacesEqual(namespace, namespace_data)
|
||||
if tenant == self.tenant1:
|
||||
tenant1_namespaces.append(namespace)
|
||||
@@ -277,10 +272,9 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
'X-Roles': 'reader,member'})
|
||||
for prop in properties:
|
||||
for namespace, property_name in prop.items():
|
||||
path = \
|
||||
self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace, property_name))
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace, property_name)
|
||||
response = self.api_get(path, headers=headers)
|
||||
if namespace.split('_')[1] == 'public':
|
||||
expected = http.OK
|
||||
else:
|
||||
@@ -291,12 +285,12 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(expected, response.status_code)
|
||||
|
||||
# Make sure the same holds for listing
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s/properties' % namespace)
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/properties' % namespace
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(expected, response.status_code)
|
||||
if expected == http.OK:
|
||||
resp_props = response.json()['properties'].values()
|
||||
resp_props = jsonutils.loads(
|
||||
response.text)['properties'].values()
|
||||
self.assertEqual(
|
||||
sorted(prop.values()),
|
||||
sorted([x['name']
|
||||
@@ -320,12 +314,12 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
"title": "string property",
|
||||
"description": "desc-UPDATED",
|
||||
}
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace, property_name))
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace, property_name)
|
||||
|
||||
# Update property should fail with non admin role
|
||||
headers['X-Roles'] = "reader,member"
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Should work with admin role
|
||||
@@ -336,9 +330,9 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
# Delete property should not be allowed to non admin role
|
||||
for prop in total_properties:
|
||||
for namespace, property_name in prop.items():
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace, property_name))
|
||||
response = requests.delete(
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace, property_name)
|
||||
response = self.api_delete(
|
||||
path, headers=self._headers({
|
||||
'X-Roles': 'reader,member',
|
||||
'X-Tenant-Id': namespace.split('_')[0]
|
||||
@@ -349,11 +343,11 @@ class TestNamespaceProperties(metadef_base.MetadefFunctionalTestBase):
|
||||
headers = self._headers()
|
||||
for prop in total_properties:
|
||||
for namespace, property_name in prop.items():
|
||||
path = self._url('/v2/metadefs/namespaces/%s/properties/%s' %
|
||||
(namespace, property_name))
|
||||
response = requests.delete(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
|
||||
namespace, property_name)
|
||||
response = self.api_delete(path, headers=headers)
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# Deleted property should not be exist
|
||||
response = requests.get(path, headers=headers)
|
||||
# Deleted property should not exist
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
@@ -16,7 +16,6 @@
|
||||
import http.client as http
|
||||
|
||||
from oslo_serialization import jsonutils
|
||||
import requests
|
||||
|
||||
from glance.tests.functional.v2 import metadef_base
|
||||
|
||||
@@ -25,61 +24,55 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestMetadefResourceTypes, self).setUp()
|
||||
self.cleanup()
|
||||
self.api_server.deployment_flavor = 'noauth'
|
||||
self.start_servers(**self.__dict__.copy())
|
||||
self.start_server(enable_cache=False)
|
||||
|
||||
def test_metadef_resource_types_lifecycle(self):
|
||||
# Namespace should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/MyNamespace'
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create a namespace
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
namespace_name = 'MyNamespace'
|
||||
data = jsonutils.dumps({
|
||||
data = {
|
||||
"namespace": namespace_name,
|
||||
"display_name": "My User Friendly Namespace",
|
||||
"description": "My description",
|
||||
"visibility": "public",
|
||||
"protected": False,
|
||||
"owner": "The Test Owner"
|
||||
})
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Resource type should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/%s/resource_types' %
|
||||
(namespace_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/resource_types' % (namespace_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
metadef_resource_type = jsonutils.loads(response.text)
|
||||
self.assertEqual(
|
||||
0, len(metadef_resource_type['resource_type_associations']))
|
||||
|
||||
# Create a resource type
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace/resource_types')
|
||||
path = '/v2/metadefs/namespaces/MyNamespace/resource_types'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
metadef_resource_type_name = "resource_type1"
|
||||
data = jsonutils.dumps(
|
||||
{
|
||||
"name": "resource_type1",
|
||||
"prefix": "hw_",
|
||||
"properties_target": "image",
|
||||
}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
data = {
|
||||
"name": "resource_type1",
|
||||
"prefix": "hw_",
|
||||
"properties_target": "image",
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Attempt to insert a duplicate
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CONFLICT, response.status_code)
|
||||
|
||||
# Get the metadef resource type created above
|
||||
path = self._url('/v2/metadefs/namespaces/%s/resource_types' %
|
||||
(namespace_name))
|
||||
response = requests.get(path,
|
||||
path = '/v2/metadefs/namespaces/%s/resource_types' % (namespace_name)
|
||||
response = self.api_get(path,
|
||||
headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
metadef_resource_type = jsonutils.loads(response.text)
|
||||
@@ -118,15 +111,14 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
value, key)
|
||||
|
||||
# Deassociate of metadef resource type resource_type1
|
||||
path = self._url('/v2/metadefs/namespaces/%s/resource_types/%s' %
|
||||
(namespace_name, metadef_resource_type_name))
|
||||
response = requests.delete(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/resource_types/%s' % (
|
||||
namespace_name, metadef_resource_type_name)
|
||||
response = self.api_delete(path, headers=self._headers())
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# resource_type1 should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/%s/resource_types' %
|
||||
(namespace_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/resource_types' % (namespace_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
metadef_resource_type = jsonutils.loads(response.text)
|
||||
self.assertEqual(
|
||||
0, len(metadef_resource_type['resource_type_associations']))
|
||||
@@ -140,11 +132,11 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
"prefix": "hw_",
|
||||
"properties_target": "image"
|
||||
}
|
||||
path = self._url('/v2/metadefs/namespaces/%s/resource_types' %
|
||||
(namespace['namespace']))
|
||||
response = requests.post(path, headers=headers, json=data)
|
||||
path = ('/v2/metadefs/namespaces/%s/'
|
||||
'resource_types') % (namespace['namespace'])
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
rs_type = response.json()
|
||||
rs_type = jsonutils.loads(response.text)
|
||||
resource_type = dict()
|
||||
resource_type[namespace['namespace']] = rs_type['name']
|
||||
resource_types.append(resource_type)
|
||||
@@ -153,7 +145,7 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def test_role_base_metadef_resource_types_lifecycle(self):
|
||||
# Create public and private namespaces for tenant1 and tenant2
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
tenant1_namespaces = []
|
||||
tenant2_namespaces = []
|
||||
@@ -169,6 +161,7 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
}
|
||||
namespace = self.create_namespace(path, headers,
|
||||
namespace_data)
|
||||
namespace = jsonutils.loads(namespace.text)
|
||||
self.assertNamespacesEqual(namespace, namespace_data)
|
||||
if tenant == self.tenant1:
|
||||
tenant1_namespaces.append(namespace)
|
||||
@@ -185,9 +178,9 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
headers = self._headers({'X-Tenant-Id': tenant,
|
||||
'X-Roles': 'reader,member'})
|
||||
for namespace in namespaces:
|
||||
path = self._url('/v2/metadefs/namespaces/%s/resource_types' %
|
||||
(namespace['namespace']))
|
||||
response = requests.get(path, headers=headers)
|
||||
path = ('/v2/metadefs/namespaces/%s/'
|
||||
'resource_types') % (namespace['namespace'])
|
||||
response = self.api_get(path, headers=headers)
|
||||
if namespace['visibility'] == 'public':
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
else:
|
||||
@@ -195,12 +188,12 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def _check_resource_types(tenant, total_rs_types):
|
||||
# Resource types are visible across tenants for all users
|
||||
path = self._url('/v2/metadefs/resource_types')
|
||||
path = '/v2/metadefs/resource_types'
|
||||
headers = self._headers({'X-Tenant-Id': tenant,
|
||||
'X-Roles': 'reader,member'})
|
||||
response = requests.get(path, headers=headers)
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
metadef_resource_type = response.json()
|
||||
metadef_resource_type = jsonutils.loads(response.text)
|
||||
|
||||
# The resource types list count should be same as the total
|
||||
# resource types created across the tenants.
|
||||
@@ -227,10 +220,9 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
# Disassociate resource type should not be allowed to non admin role
|
||||
for resource_type in total_resource_types:
|
||||
for namespace, rs_type in resource_type.items():
|
||||
path = \
|
||||
self._url('/v2/metadefs/namespaces/%s/resource_types/%s' %
|
||||
(namespace, rs_type))
|
||||
response = requests.delete(
|
||||
path = '/v2/metadefs/namespaces/%s/resource_types/%s' % (
|
||||
namespace, rs_type)
|
||||
response = self.api_delete(
|
||||
path, headers=self._headers({
|
||||
'X-Roles': 'reader,member',
|
||||
'X-Tenant-Id': namespace.split('_')[0]
|
||||
@@ -241,19 +233,17 @@ class TestMetadefResourceTypes(metadef_base.MetadefFunctionalTestBase):
|
||||
headers = self._headers()
|
||||
for resource_type in total_resource_types:
|
||||
for namespace, rs_type in resource_type.items():
|
||||
path = \
|
||||
self._url('/v2/metadefs/namespaces/%s/resource_types/%s' %
|
||||
(namespace, rs_type))
|
||||
response = requests.delete(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/resource_types/%s' % (
|
||||
namespace, rs_type)
|
||||
response = self.api_delete(path, headers=headers)
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# Disassociated resource type should not be exist
|
||||
# When the specified resource type is not associated with given
|
||||
# namespace then it returns empty list in response instead of
|
||||
# raising not found error
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s/resource_types' % namespace)
|
||||
response = requests.get(path, headers=headers)
|
||||
metadef_resource_type = response.json()
|
||||
path = '/v2/metadefs/namespaces/%s/resource_types' % namespace
|
||||
response = self.api_get(path, headers=headers)
|
||||
metadef_resource_type = jsonutils.loads(response.text)
|
||||
self.assertEqual(
|
||||
[], metadef_resource_type['resource_type_associations'])
|
||||
|
||||
@@ -16,7 +16,6 @@
|
||||
import http.client as http
|
||||
|
||||
from oslo_serialization import jsonutils
|
||||
import requests
|
||||
|
||||
from glance.tests.functional.v2 import metadef_base
|
||||
|
||||
@@ -25,52 +24,49 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestMetadefTags, self).setUp()
|
||||
self.cleanup()
|
||||
self.api_server.deployment_flavor = 'noauth'
|
||||
self.start_servers(**self.__dict__.copy())
|
||||
self.start_server(enable_cache=False)
|
||||
|
||||
def test_metadata_tags_lifecycle(self):
|
||||
# Namespace should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/MyNamespace')
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/MyNamespace'
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create a namespace
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
namespace_name = 'MyNamespace'
|
||||
data = jsonutils.dumps({
|
||||
data = {
|
||||
"namespace": namespace_name,
|
||||
"display_name": "My User Friendly Namespace",
|
||||
"description": "My description",
|
||||
"visibility": "public",
|
||||
"protected": False,
|
||||
"owner": "The Test Owner"}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
"owner": "The Test Owner"
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Metadata tag should not exist
|
||||
metadata_tag_name = "tag1"
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace_name, metadata_tag_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace_name, metadata_tag_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create the metadata tag
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
response = requests.post(path, headers=headers)
|
||||
response = self.api_post(path, headers=headers)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Get the metadata tag created above
|
||||
response = requests.get(path,
|
||||
response = self.api_get(path,
|
||||
headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
metadata_tag = jsonutils.loads(response.text)
|
||||
self.assertEqual(metadata_tag_name, metadata_tag['name'])
|
||||
|
||||
# Returned tag should match the created tag
|
||||
metadata_tag = jsonutils.loads(response.text)
|
||||
checked_keys = set([
|
||||
'name',
|
||||
'created_at',
|
||||
@@ -91,21 +87,19 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
# Try to create a duplicate metadata tag
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
response = requests.post(path, headers=headers)
|
||||
response = self.api_post(path, headers=headers)
|
||||
self.assertEqual(http.CONFLICT, response.status_code)
|
||||
|
||||
# The metadata_tag should be mutable
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace_name, metadata_tag_name))
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace_name, metadata_tag_name)
|
||||
media_type = 'application/json'
|
||||
headers = self._headers({'content-type': media_type})
|
||||
metadata_tag_name = "tag1-UPDATED"
|
||||
data = jsonutils.dumps(
|
||||
{
|
||||
"name": metadata_tag_name
|
||||
}
|
||||
)
|
||||
response = requests.put(path, headers=headers, data=data)
|
||||
data = {
|
||||
"name": metadata_tag_name
|
||||
}
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
|
||||
# Returned metadata_tag should reflect the changes
|
||||
@@ -113,79 +107,93 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual('tag1-UPDATED', metadata_tag['name'])
|
||||
|
||||
# Updates should persist across requests
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace_name, metadata_tag_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace_name, metadata_tag_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
self.assertEqual('tag1-UPDATED', metadata_tag['name'])
|
||||
|
||||
# Deletion of metadata_tag_name
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace_name, metadata_tag_name))
|
||||
response = requests.delete(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace_name, metadata_tag_name)
|
||||
response = self.api_delete(path, headers=self._headers())
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# metadata_tag_name should not exist
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace_name, metadata_tag_name))
|
||||
response = requests.get(path, headers=self._headers())
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace_name, metadata_tag_name)
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create multiple tags.
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags' %
|
||||
(namespace_name))
|
||||
path = '/v2/metadefs/namespaces/%s/tags' % namespace_name
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
data = jsonutils.dumps(
|
||||
{"tags": [{"name": "tag1"}, {"name": "tag2"}, {"name": "tag3"}]}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
data = {
|
||||
"tags": [
|
||||
{"name": "tag1"},
|
||||
{"name": "tag2"},
|
||||
{"name": "tag3"}
|
||||
]
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# List out the three new tags.
|
||||
response = requests.get(path, headers=self._headers())
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
tags = jsonutils.loads(response.text)['tags']
|
||||
self.assertEqual(3, len(tags))
|
||||
|
||||
# Attempt to create bogus duplicate tag4
|
||||
data = jsonutils.dumps(
|
||||
{"tags": [{"name": "tag4"}, {"name": "tag5"}, {"name": "tag4"}]}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
data = {
|
||||
"tags": [
|
||||
{"name": "tag4"},
|
||||
{"name": "tag5"},
|
||||
{"name": "tag4"}
|
||||
]
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CONFLICT, response.status_code)
|
||||
|
||||
# Verify the previous 3 still exist
|
||||
response = requests.get(path, headers=self._headers())
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
tags = jsonutils.loads(response.text)['tags']
|
||||
self.assertEqual(3, len(tags))
|
||||
|
||||
# Create new tags and append to existing tags.
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags' %
|
||||
(namespace_name))
|
||||
path = '/v2/metadefs/namespaces/%s/tags' % namespace_name
|
||||
headers = self._headers({'content-type': 'application/json',
|
||||
'X-Openstack-Append': 'True'})
|
||||
data = jsonutils.dumps(
|
||||
{"tags": [{"name": "tag4"}, {"name": "tag5"}, {"name": "tag6"}]}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
data = {
|
||||
"tags": [
|
||||
{"name": "tag4"},
|
||||
{"name": "tag5"},
|
||||
{"name": "tag6"}
|
||||
]
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# List out all six tags.
|
||||
response = requests.get(path, headers=self._headers())
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
tags = jsonutils.loads(response.text)['tags']
|
||||
self.assertEqual(6, len(tags))
|
||||
|
||||
# Attempt to create duplicate existing tag6
|
||||
data = jsonutils.dumps(
|
||||
{"tags": [{"name": "tag6"}, {"name": "tag7"}, {"name": "tag8"}]}
|
||||
)
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
data = {
|
||||
"tags": [
|
||||
{"name": "tag6"},
|
||||
{"name": "tag7"},
|
||||
{"name": "tag8"}
|
||||
]
|
||||
}
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CONFLICT, response.status_code)
|
||||
|
||||
# Verify the previous 6 still exist
|
||||
response = requests.get(path, headers=self._headers())
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
tags = jsonutils.loads(response.text)['tags']
|
||||
self.assertEqual(6, len(tags))
|
||||
@@ -195,11 +203,11 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
for namespace in namespaces:
|
||||
headers = self._headers({'X-Tenant-Id': namespace['owner']})
|
||||
tag_name = "tag_of_%s" % (namespace['namespace'])
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace['namespace'], tag_name))
|
||||
response = requests.post(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace['namespace'], tag_name)
|
||||
response = self.api_post(path, headers=headers)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
tag_metadata = response.json()
|
||||
tag_metadata = jsonutils.loads(response.text)
|
||||
metadef_tags = dict()
|
||||
metadef_tags[namespace['namespace']] = tag_metadata['name']
|
||||
tags.append(metadef_tags)
|
||||
@@ -208,20 +216,20 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
|
||||
def _update_tags(self, path, headers, data):
|
||||
# The tag should be mutable
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.OK, response.status_code, response.text)
|
||||
# Returned metadata_tag should reflect the changes
|
||||
metadata_tag = response.json()
|
||||
metadata_tag = jsonutils.loads(response.text)
|
||||
self.assertEqual(data['name'], metadata_tag['name'])
|
||||
|
||||
# Updates should persist across requests
|
||||
response = requests.get(path, headers=self._headers())
|
||||
response = self.api_get(path, headers=self._headers())
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
self.assertEqual(data['name'], metadata_tag['name'])
|
||||
|
||||
def test_role_base_metadata_tags_lifecycle(self):
|
||||
# Create public and private namespaces for tenant1 and tenant2
|
||||
path = self._url('/v2/metadefs/namespaces')
|
||||
path = '/v2/metadefs/namespaces'
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
tenant1_namespaces = []
|
||||
tenant2_namespaces = []
|
||||
@@ -237,6 +245,7 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
}
|
||||
namespace = self.create_namespace(path, headers,
|
||||
namespace_data)
|
||||
namespace = jsonutils.loads(namespace.text)
|
||||
self.assertNamespacesEqual(namespace, namespace_data)
|
||||
if tenant == self.tenant1:
|
||||
tenant1_namespaces.append(namespace)
|
||||
@@ -253,9 +262,9 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
'X-Roles': 'reader,member'})
|
||||
for tag in tags:
|
||||
for namespace, tag_name in tag.items():
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace, tag_name))
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace, tag_name)
|
||||
response = self.api_get(path, headers=headers)
|
||||
if namespace.split('_')[1] == 'public':
|
||||
expected = http.OK
|
||||
else:
|
||||
@@ -266,12 +275,11 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
self.assertEqual(expected, response.status_code)
|
||||
|
||||
# Make sure the same holds for listing
|
||||
path = self._url(
|
||||
'/v2/metadefs/namespaces/%s/tags' % namespace)
|
||||
response = requests.get(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/tags' % namespace
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(expected, response.status_code)
|
||||
if expected == http.OK:
|
||||
resp_props = response.json()['tags']
|
||||
resp_props = jsonutils.loads(response.text)['tags']
|
||||
self.assertEqual(
|
||||
sorted(tag.values()),
|
||||
sorted([x['name']
|
||||
@@ -291,12 +299,12 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
for namespace, tag_name in tag.items():
|
||||
data = {
|
||||
"name": tag_name}
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace, tag_name))
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace, tag_name)
|
||||
|
||||
# Update tag should fail with non admin role
|
||||
headers['X-Roles'] = "reader,member"
|
||||
response = requests.put(path, headers=headers, json=data)
|
||||
response = self.api_put(path, headers=headers, json=data)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Should work with admin role
|
||||
@@ -307,9 +315,9 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
# Delete tags should not be allowed to non admin role
|
||||
for tag in total_tags:
|
||||
for namespace, tag_name in tag.items():
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace, tag_name))
|
||||
response = requests.delete(
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace, tag_name)
|
||||
response = self.api_delete(
|
||||
path, headers=self._headers({
|
||||
'X-Roles': 'reader,member',
|
||||
'X-Tenant-Id': namespace.split('_')[0]
|
||||
@@ -320,13 +328,13 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
headers = self._headers()
|
||||
for tag in total_tags:
|
||||
for namespace, tag_name in tag.items():
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags/%s' %
|
||||
(namespace, tag_name))
|
||||
response = requests.delete(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/tags/%s' % (
|
||||
namespace, tag_name)
|
||||
response = self.api_delete(path, headers=headers)
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# Deleted tags should not be exist
|
||||
response = requests.get(path, headers=headers)
|
||||
# Deleted tags should not exist
|
||||
response = self.api_get(path, headers=headers)
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Create multiple tags should not be allowed to non admin role
|
||||
@@ -336,32 +344,29 @@ class TestMetadefTags(metadef_base.MetadefFunctionalTestBase):
|
||||
"tags": [{"name": "tag1"}, {"name": "tag2"}, {"name": "tag3"}]
|
||||
}
|
||||
for namespace in tenant1_namespaces:
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags' %
|
||||
(namespace['namespace']))
|
||||
response = requests.post(path, headers=headers, json=data)
|
||||
path = '/v2/metadefs/namespaces/%s/tags' % (
|
||||
namespace['namespace'])
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Create multiple tags.
|
||||
headers = self._headers({'content-type': 'application/json'})
|
||||
for namespace in tenant1_namespaces:
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags' %
|
||||
(namespace['namespace']))
|
||||
response = requests.post(path, headers=headers, json=data)
|
||||
path = '/v2/metadefs/namespaces/%s/tags' % (namespace['namespace'])
|
||||
response = self.api_post(path, headers=headers, json=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
|
||||
# Delete multiple tags should not be allowed with non admin role
|
||||
headers = self._headers({'content-type': 'application/json',
|
||||
'X-Roles': 'reader,member'})
|
||||
for namespace in tenant1_namespaces:
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags' %
|
||||
(namespace['namespace']))
|
||||
response = requests.delete(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/tags' % (namespace['namespace'])
|
||||
response = self.api_delete(path, headers=headers)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Delete multiple tags created above created tags
|
||||
headers = self._headers()
|
||||
for namespace in tenant1_namespaces:
|
||||
path = self._url('/v2/metadefs/namespaces/%s/tags' %
|
||||
(namespace['namespace']))
|
||||
response = requests.delete(path, headers=headers)
|
||||
path = '/v2/metadefs/namespaces/%s/tags' % (namespace['namespace'])
|
||||
response = self.api_delete(path, headers=headers)
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
Reference in New Issue
Block a user