add taskrun obj

Author: tarekbadrsh

cli/cdevents/cli/taskrun.py

@@ -1,16 +1,12 @@
 """Module for cli taskrun 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.taskrun import TaskRun, TaskRunType
 
 # pylint: disable=unused-argument
 def common_taskrun_options(function):
@@ -57,10 +53,10 @@ def started(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_taskrun_event(event_type.TaskRunStartedEventV1, id, name, pipelineid, data)
+    taskrun = TaskRun(taskrun_type=TaskRunType.TaskRunStartedEventV1, id=id, name=name, pipelineid=pipelineid)
+    taskrun_event = taskrun.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(taskrun_event)
 
 @click.command(help=add_disclaimer_text("TaskRun Finished CloudEvent."))
 @common_taskrun_options
@@ -71,8 +67,8 @@ def finished(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_taskrun_event(event_type.TaskRunFinishedEventV1, id, name, pipelineid, data)
+    taskrun = TaskRun(taskrun_type=TaskRunType.TaskRunFinishedEventV1, id=id, name=name, pipelineid=pipelineid)
+    taskrun_event = taskrun.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(taskrun_event)
 

core/cdevents/core/event_type.py

@@ -9,10 +9,6 @@
 ChangeMergedEventV1    :str = "cd.repository.change.merged.v1"
 ChangeAbandonedEventV1 :str = "cd.repository.change.abandoned.v1"
 
-# TaskRun events
-TaskRunStartedEventV1  :str = "cd.taskrun.started.v1"
-TaskRunFinishedEventV1 :str = "cd.taskrun.finished.v1"
-
 # Test Events
 TestCaseStartedEventV1  :str = "cd.test.case.started.v1"
 TestCaseQueuedEventV1   :str = "cd.test.case.queued.v1"

core/cdevents/core/taskrun.py

@@ -0,0 +1,38 @@
+"""taskrun"""
+
+from enum import Enum
+from cdevents.core.events import Events
+
+class TaskRunType(Enum):
+    TaskRunStartedEventV1  :str = "cd.taskrun.started.v1"
+    TaskRunFinishedEventV1 :str = "cd.taskrun.finished.v1"
+
+
+class TaskRun(Events):
+    """Taskrun."""
+
+    def __init__(self, taskrun_type: TaskRunType, id: str, name: str, pipelineid: str):
+        """Initializes class.
+        """
+        self._event_type = taskrun_type
+        self._id= id
+        self._name = name
+        self._pipelineid = pipelineid
+
+        
+    def create_extensions(self):
+        """Create extensions.
+        """
+        extensions = {
+            "taskrunid": self._id,
+            "taskrunname": self._name,
+            "taskrunpipelineid": self._pipelineid,
+        }
+        return extensions
+    
+    def create_event(self, data: dict={}):
+        """Create taskrun event.
+        """
+        extensions = self.create_extensions()
+        event = super().create_event(event_type=self._event_type.value, extensions=extensions, data=data)
+        return event

core/tests/test_service.py

@@ -1,40 +1,44 @@
 from cdevents.core import event_type
 import pytest
 
-from cdevents.core.events import Events
+from cdevents.core.service import Service, ServiceType
 
 @pytest.mark.unit
 def test_service_deployed():
-    event = Events().create_service_event(event_type.ServiceDeployedEventV1, envid="_envid", name="_name", version="_version", data={"service": "_service"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.ServiceDeployedEventV1
-    assert event._attributes["extensions"] == {"serviceenvid": "_envid", "servicename": "_name", "serviceversion": "_version"}
-    assert event.data == {"service": "_service"}
+    service = Service(service_type=ServiceType.ServiceDeployedEventV1, envid="_envid", name="_name", version="_version")
+    service_event = service.create_event(data={"key1": "value1"})
+    assert service_event is not None
+    assert service_event._attributes["type"] == ServiceType.ServiceDeployedEventV1.value
+    assert service_event._attributes["extensions"] == {"serviceenvid": "_envid", "servicename": "_name", "serviceversion": "_version"}
+    assert service_event.data == {"key1": "value1"}
 
 
 @pytest.mark.unit
 def test_service_upgraded():
-    event = Events().create_service_event(event_type.ServiceUpgradedEventV1, envid="_envid", name="_name", version="_version", data={"service": "_service"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.ServiceUpgradedEventV1
-    assert event._attributes["extensions"] == {"serviceenvid": "_envid", "servicename": "_name", "serviceversion": "_version"}
-    assert event.data == {"service": "_service"}
+    service = Service(service_type=ServiceType.ServiceUpgradedEventV1, envid="_envid", name="_name", version="_version")
+    service_event = service.create_event(data={"key1": "value1"})
+    assert service_event is not None
+    assert service_event._attributes["type"] == ServiceType.ServiceUpgradedEventV1.value
+    assert service_event._attributes["extensions"] == {"serviceenvid": "_envid", "servicename": "_name", "serviceversion": "_version"}
+    assert service_event.data == {"key1": "value1"}
 
 
 @pytest.mark.unit
 def test_service_rolledback():
-    event = Events().create_service_event(event_type.ServiceRolledbackEventV1, envid="_envid", name="_name", version="_version", data={"service": "_service"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.ServiceRolledbackEventV1
-    assert event._attributes["extensions"] == {"serviceenvid": "_envid", "servicename": "_name", "serviceversion": "_version"}
-    assert event.data == {"service": "_service"}
+    service = Service(service_type=ServiceType.ServiceRolledbackEventV1, envid="_envid", name="_name", version="_version")
+    service_event = service.create_event(data={"key1": "value1"})
+    assert service_event is not None
+    assert service_event._attributes["type"] == ServiceType.ServiceRolledbackEventV1.value
+    assert service_event._attributes["extensions"] == {"serviceenvid": "_envid", "servicename": "_name", "serviceversion": "_version"}
+    assert service_event.data == {"key1": "value1"}
 
 
 @pytest.mark.unit
 def test_service_removed():
-    event = Events().create_service_event(event_type.ServiceRemovedEventV1, envid="_envid", name="_name", version="_version", data={"service": "_service"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.ServiceRemovedEventV1
-    assert event._attributes["extensions"] == {"serviceenvid": "_envid", "servicename": "_name", "serviceversion": "_version"}
-    assert event.data == {"service": "_service"}
+    service = Service(service_type=ServiceType.ServiceRemovedEventV1, envid="_envid", name="_name", version="_version")
+    service_event = service.create_event(data={"key1": "value1"})
+    assert service_event is not None
+    assert service_event._attributes["type"] == ServiceType.ServiceRemovedEventV1.value
+    assert service_event._attributes["extensions"] == {"serviceenvid": "_envid", "servicename": "_name", "serviceversion": "_version"}
+    assert service_event.data == {"key1": "value1"}
 

core/tests/test_taskrun.py

@@ -1,20 +1,22 @@
-from cdevents.core import event_type
 import pytest
 
-from cdevents.core.events import Events
+from cdevents.core.taskrun import TaskRun, TaskRunType
 
 @pytest.mark.unit
 def test_taskrun_started():
-    event = Events().create_taskrun_event(event_type.TaskRunStartedEventV1, id="_id", name="_name", pipelineid="_pipelineid", data={"taskrun": "_taskrun"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.TaskRunStartedEventV1
-    assert event._attributes["extensions"] == {"taskrunid": "_id", "taskrunname": "_name", "taskrunpipelineid": "_pipelineid"}
-    assert event.data == {"taskrun": "_taskrun"}
+    taskrun = TaskRun(taskrun_type=TaskRunType.TaskRunStartedEventV1,id="_id", name="_name", pipelineid="_pipelineid")
+    taskrun_event = taskrun.create_event(data={"key1": "value1"})
+    assert taskrun_event is not None
+    assert taskrun_event._attributes["type"] == TaskRunType.TaskRunStartedEventV1.value
+    assert taskrun_event._attributes["extensions"] == {"taskrunid": "_id", "taskrunname": "_name", "taskrunpipelineid": "_pipelineid"}
+    assert taskrun_event.data == {"key1": "value1"}
+
 
 @pytest.mark.unit
 def test_taskrun_finished():
-    event = Events().create_taskrun_event(event_type.TaskRunFinishedEventV1, id="_id", name="_name", pipelineid="_pipelineid", data={"taskrun": "_taskrun"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.TaskRunFinishedEventV1
-    assert event._attributes["extensions"] == {"taskrunid": "_id", "taskrunname": "_name", "taskrunpipelineid": "_pipelineid"}
-    assert event.data == {"taskrun": "_taskrun"}
+    taskrun = TaskRun(taskrun_type=TaskRunType.TaskRunFinishedEventV1,id="_id", name="_name", pipelineid="_pipelineid")
+    taskrun_event = taskrun.create_event(data={"key1": "value1"})
+    assert taskrun_event is not None
+    assert taskrun_event._attributes["type"] == TaskRunType.TaskRunFinishedEventV1.value
+    assert taskrun_event._attributes["extensions"] == {"taskrunid": "_id", "taskrunname": "_name", "taskrunpipelineid": "_pipelineid"}
+    assert taskrun_event.data == {"key1": "value1"}