Validated URLs in v2 endpoint creation API

Fixed the 500 response when endpoint is created without specifying
either adminurl, publicurl or internalurl empty.
Added unit test coverage for v2 endpoint create API.

Fixes LP: #1097747

Change-Id: I1875859bd7e197e985a5b5395c8caefff3c83b11

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)