add artifact obj

Author: tarekbadrsh

cli/cdevents/cli/artifact.py

@@ -1,16 +1,13 @@
 """Module for cli artifact commands."""
 from __future__ import annotations
 
-import os
 from typing import List
-
 import click
 
 from cdevents.cli.utils import add_disclaimer_text, print_function_args
 from cdevents.cli.cdevents_command import CDeventsCommand
 
-from cdevents.core.events import Events
-from cdevents.core import event_type
+from cdevents.core.artifact import Artifact, ArtifactType
 
 # pylint: disable=unused-argument
 def common_artifact_options(function):
@@ -57,10 +54,10 @@ def packaged(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_artifact_event(event_type.ArtifactPackagedEventV1, id, name, version, data)
+    artifact = Artifact(ArtifactType.ArtifactPackagedEventV1, id, name, version)
+    artifact_event = artifact.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(artifact_event)
 
 
 @click.command(help=add_disclaimer_text("Artifact Published CloudEvent."))
@@ -72,8 +69,7 @@ def published(
     data: List[str] = None,
 ):
     print_function_args()
-
-    e = Events()
-    new_event = e.create_artifact_event(event_type.ArtifactPublishedEventV1, id, name, version, data)
+    artifact = Artifact(ArtifactType.ArtifactPublishedEventV1, id, name, version)
+    artifact_event = artifact.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(artifact_event)

core/cdevents/core/artifact.py

@@ -0,0 +1,37 @@
+"""artifact"""
+
+from enum import Enum
+from cdevents.core.events import Events
+
+class ArtifactType(Enum):
+    ArtifactPackagedEventV1: str = "cd.artifact.packaged.v1"
+    ArtifactPublishedEventV1: str = "cd.artifact.published.v1"
+
+
+class Artifact(Events):
+    """Events."""
+
+    def __init__(self, artifact_type: ArtifactType, id: str, name: str, version: str):
+        """Initializes class.
+        """
+        self._event_type = artifact_type
+        self._id = id
+        self._name = name
+        self._version = version
+    
+    def create_extensions(self):
+        """Create extensions.
+        """
+        extensions = {
+            "artifactid": self._id,
+            "artifactname": self._name,
+            "artifactversion": self._version,
+        }
+        return extensions
+    
+    def create_event(self, data: dict={}):
+        """Create artifact event.
+        """
+        extensions = self.create_extensions()
+        event = super().create_event(event_type=self._event_type.value, extensions=extensions, data=data)
+        return event

core/cdevents/core/event_type.py

@@ -1,8 +1,10 @@
 """Constants Event types."""
 
-# Artifact Events
-ArtifactPackagedEventV1  :str = "cd.artifact.packaged.v1"
-ArtifactPublishedEventV1 :str = "cd.artifact.published.v1"
+# pylint: TODO: 
+
+# # Artifact Events
+# ArtifactPackagedEventV1  :str = "cd.artifact.packaged.v1"
+# ArtifactPublishedEventV1 :str = "cd.artifact.published.v1"
 
 # Branch Events
 BranchCreatedEventV1 :str = "cd.repository.branch.created.v1"

core/cdevents/core/events.py

@@ -22,31 +22,18 @@ def create_event(self, event_type: str, extensions:dict, data = {}) -> CloudEven
 
         return event
 
-    def create_artifact_event(self, event_type: str , id: str, name: str, version: str, data = {}) -> CloudEvent:
-        """Create artifact event.
-        """
-        
-        extensions = {
-            "artifactid": id,
-            "artifactname": name,
-            "artifactversion": version,
-        }
-        event = self.create_event(event_type, extensions, data)
-
-        return event
-
-    def create_branch_event(self, event_type: str , id: str, name: str, repoid: str, data = {}) -> CloudEvent:
-        """Create branch event.
-        """
+    # def create_branch_event(self, event_type: str , id: str, name: str, repoid: str, data = {}) -> CloudEvent:
+    #     """Create branch event.
+    #     """
 
-        extensions = {
-            "branchid": id,
-            "branchname": name,
-            "branchrepositoryid": repoid,
-        }
-        event = self.create_event(event_type, extensions, data)
-
-        return event
+    #     extensions = {
+    #         "branchid": id,
+    #         "branchname": name,
+    #         "branchrepositoryid": repoid,
+    #     }
+    #     event = self.create_event(event_type, extensions, data)
+
+    #     return event
 
 
     def create_build_event(self, event_type: str , id: str, name: str, artifact: str, data = {}) -> CloudEvent:

core/tests/test_artifact.py

@@ -1,20 +1,22 @@
-from cdevents.core import event_type
 import pytest
 
-from cdevents.core.events import Events
+from cdevents.core.artifact import Artifact, ArtifactType
 
 @pytest.mark.unit
 def test_artifact_packaged_v1():
-    event = Events().create_artifact_event(event_type.ArtifactPackagedEventV1, id="_id", name="_name", version="_version", data={"artifact": "_artifact"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.ArtifactPackagedEventV1
-    assert event._attributes["extensions"] == {"artifactid": "_id", "artifactname": "_name", "artifactversion": "_version"}
-    assert event.data == {"artifact": "_artifact"}
+    artifact = Artifact(artifact_type=ArtifactType.ArtifactPackagedEventV1, id="_id", name="_name", version="_version")
+    artifact_event = artifact.create_event(data={"key1": "value1"})
+    assert artifact_event is not None
+    assert artifact_event._attributes["type"] == "cd.artifact.packaged.v1"
+    assert artifact_event._attributes["extensions"] == {"artifactid": "_id", "artifactname": "_name", "artifactversion": "_version"}
+    assert artifact_event.data == {"key1": "value1"}
 
 @pytest.mark.unit
 def test_artifact_published_v1():
-    event = Events().create_artifact_event(event_type.ArtifactPublishedEventV1, id="_id", name="_name", version="_version", data={"artifact": "_artifact"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.ArtifactPublishedEventV1
-    assert event._attributes["extensions"] == {"artifactid": "_id", "artifactname": "_name", "artifactversion": "_version"}
-    assert event.data == {"artifact": "_artifact"}
+    artifact = Artifact(artifact_type=ArtifactType.ArtifactPublishedEventV1, id="_id", name="_name", version="_version")
+    artifact_event = artifact.create_event(data={"key1": "value1"})
+    assert artifact_event is not None
+    assert artifact_event._attributes["type"] == "cd.artifact.published.v1"
+    assert artifact_event._attributes["extensions"] == {"artifactid": "_id", "artifactname": "_name", "artifactversion": "_version"}
+    assert artifact_event.data == {"key1": "value1"}
+