feat(toolchain): add function to create toolchain event (#39)

* feat(toolchain): add eu-es region

Signed-off-by: Omar Al Bastami <omar.albastami@ibm.com>

* fix: lint

Signed-off-by: Omar Al Bastami <omar.albastami@ibm.com>

* fix: replace double quotes with single quotes in version

Signed-off-by: Omar Al Bastami <omar.albastami@ibm.com>

* fix: exclude version.py from linter to avoid deployment errors

Signed-off-by: Omar Al Bastami <omar.albastami@ibm.com>

* build: bump python core

Signed-off-by: Omar Al Bastami <omar.albastami@ibm.com>

* fix: resolve conflict

Signed-off-by: Omar Al Bastami <omar.albastami@ibm.com>

* fix: lint fixes

Signed-off-by: Omar Al Bastami <omar.albastami@ibm.com>

---------

Signed-off-by: Omar Al Bastami <omar.albastami@ibm.com>
Co-authored-by: Omar Al Bastami <omar.albastami@ibm.com>

Makefile

@@ -35,7 +35,7 @@ test-examples:
 	${PYTHON} -m pytest examples
 
 lint:
-	# ${PYTHON} -m pylint ${LINT_DIRS} --exit-zero
+	${PYTHON} -m pylint ${LINT_DIRS} --exit-zero
 	black --check ${LINT_DIRS} --exclude="ibm_continuous_delivery/version.py"
 
 lint-fix:

examples/test_cd_toolchain_v2_examples.py

@@ -198,6 +198,30 @@ def test_update_toolchain_example(self):
         except ApiException as e:
             pytest.fail(str(e))
 
+    @needscredentials
+    def test_create_toolchain_event_example(self):
+        """
+        create_toolchain_event request example
+        """
+        try:
+            print("\ncreate_toolchain_event() result:")
+            # begin-create_toolchain_event
+
+            response = cd_toolchain_service.create_toolchain_event(
+                toolchain_id=toolchain_id_link,
+                title="My-custom-event",
+                description="This is my custom event",
+                content_type="application/json",
+            )
+            toolchain_event_post = response.get_result()
+
+            print(json.dumps(toolchain_event_post, indent=2))
+
+            # end-create_toolchain_event
+
+        except ApiException as e:
+            pytest.fail(str(e))
+
     @needscredentials
     def test_list_tools_example(self):
         """

ibm_continuous_delivery/cd_toolchain_v2.py

@@ -374,6 +374,82 @@ def update_toolchain(
         response = self.send(request, **kwargs)
         return response
 
+    def create_toolchain_event(
+        self,
+        toolchain_id: str,
+        title: str,
+        description: str,
+        content_type: str,
+        *,
+        data: "ToolchainEventPrototypeData" = None,
+        **kwargs,
+    ) -> DetailedResponse:
+        """
+        Create a toolchain event.
+
+        Creates and sends a custom event to Event Notifications. This requires an Event
+        Notification tool. This method is BETA and subject to change.
+
+        :param str toolchain_id: ID of the toolchain to send events from.
+        :param str title: Event title.
+        :param str description: Describes the event.
+        :param str content_type: The content type of the attached data. Supported
+               values are `text/plain`, `application/json`, and `none`.
+        :param ToolchainEventPrototypeData data: (optional) Additional data to be
+               added with the event. The format must correspond to the value of
+               `content_type`.
+        :param dict headers: A `dict` containing the request headers
+        :return: A `DetailedResponse` containing the result, headers and HTTP status code.
+        :rtype: DetailedResponse with `dict` result representing a `ToolchainEventPost` object
+        """
+
+        if not toolchain_id:
+            raise ValueError("toolchain_id must be provided")
+        if title is None:
+            raise ValueError("title must be provided")
+        if description is None:
+            raise ValueError("description must be provided")
+        if content_type is None:
+            raise ValueError("content_type must be provided")
+        if data is not None:
+            data = convert_model(data)
+        headers = {}
+        sdk_headers = get_sdk_headers(
+            service_name=self.DEFAULT_SERVICE_NAME,
+            service_version="V2",
+            operation_id="create_toolchain_event",
+        )
+        headers.update(sdk_headers)
+
+        data = {
+            "title": title,
+            "description": description,
+            "content_type": content_type,
+            "data": data,
+        }
+        data = {k: v for (k, v) in data.items() if v is not None}
+        data = json.dumps(data)
+        headers["content-type"] = "application/json"
+
+        if "headers" in kwargs:
+            headers.update(kwargs.get("headers"))
+            del kwargs["headers"]
+        headers["Accept"] = "application/json"
+
+        path_param_keys = ["toolchain_id"]
+        path_param_values = self.encode_path_vars(toolchain_id)
+        path_param_dict = dict(zip(path_param_keys, path_param_values))
+        url = "/toolchains/{toolchain_id}/events".format(**path_param_dict)
+        request = self.prepare_request(
+            method="POST",
+            url=url,
+            headers=headers,
+            data=data,
+        )
+
+        response = self.send(request, **kwargs)
+        return response
+
     #########################
     # Tools
     #########################
@@ -1570,6 +1646,211 @@ def __ne__(self, other: "ToolchainCollectionPrevious") -> bool:
         return not self == other
 
 
+class ToolchainEventPost:
+    """
+    Response structure for POST toolchain event.
+
+    :attr str id: Event ID.
+    """
+
+    def __init__(
+        self,
+        id: str,
+    ) -> None:
+        """
+        Initialize a ToolchainEventPost object.
+
+        :param str id: Event ID.
+        """
+        self.id = id
+
+    @classmethod
+    def from_dict(cls, _dict: Dict) -> "ToolchainEventPost":
+        """Initialize a ToolchainEventPost object from a json dictionary."""
+        args = {}
+        if "id" in _dict:
+            args["id"] = _dict.get("id")
+        else:
+            raise ValueError(
+                "Required property 'id' not present in ToolchainEventPost JSON"
+            )
+        return cls(**args)
+
+    @classmethod
+    def _from_dict(cls, _dict):
+        """Initialize a ToolchainEventPost object from a json dictionary."""
+        return cls.from_dict(_dict)
+
+    def to_dict(self) -> Dict:
+        """Return a json dictionary representing this model."""
+        _dict = {}
+        if hasattr(self, "id") and self.id is not None:
+            _dict["id"] = self.id
+        return _dict
+
+    def _to_dict(self):
+        """Return a json dictionary representing this model."""
+        return self.to_dict()
+
+    def __str__(self) -> str:
+        """Return a `str` version of this ToolchainEventPost object."""
+        return json.dumps(self.to_dict(), indent=2)
+
+    def __eq__(self, other: "ToolchainEventPost") -> bool:
+        """Return `true` when self and other are equal, false otherwise."""
+        if not isinstance(other, self.__class__):
+            return False
+        return self.__dict__ == other.__dict__
+
+    def __ne__(self, other: "ToolchainEventPost") -> bool:
+        """Return `true` when self and other are not equal, false otherwise."""
+        return not self == other
+
+
+class ToolchainEventPrototypeData:
+    """
+    Additional data to be added with the event. The format must correspond to the value of
+    `content_type`.
+
+    :attr ToolchainEventPrototypeDataApplicationJson application_json: (optional)
+          Contains JSON data to be added with the event. `content_type` must be set to
+          `application/json`.
+    :attr str text_plain: (optional) Contains text data to be added with the event.
+          `content_type` must be set to `text/plain`.
+    """
+
+    def __init__(
+        self,
+        *,
+        application_json: "ToolchainEventPrototypeDataApplicationJson" = None,
+        text_plain: str = None,
+    ) -> None:
+        """
+        Initialize a ToolchainEventPrototypeData object.
+
+        :param ToolchainEventPrototypeDataApplicationJson application_json:
+               (optional) Contains JSON data to be added with the event. `content_type`
+               must be set to `application/json`.
+        :param str text_plain: (optional) Contains text data to be added with the
+               event. `content_type` must be set to `text/plain`.
+        """
+        self.application_json = application_json
+        self.text_plain = text_plain
+
+    @classmethod
+    def from_dict(cls, _dict: Dict) -> "ToolchainEventPrototypeData":
+        """Initialize a ToolchainEventPrototypeData object from a json dictionary."""
+        args = {}
+        if "application_json" in _dict:
+            args[
+                "application_json"
+            ] = ToolchainEventPrototypeDataApplicationJson.from_dict(
+                _dict.get("application_json")
+            )
+        if "text_plain" in _dict:
+            args["text_plain"] = _dict.get("text_plain")
+        return cls(**args)
+
+    @classmethod
+    def _from_dict(cls, _dict):
+        """Initialize a ToolchainEventPrototypeData object from a json dictionary."""
+        return cls.from_dict(_dict)
+
+    def to_dict(self) -> Dict:
+        """Return a json dictionary representing this model."""
+        _dict = {}
+        if hasattr(self, "application_json") and self.application_json is not None:
+            if isinstance(self.application_json, dict):
+                _dict["application_json"] = self.application_json
+            else:
+                _dict["application_json"] = self.application_json.to_dict()
+        if hasattr(self, "text_plain") and self.text_plain is not None:
+            _dict["text_plain"] = self.text_plain
+        return _dict
+
+    def _to_dict(self):
+        """Return a json dictionary representing this model."""
+        return self.to_dict()
+
+    def __str__(self) -> str:
+        """Return a `str` version of this ToolchainEventPrototypeData object."""
+        return json.dumps(self.to_dict(), indent=2)
+
+    def __eq__(self, other: "ToolchainEventPrototypeData") -> bool:
+        """Return `true` when self and other are equal, false otherwise."""
+        if not isinstance(other, self.__class__):
+            return False
+        return self.__dict__ == other.__dict__
+
+    def __ne__(self, other: "ToolchainEventPrototypeData") -> bool:
+        """Return `true` when self and other are not equal, false otherwise."""
+        return not self == other
+
+
+class ToolchainEventPrototypeDataApplicationJson:
+    """
+    Contains JSON data to be added with the event. `content_type` must be set to
+    `application/json`.
+
+    :attr dict content: JSON-formatted key-value pairs representing any additional
+          information to be included with the event.
+    """
+
+    def __init__(
+        self,
+        content: dict,
+    ) -> None:
+        """
+        Initialize a ToolchainEventPrototypeDataApplicationJson object.
+
+        :param dict content: JSON-formatted key-value pairs representing any
+               additional information to be included with the event.
+        """
+        self.content = content
+
+    @classmethod
+    def from_dict(cls, _dict: Dict) -> "ToolchainEventPrototypeDataApplicationJson":
+        """Initialize a ToolchainEventPrototypeDataApplicationJson object from a json dictionary."""
+        args = {}
+        if "content" in _dict:
+            args["content"] = _dict.get("content")
+        else:
+            raise ValueError(
+                "Required property 'content' not present in ToolchainEventPrototypeDataApplicationJson JSON"
+            )
+        return cls(**args)
+
+    @classmethod
+    def _from_dict(cls, _dict):
+        """Initialize a ToolchainEventPrototypeDataApplicationJson object from a json dictionary."""
+        return cls.from_dict(_dict)
+
+    def to_dict(self) -> Dict:
+        """Return a json dictionary representing this model."""
+        _dict = {}
+        if hasattr(self, "content") and self.content is not None:
+            _dict["content"] = self.content
+        return _dict
+
+    def _to_dict(self):
+        """Return a json dictionary representing this model."""
+        return self.to_dict()
+
+    def __str__(self) -> str:
+        """Return a `str` version of this ToolchainEventPrototypeDataApplicationJson object."""
+        return json.dumps(self.to_dict(), indent=2)
+
+    def __eq__(self, other: "ToolchainEventPrototypeDataApplicationJson") -> bool:
+        """Return `true` when self and other are equal, false otherwise."""
+        if not isinstance(other, self.__class__):
+            return False
+        return self.__dict__ == other.__dict__
+
+    def __ne__(self, other: "ToolchainEventPrototypeDataApplicationJson") -> bool:
+        """Return `true` when self and other are not equal, false otherwise."""
+        return not self == other
+
+
 class ToolchainModel:
     """
     Model describing toolchain resource.

ibm_continuous_delivery/version.py

@@ -17,4 +17,4 @@
 """
 Version of ibm_continuous_delivery
 """
-__version__ = '1.4.0'
+__version__ = '1.4.0'

requirements.txt

@@ -1 +1 @@
-ibm_cloud_sdk_core>=3.17.0,<4.0.0
+ibm_cloud_sdk_core>=3.18.1,<4.0.0

test/integration/test_cd_toolchain_v2.py

@@ -159,6 +159,34 @@ def test_update_toolchain(self):
         toolchain_patch = response.get_result()
         assert toolchain_patch is not None
 
+    @needscredentials
+    def test_create_toolchain_event(self):
+        # Construct a dict representation of a ToolchainEventPrototypeDataApplicationJson model
+        toolchain_event_prototype_data_application_json_model = {
+            "content": {
+                "customKey1": "myCustomData",
+                "customKey2": 123,
+                "customKey3": {"nestedKey": "moreData"},
+            },
+        }
+        # Construct a dict representation of a ToolchainEventPrototypeData model
+        toolchain_event_prototype_data_model = {
+            "application_json": toolchain_event_prototype_data_application_json_model,
+            "text_plain": "This event is dispatched because the pipeline failed",
+        }
+
+        response = self.cd_toolchain_service.create_toolchain_event(
+            toolchain_id=toolchain_id_link,
+            title="My-custom-event",
+            description="This is my custom event",
+            content_type="application/json",
+            data=toolchain_event_prototype_data_model,
+        )
+
+        assert response.get_status_code() == 200
+        toolchain_event_post = response.get_result()
+        assert toolchain_event_post is not None
+
     @needscredentials
     def test_list_tools(self):
         response = self.cd_toolchain_service.list_tools(