From 4ff7032c71aa8235f12daa856b748b0dfcf6d5da Mon Sep 17 00:00:00 2001 From: omar-albastami <100662897+omar-albastami@users.noreply.github.com> Date: Fri, 15 Dec 2023 09:15:51 -0700 Subject: [PATCH] feat(toolchain): add function to create toolchain event (#39) * feat(toolchain): add eu-es region Signed-off-by: Omar Al Bastami * fix: lint Signed-off-by: Omar Al Bastami * fix: replace double quotes with single quotes in version Signed-off-by: Omar Al Bastami * fix: exclude version.py from linter to avoid deployment errors Signed-off-by: Omar Al Bastami * build: bump python core Signed-off-by: Omar Al Bastami * fix: resolve conflict Signed-off-by: Omar Al Bastami * fix: lint fixes Signed-off-by: Omar Al Bastami --------- Signed-off-by: Omar Al Bastami Co-authored-by: Omar Al Bastami --- Makefile | 2 +- examples/test_cd_toolchain_v2_examples.py | 24 ++ ibm_continuous_delivery/cd_toolchain_v2.py | 281 ++++++++++++++++++++ ibm_continuous_delivery/version.py | 2 +- requirements.txt | 2 +- test/integration/test_cd_toolchain_v2.py | 28 ++ test/unit/test_cd_toolchain_v2.py | 283 +++++++++++++++++++++ 7 files changed, 619 insertions(+), 3 deletions(-) diff --git a/Makefile b/Makefile index cf95e32..bb10b92 100644 --- a/Makefile +++ b/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: diff --git a/examples/test_cd_toolchain_v2_examples.py b/examples/test_cd_toolchain_v2_examples.py index a037ed8..f352704 100644 --- a/examples/test_cd_toolchain_v2_examples.py +++ b/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): """ diff --git a/ibm_continuous_delivery/cd_toolchain_v2.py b/ibm_continuous_delivery/cd_toolchain_v2.py index 33249c8..5c5b5ed 100644 --- a/ibm_continuous_delivery/cd_toolchain_v2.py +++ b/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. diff --git a/ibm_continuous_delivery/version.py b/ibm_continuous_delivery/version.py index 41ddf09..6ff9b84 100644 --- a/ibm_continuous_delivery/version.py +++ b/ibm_continuous_delivery/version.py @@ -17,4 +17,4 @@ """ Version of ibm_continuous_delivery """ -__version__ = '1.4.0' +__version__ = '1.4.0' \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 05b421f..9725512 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1 @@ -ibm_cloud_sdk_core>=3.17.0,<4.0.0 \ No newline at end of file +ibm_cloud_sdk_core>=3.18.1,<4.0.0 \ No newline at end of file diff --git a/test/integration/test_cd_toolchain_v2.py b/test/integration/test_cd_toolchain_v2.py index 6b3c3e8..d2074f8 100644 --- a/test/integration/test_cd_toolchain_v2.py +++ b/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( diff --git a/test/unit/test_cd_toolchain_v2.py b/test/unit/test_cd_toolchain_v2.py index d70a772..fa0eedf 100644 --- a/test/unit/test_cd_toolchain_v2.py +++ b/test/unit/test_cd_toolchain_v2.py @@ -715,6 +715,141 @@ def test_update_toolchain_value_error_with_retries(self): self.test_update_toolchain_value_error() +class TestCreateToolchainEvent: + """ + Test Class for create_toolchain_event + """ + + @responses.activate + def test_create_toolchain_event_all_params(self): + """ + create_toolchain_event() + """ + # Set up mock + url = preprocess_url("/toolchains/testString/events") + mock_response = '{"id": "id"}' + responses.add( + responses.POST, + url, + body=mock_response, + content_type="application/json", + status=200, + ) + + # Construct a dict representation of a ToolchainEventPrototypeDataApplicationJson model + toolchain_event_prototype_data_application_json_model = {} + toolchain_event_prototype_data_application_json_model["content"] = { + "anyKey": "anyValue" + } + + # Construct a dict representation of a ToolchainEventPrototypeData model + toolchain_event_prototype_data_model = {} + toolchain_event_prototype_data_model[ + "application_json" + ] = toolchain_event_prototype_data_application_json_model + toolchain_event_prototype_data_model[ + "text_plain" + ] = "This event is dispatched because the pipeline failed" + + # Set up parameter values + toolchain_id = "testString" + title = "My-custom-event" + description = "This is my custom event" + content_type = "application/json" + data = toolchain_event_prototype_data_model + + # Invoke method + response = _service.create_toolchain_event( + toolchain_id, + title, + description, + content_type, + data=data, + headers={}, + ) + + # Check for correct operation + assert len(responses.calls) == 1 + assert response.status_code == 200 + # Validate body params + req_body = json.loads(str(responses.calls[0].request.body, "utf-8")) + assert req_body["title"] == "My-custom-event" + assert req_body["description"] == "This is my custom event" + assert req_body["content_type"] == "application/json" + assert req_body["data"] == toolchain_event_prototype_data_model + + def test_create_toolchain_event_all_params_with_retries(self): + # Enable retries and run test_create_toolchain_event_all_params. + _service.enable_retries() + self.test_create_toolchain_event_all_params() + + # Disable retries and run test_create_toolchain_event_all_params. + _service.disable_retries() + self.test_create_toolchain_event_all_params() + + @responses.activate + def test_create_toolchain_event_value_error(self): + """ + test_create_toolchain_event_value_error() + """ + # Set up mock + url = preprocess_url("/toolchains/testString/events") + mock_response = '{"id": "id"}' + responses.add( + responses.POST, + url, + body=mock_response, + content_type="application/json", + status=200, + ) + + # Construct a dict representation of a ToolchainEventPrototypeDataApplicationJson model + toolchain_event_prototype_data_application_json_model = {} + toolchain_event_prototype_data_application_json_model["content"] = { + "anyKey": "anyValue" + } + + # Construct a dict representation of a ToolchainEventPrototypeData model + toolchain_event_prototype_data_model = {} + toolchain_event_prototype_data_model[ + "application_json" + ] = toolchain_event_prototype_data_application_json_model + toolchain_event_prototype_data_model[ + "text_plain" + ] = "This event is dispatched because the pipeline failed" + + # Set up parameter values + toolchain_id = "testString" + title = "My-custom-event" + description = "This is my custom event" + content_type = "application/json" + data = toolchain_event_prototype_data_model + + # Pass in all but one required param and check for a ValueError + req_param_dict = { + "toolchain_id": toolchain_id, + "title": title, + "description": description, + "content_type": content_type, + } + for param in req_param_dict.keys(): + req_copy = { + key: val if key is not param else None + for (key, val) in req_param_dict.items() + } + with pytest.raises(ValueError): + _service.create_toolchain_event(**req_copy) + + def test_create_toolchain_event_value_error_with_retries(self): + # Enable retries and run test_create_toolchain_event_value_error. + _service.enable_retries() + self.test_create_toolchain_event_value_error() + + # Disable retries and run test_create_toolchain_event_value_error. + _service.disable_retries() + self.test_create_toolchain_event_value_error() + + # endregion ############################################################################## # End of Service: Toolchains @@ -1733,6 +1868,154 @@ def test_toolchain_collection_previous_serialization(self): ) +class TestModel_ToolchainEventPost: + """ + Test Class for ToolchainEventPost + """ + + def test_toolchain_event_post_serialization(self): + """ + Test serialization/deserialization for ToolchainEventPost + """ + + # Construct a json representation of a ToolchainEventPost model + toolchain_event_post_model_json = {} + toolchain_event_post_model_json["id"] = "testString" + + # Construct a model instance of ToolchainEventPost by calling from_dict on the json representation + toolchain_event_post_model = ToolchainEventPost.from_dict( + toolchain_event_post_model_json + ) + assert toolchain_event_post_model != False + + # Construct a model instance of ToolchainEventPost by calling from_dict on the json representation + toolchain_event_post_model_dict = ToolchainEventPost.from_dict( + toolchain_event_post_model_json + ).__dict__ + toolchain_event_post_model2 = ToolchainEventPost( + **toolchain_event_post_model_dict + ) + + # Verify the model instances are equivalent + assert toolchain_event_post_model == toolchain_event_post_model2 + + # Convert model instance back to dict and verify no loss of data + toolchain_event_post_model_json2 = toolchain_event_post_model.to_dict() + assert toolchain_event_post_model_json2 == toolchain_event_post_model_json + + +class TestModel_ToolchainEventPrototypeData: + """ + Test Class for ToolchainEventPrototypeData + """ + + def test_toolchain_event_prototype_data_serialization(self): + """ + Test serialization/deserialization for ToolchainEventPrototypeData + """ + + # Construct dict forms of any model objects needed in order to build this model. + + toolchain_event_prototype_data_application_json_model = ( + {} + ) # ToolchainEventPrototypeDataApplicationJson + toolchain_event_prototype_data_application_json_model["content"] = { + "anyKey": "anyValue" + } + + # Construct a json representation of a ToolchainEventPrototypeData model + toolchain_event_prototype_data_model_json = {} + toolchain_event_prototype_data_model_json[ + "application_json" + ] = toolchain_event_prototype_data_application_json_model + toolchain_event_prototype_data_model_json[ + "text_plain" + ] = "This event is dispatched because the pipeline failed" + + # Construct a model instance of ToolchainEventPrototypeData by calling from_dict on the json representation + toolchain_event_prototype_data_model = ToolchainEventPrototypeData.from_dict( + toolchain_event_prototype_data_model_json + ) + assert toolchain_event_prototype_data_model != False + + # Construct a model instance of ToolchainEventPrototypeData by calling from_dict on the json representation + toolchain_event_prototype_data_model_dict = ( + ToolchainEventPrototypeData.from_dict( + toolchain_event_prototype_data_model_json + ).__dict__ + ) + toolchain_event_prototype_data_model2 = ToolchainEventPrototypeData( + **toolchain_event_prototype_data_model_dict + ) + + # Verify the model instances are equivalent + assert ( + toolchain_event_prototype_data_model + == toolchain_event_prototype_data_model2 + ) + + # Convert model instance back to dict and verify no loss of data + toolchain_event_prototype_data_model_json2 = ( + toolchain_event_prototype_data_model.to_dict() + ) + assert ( + toolchain_event_prototype_data_model_json2 + == toolchain_event_prototype_data_model_json + ) + + +class TestModel_ToolchainEventPrototypeDataApplicationJson: + """ + Test Class for ToolchainEventPrototypeDataApplicationJson + """ + + def test_toolchain_event_prototype_data_application_json_serialization(self): + """ + Test serialization/deserialization for ToolchainEventPrototypeDataApplicationJson + """ + + # Construct a json representation of a ToolchainEventPrototypeDataApplicationJson model + toolchain_event_prototype_data_application_json_model_json = {} + toolchain_event_prototype_data_application_json_model_json["content"] = { + "anyKey": "anyValue" + } + + # Construct a model instance of ToolchainEventPrototypeDataApplicationJson by calling from_dict on the json representation + toolchain_event_prototype_data_application_json_model = ( + ToolchainEventPrototypeDataApplicationJson.from_dict( + toolchain_event_prototype_data_application_json_model_json + ) + ) + assert toolchain_event_prototype_data_application_json_model != False + + # Construct a model instance of ToolchainEventPrototypeDataApplicationJson by calling from_dict on the json representation + toolchain_event_prototype_data_application_json_model_dict = ( + ToolchainEventPrototypeDataApplicationJson.from_dict( + toolchain_event_prototype_data_application_json_model_json + ).__dict__ + ) + toolchain_event_prototype_data_application_json_model2 = ( + ToolchainEventPrototypeDataApplicationJson( + **toolchain_event_prototype_data_application_json_model_dict + ) + ) + + # Verify the model instances are equivalent + assert ( + toolchain_event_prototype_data_application_json_model + == toolchain_event_prototype_data_application_json_model2 + ) + + # Convert model instance back to dict and verify no loss of data + toolchain_event_prototype_data_application_json_model_json2 = ( + toolchain_event_prototype_data_application_json_model.to_dict() + ) + assert ( + toolchain_event_prototype_data_application_json_model_json2 + == toolchain_event_prototype_data_application_json_model_json + ) + + class TestModel_ToolchainModel: """ Test Class for ToolchainModel