Merge "Validated URLs in v2 endpoint creation API"

1 files changed, 80 insertions, 0 deletions

diff --git a/tests/test_catalog.py b/tests/test_catalog.py
new file mode 100644
index 00000000..690dde35
--- /dev/null
+++ b/tests/test_catalog.py
@@ -0,0 +1,80 @@
+import uuid
+
+from keystone.common.sql import util as sql_util
+from keystone import test
+
+import test_content_types
+
+
+BASE_URL = 'http://127.0.0.1:35357/v2'
+
+
+class V2CatalogTestCase(test_content_types.RestfulTestCase):
+    def setUp(self):
+        super(V2CatalogTestCase, self).setUp()
+        self.service_id = uuid.uuid4().hex
+        self.service = self.new_service_ref()
+        self.service['id'] = self.service_id
+        self.catalog_api.create_service(
+            self.service_id,
+            self.service.copy())
+
+    def new_ref(self):
+        """Populates a ref with attributes common to all API entities."""
+        return {
+            'id': uuid.uuid4().hex,
+            'name': uuid.uuid4().hex,
+            'description': uuid.uuid4().hex,
+            'enabled': True}
+
+    def new_service_ref(self):
+        ref = self.new_ref()
+        ref['type'] = uuid.uuid4().hex
+        return ref
+
+    def _get_token_id(self, r):
+        """Applicable only to JSON."""
+        return r.body['access']['token']['id']
+
+    def assertValidErrorResponse(self, response):
+        self.assertEqual(response.status, 400)
+
+    def _endpoint_create(self, expected_status=200, missing_param=None):
+        path = '/v2.0/endpoints'
+        body = {
+            "endpoint": {
+                "adminurl": "http://localhost:8080",
+                "service_id": self.service_id,
+                "region": "regionOne",
+                "internalurl": "http://localhost:8080",
+                "publicurl": "http://localhost:8080"
+            }
+        }
+        if missing_param:
+            body['endpoint'][missing_param] = None
+        r = self.admin_request(method='POST', token=self.get_scoped_token(),
+                               path=path, expected_status=expected_status,
+                               body=body)
+        return body, r
+
+    def test_endpoint_create(self):
+        req_body, response = self._endpoint_create(expected_status=200)
+        self.assertTrue('endpoint' in response.body)
+        self.assertTrue('id' in response.body['endpoint'])
+        for field, value in req_body['endpoint'].iteritems():
+            self.assertEqual(response.body['endpoint'][field], value)
+
+    def test_endpoint_create_with_missing_adminurl(self):
+        req_body, response = self._endpoint_create(expected_status=200,
+                                                   missing_param='adminurl')
+        self.assertEqual(response.status, 200)
+
+    def test_endpoint_create_with_missing_internalurl(self):
+        req_body, response = self._endpoint_create(expected_status=200,
+                                                   missing_param='internalurl')
+        self.assertEqual(response.status, 200)
+
+    def test_endpoint_create_with_missing_publicurl(self):
+        req_body, response = self._endpoint_create(expected_status=400,
+                                                   missing_param='publicurl')
+        self.assertValidErrorResponse(response)