[service credential bindings] Implement create operation

antechrestos · antechrestos · commit 11a8bebc6ec9 · 2022-02-19T09:58:52.000+01:00
Close #164
diff --git a/main/cloudfoundry_client/v3/entities.py b/main/cloudfoundry_client/v3/entities.py
@@ -143,6 +143,10 @@ def _patch(self, url: str, data: dict, entity_type: Optional[ENTITY_TYPE] = None
 
     def _delete(self, url: str) -> Optional[str]:
         response = self.client.delete(url)
+        return self._location(response)
+
+    @staticmethod
+    def _location(response):
         try:
             return response.headers["Location"]
         except (AttributeError, KeyError):
@@ -152,14 +156,19 @@ def _remove(self, resource_id: str, asynchronous: bool = True) -> Optional[str]:
         url = "%s%s/%s" % (self.target_endpoint, self.entity_uri, resource_id)
         job_location = self._delete(url)
         if job_location is not None:
-            job_url = urlparse(job_location)
-            job_guid = job_url.path.rsplit("/", 1)[-1]
+            job_guid = self._extract_job_guid(job_location)
             if not asynchronous:
                 self.client.v3.jobs.wait_for_job_completion(job_guid)
             else:
                 return job_guid
         return None
 
+    @staticmethod
+    def _extract_job_guid(job_location):
+        job_url = urlparse(job_location)
+        job_guid = job_url.path.rsplit("/", 1)[-1]
+        return job_guid
+
     def _list(self, requested_path: str, entity_type: Optional[ENTITY_TYPE] = None, **kwargs) -> PaginateEntities:
         url_requested = EntityManager._get_url_with_encoded_params("%s%s" % (self.target_endpoint, requested_path), **kwargs)
         for element in self._paginate(url_requested, entity_type):
diff --git a/main/cloudfoundry_client/v3/service_credential_bindings.py b/main/cloudfoundry_client/v3/service_credential_bindings.py
@@ -1,6 +1,6 @@
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional, Union
 
-from cloudfoundry_client.v3.entities import EntityManager
+from cloudfoundry_client.v3.entities import EntityManager, Entity, ToOneRelationship
 
 if TYPE_CHECKING:
     from cloudfoundry_client.client import CloudFoundryClient
@@ -10,3 +10,42 @@ class ServiceCredentialBindingManager(EntityManager):
     def __init__(self, target_endpoint: str, client: "CloudFoundryClient"):
         super(ServiceCredentialBindingManager, self).__init__(target_endpoint, client,
                                                               "/v3/service_credential_bindings")
+
+    def create(
+            self,
+            name: str,
+            service_credential_binding_type: str,
+            service_instance_guid: str,
+            application_guid: Optional[str],
+            parameters: Optional[dict],
+            meta_labels: Optional[dict],
+            meta_annotations: Optional[dict],
+            asynchronous: bool = True,
+    ) -> Union[str, Entity, None]:
+        data = {
+            "name": name,
+            "type": service_credential_binding_type,
+            "relationships": {"service_instance": ToOneRelationship(service_instance_guid)},
+        }
+        if application_guid:
+            data["relationships"]["app"] = ToOneRelationship(application_guid)
+        if parameters:
+            data["parameters"] = parameters
+        if meta_labels or meta_annotations:
+            metadata = dict()
+            if meta_labels:
+                metadata["labels"] = meta_labels
+            if meta_annotations:
+                metadata["annotations"] = meta_annotations
+            data["metadata"] = metadata
+        url = "%s%s" % (self.target_endpoint, self.entity_uri)
+        response = self.client.post(url, json=data)
+        location = super(ServiceCredentialBindingManager, self)._location(response)
+        if location:
+            job_guid = super(ServiceCredentialBindingManager, self)._extract_job_guid(location)
+            if asynchronous:
+                return job_guid
+            else:
+                self.client.v3.jobs.wait_for_job_completion(job_guid)
+                return None
+        return super(ServiceCredentialBindingManager, self)._read_response(response, None)
diff --git a/test/fixtures/v3/service_credential_bindings/POST_response.json b/test/fixtures/v3/service_credential_bindings/POST_response.json
@@ -0,0 +1,50 @@
+{
+  "guid": "dde5ad2a-d8f4-44dc-a56f-0452d744f1c3",
+  "created_at": "2015-11-13T17:02:56Z",
+  "updated_at": "2016-06-08T16:41:26Z",
+  "name": "some-name",
+  "type": "app",
+  "last_operation": {
+    "type": "create",
+    "state": "succeeded",
+    "created_at": "2015-11-13T17:02:56Z",
+    "updated_at": "2016-06-08T16:41:26Z"
+  },
+  "metadata": {
+    "annotations": {
+      "foo": "bar"
+    },
+    "labels": {
+      "baz": "qux"
+    }
+  },
+  "relationships": {
+    "app": {
+      "data": {
+        "guid": "74f7c078-0934-470f-9883-4fddss5b8f13"
+      }
+    },
+    "service_instance": {
+      "data": {
+        "guid": "8bfe4c1b-9e18-45b1-83be-124163f31f9e"
+      }
+    }
+  },
+  "links": {
+    "self": {
+      "href": "https://api.example.org/v3/service_credential_bindings/dde5ad2a-d8f4-44dc-a56f-0452d744f1c3"
+    },
+    "details": {
+      "href": "https://api.example.org/v3/service_credential_bindings/dde5ad2a-d8f4-44dc-a56f-0452d744f1c3/details"
+    },
+    "parameters": {
+      "href": "https://api.example.org/v3/service_credential_bindings/dde5ad2a-d8f4-44dc-a56f-0452d744f1c3/parameters"
+    },
+    "service_instance": {
+      "href": "https://api.example.org/v3/service_instances/8bfe4c1b-9e18-45b1-83be-124163f31f9e"
+    },
+    "app": {
+      "href": "https://api.example.org/v3/apps/74f7c078-0934-470f-9883-4fddss5b8f13"
+    }
+  }
+}
diff --git a/test/v3/test_service_credential_bindings.py b/test/v3/test_service_credential_bindings.py
@@ -13,6 +13,67 @@ def setUpClass(cls):
     def setUp(self):
         self.build_client()
 
+    def test_create_managed_service_instance(self):
+        self.client.post.return_value = self.mock_response(
+            "/v3/service_credential_bindings", HTTPStatus.ACCEPTED,
+            dict(Location="https://api.example.org/v3/jobs/af5c57f6-8769-41fa-a499-2c84ed896788")
+        )
+        location = self.client.v3.service_credential_bindings.create("some-binding-name", "app",
+                                                                     "7304bc3c-7010-11ea-8840-48bf6bec2d78",
+                                                                     "e0e4417c-74ee-11ea-a604-48bf6bec2d78",
+                                                                     parameters=dict(key1="value1", key2="value2"),
+                                                                     meta_labels=dict(foo="bar"),
+                                                                     meta_annotations=dict(baz="qux"))
+
+        self.assertEqual("af5c57f6-8769-41fa-a499-2c84ed896788", location)
+        self.client.post.assert_called_with(
+            self.client.post.return_value.url,
+            json={
+                "type": "app",
+                "name": "some-binding-name",
+                "relationships": {
+                    "service_instance": {
+                        "data": {"guid": "7304bc3c-7010-11ea-8840-48bf6bec2d78"}
+                    },
+                    "app": {
+                        "data": {"guid": "e0e4417c-74ee-11ea-a604-48bf6bec2d78"}
+                    }
+                },
+                "parameters": {"key1": "value1", "key2": "value2"},
+                "metadata": {"labels": {"foo": "bar"}, "annotations": {"baz": "qux"}},
+            },
+        )
+
+    def test_create_user_provided_service_instance(self):
+        self.client.post.return_value = self.mock_response(
+            "/v3/service_credential_bindings", HTTPStatus.ACCEPTED,
+            None,
+            "v3", "service_credential_bindings", "POST_response.json"
+        )
+        result = self.client.v3.service_credential_bindings.create("some-binding-name", "key",
+                                                                   "7304bc3c-7010-11ea-8840-48bf6bec2d78",
+                                                                   None,
+                                                                   parameters=dict(key1="value1", key2="value2"),
+                                                                   meta_labels=dict(foo="bar"),
+                                                                   meta_annotations=dict(baz="qux"))
+        self.assertIsNotNone(result)
+        self.assertEqual("some-name", result["name"])
+        self.assertIsInstance(result, Entity)
+        self.client.post.assert_called_with(
+            self.client.post.return_value.url,
+            json={
+                "type": "key",
+                "name": "some-binding-name",
+                "relationships": {
+                    "service_instance": {
+                        "data": {"guid": "7304bc3c-7010-11ea-8840-48bf6bec2d78"}
+                    }
+                },
+                "parameters": {"key1": "value1", "key2": "value2"},
+                "metadata": {"labels": {"foo": "bar"}, "annotations": {"baz": "qux"}},
+            },
+        )
+
     def test_list(self):
         self.client.get.return_value = self.mock_response(
             "/v3/service_credential_bindings", HTTPStatus.OK, None, "v3", "service_credential_bindings",
