add repository obj

Author: tarekbadrsh

cli/cdevents/cli/repository.py

@@ -1,16 +1,12 @@
 """Module for cli repository 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.repository import Repository, RepositoryType
 
 # pylint: disable=unused-argument
 def common_repository_options(function):
@@ -57,10 +53,10 @@ def created(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_repository_event(event_type.RepositoryCreatedEventV1, id, name, url, data)
+    repository = Repository(repository_type=RepositoryType.RepositoryCreatedEventV1, id=id, name=name, url=url)
+    pipelinerun_event = repository.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(pipelinerun_event)
 
 
 @click.command(help=add_disclaimer_text("Repository Modified CloudEvent."))
@@ -72,10 +68,10 @@ def modified(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_repository_event(event_type.RepositoryModifiedEventV1, id, name, url, data)
+    repository = Repository(repository_type=RepositoryType.RepositoryModifiedEventV1, id=id, name=name, url=url)
+    pipelinerun_event = repository.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(pipelinerun_event)
 
 
 @click.command(help=add_disclaimer_text("Repository Deleted CloudEvent."))
@@ -87,8 +83,8 @@ def deleted(
     data: List[str] = None,
 ):
     print_function_args()
-    e = Events()
-    new_event = e.create_repository_event(event_type.RepositoryModifiedEventV1, id, name, url, data)
+    repository = Repository(repository_type=RepositoryType.RepositoryDeletedEventV1, id=id, name=name, url=url)
+    pipelinerun_event = repository.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(pipelinerun_event)
 

core/cdevents/core/events.py

@@ -22,19 +22,6 @@ def create_event(self, event_type: str, extensions:dict, data = {}) -> CloudEven
 
         return event
 
-    def create_repository_event(self, event_type: str , id: str, name: str, url: str, data = {}) -> CloudEvent:
-        """Create repository event.
-        """
-            
-        extensions = {
-            "repositoryid": id,
-            "repositoryname": name,
-            "repositoryurl": url,
-        }
-
-        event = self.create_event(event_type, extensions, data)
-
-        return event
 
 
     def create_service_event(self, event_type: str , envid: str, name: str, version: str, data = {}) -> CloudEvent:

core/cdevents/core/repository.py

@@ -0,0 +1,38 @@
+"""repository"""
+
+from enum import Enum
+from cdevents.core.events import Events
+
+class RepositoryType(Enum):
+    RepositoryCreatedEventV1  :str = "cd.repository.created.v1"
+    RepositoryModifiedEventV1 :str = "cd.repository.modified.v1"
+    RepositoryDeletedEventV1  :str = "cd.repository.deleted.v1"
+
+
+class Repository(Events):
+    """Repository."""
+
+    def __init__(self, repository_type: RepositoryType, id: str, name: str, url: str):
+        """Initializes class.
+        """
+        self._event_type = repository_type
+        self._id = id
+        self._name = name
+        self._url = url
+        
+    def create_extensions(self):
+        """Create extensions.
+        """
+        extensions = {
+            "repositoryid": self._id,
+            "repositoryname": self._name,
+            "repositoryurl": self._url,
+        }
+        return extensions
+    
+    def create_event(self, data: dict={}):
+        """Create Repository event.
+        """
+        extensions = self.create_extensions()
+        event = super().create_event(event_type=self._event_type.value, extensions=extensions, data=data)
+        return event

core/tests/test_repository.py

@@ -1,28 +1,34 @@
 from cdevents.core import event_type
 import pytest
 
-from cdevents.core.events import Events
+from cdevents.core.repository import Repository, RepositoryType
 
 @pytest.mark.unit
 def test_repository_created():
-    event = Events().create_repository_event(event_type.RepositoryCreatedEventV1, id="_id", name="_name", url="_url", data={"repository": "_repository"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.RepositoryCreatedEventV1
-    assert event._attributes["extensions"] ==  {"repositoryid": "_id", "repositoryname": "_name", "repositoryurl": "_url"}
-    assert event.data == {"repository": "_repository"}
+    repository = Repository(repository_type=RepositoryType.RepositoryCreatedEventV1, id="_id", name="_name", url="_url")
+    repository_event = repository.create_event(data={"key1": "value1"})
+    assert repository_event is not None
+    assert repository_event._attributes["type"] == RepositoryType.RepositoryCreatedEventV1.value
+    assert repository_event._attributes["extensions"] == {"repositoryid": "_id", "repositoryname": "_name", "repositoryurl": "_url"}
+    assert repository_event.data == {"key1": "value1"}
+
 
 @pytest.mark.unit
 def test_repository_modified():
-    event = Events().create_repository_event(event_type.RepositoryModifiedEventV1, id="_id", name="_name", url="_url", data={"repository": "_repository"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.RepositoryModifiedEventV1
-    assert event._attributes["extensions"] ==  {"repositoryid": "_id", "repositoryname": "_name", "repositoryurl": "_url"}
-    assert event.data == {"repository": "_repository"}
+    repository = Repository(repository_type=RepositoryType.RepositoryModifiedEventV1, id="_id", name="_name", url="_url")
+    repository_event = repository.create_event(data={"key1": "value1"})
+    assert repository_event is not None
+    assert repository_event._attributes["type"] == RepositoryType.RepositoryModifiedEventV1.value
+    assert repository_event._attributes["extensions"] == {"repositoryid": "_id", "repositoryname": "_name", "repositoryurl": "_url"}
+    assert repository_event.data == {"key1": "value1"}
+
 
 @pytest.mark.unit
 def test_repository_deleted():
-    event = Events().create_repository_event(event_type.RepositoryDeletedEventV1, id="_id", name="_name", url="_url", data={"repository": "_repository"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.RepositoryDeletedEventV1
-    assert event._attributes["extensions"] ==  {"repositoryid": "_id", "repositoryname": "_name", "repositoryurl": "_url"}
-    assert event.data == {"repository": "_repository"}
+    repository = Repository(repository_type=RepositoryType.RepositoryDeletedEventV1, id="_id", name="_name", url="_url")
+    repository_event = repository.create_event(data={"key1": "value1"})
+    assert repository_event is not None
+    assert repository_event._attributes["type"] == RepositoryType.RepositoryDeletedEventV1.value
+    assert repository_event._attributes["extensions"] == {"repositoryid": "_id", "repositoryname": "_name", "repositoryurl": "_url"}
+    assert repository_event.data == {"key1": "value1"}
+