add PipelineRun obj

Author: tarekbadrsh

cli/cdevents/cli/env.py

@@ -1,9 +1,6 @@
 """Module for cli environment commands."""
 from __future__ import annotations
-
-import os
 from typing import List
-
 import click
 
 from cdevents.cli.utils import add_disclaimer_text, print_function_args

cli/cdevents/cli/pipelinerun.py

@@ -1,16 +1,12 @@
 """Module for cli pipelinerun 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.pipelinerun import Pipelinerun, PipelinerunType
 
 # pylint: disable=unused-argument
 def common_pipelinerun_options(function):
@@ -73,11 +69,10 @@ def started(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_pipelinerun_event(event_type.PipelineRunStartedEventV1, id, name, status, url, errors, data)
+    pipelinerun = Pipelinerun(PipelinerunType.PipelineRunStartedEventV1, id=id, name=name, status=status, url=url, errors=errors)
+    pipelinerun_event = pipelinerun.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
-
+    cdevents_command.run(pipelinerun_event)
 
 @click.command(help=add_disclaimer_text("PipelineRun Finished CloudEvent."))
 @common_pipelinerun_options
@@ -90,10 +85,11 @@ def finished(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_pipelinerun_event(event_type.PipelineRunFinishedEventV1, id, name, status, url, errors, data)
+    pipelinerun = Pipelinerun(PipelinerunType.PipelineRunFinishedEventV1, id=id, name=name, status=status, url=url, errors=errors)
+    pipelinerun_event = pipelinerun.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(pipelinerun_event)
+
 
 @click.command(help=add_disclaimer_text("PipelineRun Queued CloudEvent."))
 @common_pipelinerun_options
@@ -106,8 +102,9 @@ def queued(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_pipelinerun_event(event_type.PipelineRunQueuedEventV1, id, name, status, url, errors, data)
+    pipelinerun = Pipelinerun(PipelinerunType.PipelineRunQueuedEventV1, id=id, name=name, status=status, url=url, errors=errors)
+    pipelinerun_event = pipelinerun.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(pipelinerun_event)
+
 

core/cdevents/core/event_type.py

@@ -9,11 +9,6 @@
 ChangeMergedEventV1    :str = "cd.repository.change.merged.v1"
 ChangeAbandonedEventV1 :str = "cd.repository.change.abandoned.v1"
 
-# PipelineRun events
-PipelineRunStartedEventV1  :str = "cd.pipelinerun.started.v1"
-PipelineRunFinishedEventV1 :str = "cd.pipelinerun.finished.v1"
-PipelineRunQueuedEventV1   :str = "cd.pipelinerun.queued.v1"
-
 # Repository events
 RepositoryCreatedEventV1  :str = "cd.repository.created.v1"
 RepositoryModifiedEventV1 :str = "cd.repository.modified.v1"

core/cdevents/core/events.py

@@ -22,21 +22,6 @@ def create_event(self, event_type: str, extensions:dict, data = {}) -> CloudEven
 
         return event
 
-    def create_pipelinerun_event(self, event_type: str , id: str, name: str, status: str, url: str, errors: str, data = {}) -> CloudEvent:
-        """Create pipelinerun event.
-        """
-            
-        extensions = {
-            "pipelinerunid": id,
-            "pipelinerunname": name,
-            "pipelinerunstatus": status,
-            "pipelinerunurl": url,
-            "pipelinerunerrors": errors,
-        }
-        event = self.create_event(event_type, extensions, data)
-
-        return event
-
     def create_repository_event(self, event_type: str , id: str, name: str, url: str, data = {}) -> CloudEvent:
         """Create repository event.
         """

core/cdevents/core/pipelinerun.py

@@ -0,0 +1,42 @@
+"""pipelinerun"""
+
+from enum import Enum
+from cdevents.core.events import Events
+
+class PipelinerunType(Enum):
+    PipelineRunStartedEventV1  :str = "cd.pipelinerun.started.v1"
+    PipelineRunFinishedEventV1 :str = "cd.pipelinerun.finished.v1"
+    PipelineRunQueuedEventV1   :str = "cd.pipelinerun.queued.v1"
+
+
+class Pipelinerun(Events):
+    """Pipelinerun."""
+
+    def __init__(self, pipelinerun_type: PipelinerunType, id: str, name: str, status: str, url: str, errors: str):
+        """Initializes class.
+        """
+        self._event_type = pipelinerun_type
+        self._id = id
+        self._name = name
+        self._status = status
+        self._url = url
+        self._errors = errors
+        
+    def create_extensions(self):
+        """Create extensions.
+        """
+        extensions = {
+            "pipelinerunid": self._id,
+            "pipelinerunname": self._name,
+            "pipelinerunstatus": self._status,
+            "pipelinerunurl": self._url,
+            "pipelinerunerrors": self._errors,
+        }
+        return extensions
+    
+    def create_event(self, data: dict={}):
+        """Create pipelinerun event.
+        """
+        extensions = self.create_extensions()
+        event = super().create_event(event_type=self._event_type.value, extensions=extensions, data=data)
+        return event

core/tests/test_environment.py

@@ -1,4 +1,3 @@
-from cdevents.core import event_type
 import pytest
 
 from cdevents.core.env import Env, EnvType

core/tests/test_pipelinerun.py

@@ -1,29 +1,30 @@
-from pickle import NONE
-from cdevents.core import event_type
 import pytest
 
-from cdevents.core.events import Events
+from cdevents.core.pipelinerun import Pipelinerun, PipelinerunType
 
 @pytest.mark.unit
 def test_pipelinerun_started():
-    event = Events().create_pipelinerun_event(event_type.PipelineRunStartedEventV1, id="_id", name="_name", status="_status", url="_url", errors="_errors", data={"pipelinerun": "_pipelinerun"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.PipelineRunStartedEventV1
-    assert event._attributes["extensions"] ==  {"pipelinerunid": "_id", "pipelinerunname": "_name", "pipelinerunstatus": "_status", "pipelinerunurl": "_url", "pipelinerunerrors": "_errors"}
-    assert event.data == {"pipelinerun": "_pipelinerun"}
+    pipelinerun = Pipelinerun(pipelinerun_type=PipelinerunType.PipelineRunStartedEventV1, id="_id", name="_name", status="_status", url="_url", errors="_errors")
+    pipelinerun_event = pipelinerun.create_event(data={"key1": "value1"})
+    assert pipelinerun_event is not None
+    assert pipelinerun_event._attributes["type"] == PipelinerunType.PipelineRunStartedEventV1.value
+    assert pipelinerun_event._attributes["extensions"] == {"pipelinerunid": "_id", "pipelinerunname": "_name", "pipelinerunstatus": "_status", "pipelinerunurl": "_url", "pipelinerunerrors": "_errors"}
+    assert pipelinerun_event.data == {"key1": "value1"}
 
 @pytest.mark.unit
 def test_pipelinerun_finished():
-    event = Events().create_pipelinerun_event(event_type.PipelineRunFinishedEventV1, id="_id", name="_name", status="_status", url="_url", errors="_errors", data={"pipelinerun": "_pipelinerun"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.PipelineRunFinishedEventV1
-    assert event._attributes["extensions"] ==  {"pipelinerunid": "_id", "pipelinerunname": "_name", "pipelinerunstatus": "_status", "pipelinerunurl": "_url", "pipelinerunerrors": "_errors"}
-    assert event.data == {"pipelinerun": "_pipelinerun"}
+    pipelinerun = Pipelinerun(pipelinerun_type=PipelinerunType.PipelineRunFinishedEventV1, id="_id", name="_name", status="_status", url="_url", errors="_errors")
+    pipelinerun_event = pipelinerun.create_event(data={"key1": "value1"})
+    assert pipelinerun_event is not None
+    assert pipelinerun_event._attributes["type"] == PipelinerunType.PipelineRunFinishedEventV1.value
+    assert pipelinerun_event._attributes["extensions"] == {"pipelinerunid": "_id", "pipelinerunname": "_name", "pipelinerunstatus": "_status", "pipelinerunurl": "_url", "pipelinerunerrors": "_errors"}
+    assert pipelinerun_event.data == {"key1": "value1"}
 
 @pytest.mark.unit
 def test_pipelinerun_queued():
-    event = Events().create_pipelinerun_event(event_type.PipelineRunQueuedEventV1, id="_id", name="_name", status="_status", url="_url", errors="_errors", data={"pipelinerun": "_pipelinerun"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.PipelineRunQueuedEventV1
-    assert event._attributes["extensions"] ==  {"pipelinerunid": "_id", "pipelinerunname": "_name", "pipelinerunstatus": "_status", "pipelinerunurl": "_url", "pipelinerunerrors": "_errors"}
-    assert event.data == {"pipelinerun": "_pipelinerun"}
+    pipelinerun = Pipelinerun(pipelinerun_type=PipelinerunType.PipelineRunQueuedEventV1, id="_id", name="_name", status="_status", url="_url", errors="_errors")
+    pipelinerun_event = pipelinerun.create_event(data={"key1": "value1"})
+    assert pipelinerun_event is not None
+    assert pipelinerun_event._attributes["type"] == PipelinerunType.PipelineRunQueuedEventV1.value
+    assert pipelinerun_event._attributes["extensions"] == {"pipelinerunid": "_id", "pipelinerunname": "_name", "pipelinerunstatus": "_status", "pipelinerunurl": "_url", "pipelinerunerrors": "_errors"}
+    assert pipelinerun_event.data == {"key1": "value1"}