add branch obj

Author: tarekbadrsh

cli/cdevents/cli/branch.py

@@ -1,16 +1,14 @@
 """Module for cli branch 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.branch import Branch, BranchType
 
 # pylint: disable=unused-argument
 def common_branch_options(function):
@@ -56,11 +54,11 @@ def created(
     repoid: str = None,
     data: List[str] = None,
 ):
-    print_function_args() 
-    e = Events()
-    new_event = e.create_branch_event(event_type.BranchCreatedEventV1, id, name, repoid, data)
+    print_function_args()
+    branch = Branch(branch_type=BranchType.BranchCreatedEventV1, id=id, name=name, repoid=repoid)
+    branch_event = branch.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(branch_event)
 
 
 @click.command(help=add_disclaimer_text("Branch Deleted CloudEvent."))
@@ -71,8 +69,8 @@ def deleted(
     repoid: str = None,
     data: List[str] = None,
 ):
-    print_function_args() 
-    e = Events()
-    new_event = e.create_branch_event(event_type.BranchDeletedEventV1, id, name, repoid, data)
+    print_function_args()
+    branch = Branch(branch_type=BranchType.BranchDeletedEventV1, id=id, name=name, repoid=repoid)
+    branch_event = branch.create_event(data)
     cdevents_command = CDeventsCommand()
-    cdevents_command.run(new_event)
+    cdevents_command.run(branch_event)

cli/tests/test_branch.py

@@ -0,0 +1,77 @@
+"""Unit tests for artifact."""
+import pytest
+from unittest.mock import patch
+from click.testing import CliRunner
+
+from cdevents.cli.branch import created, deleted
+from cdevents.cli.cdevents_command import CDeventsCommand
+
+# pylint: disable=missing-function-docstring, protected-access, missing-class-docstring
+
+@pytest.fixture
+def runner() -> CliRunner:
+    return CliRunner()
+
+
+ID_ARG = "id"
+NAME_ARG = "name"
+REPOID_ARG = "repoid"
+DATA_ARG = "data"
+
+@pytest.mark.unit
+def test_created(runner: CliRunner):
+    """Test created of a branch."""
+
+    expected_id = "task1"
+    expected_name = "My Task Run"
+    expected_repoid = "repo1"
+    expected_data = ["key1", "value1"]
+
+    with patch.object(CDeventsCommand, "run", spec=CDeventsCommand): 
+        result = runner.invoke(
+            created,
+            [
+                f"--{ID_ARG}",
+                expected_id,
+                f"--{NAME_ARG}",
+                expected_name,
+                f"--{REPOID_ARG}",
+                expected_repoid,
+                f"--{DATA_ARG}",
+                *expected_data,
+            ],
+        )
+    assert result.exit_code == 0
+    assert f"{ID_ARG}={expected_id}" in result.stdout
+    assert f"{NAME_ARG}={expected_name}" in result.stdout
+    assert f"{REPOID_ARG}={expected_repoid}" in result.stdout
+    assert f"{DATA_ARG}=({tuple(expected_data)},)" in result.stdout
+
+@pytest.mark.unit
+def test_deleted(runner: CliRunner):
+    """Test deleted of a branch."""
+
+    expected_id = "task1"
+    expected_name = "My Task Run"
+    expected_repoid = "repo1"
+    expected_data = ["key1", "value1"]
+
+    with patch.object(CDeventsCommand, "run", spec=CDeventsCommand): 
+        result = runner.invoke(
+            deleted,
+            [
+                f"--{ID_ARG}",
+                expected_id,
+                f"--{NAME_ARG}",
+                expected_name,
+                f"--{REPOID_ARG}",
+                expected_repoid,
+                f"--{DATA_ARG}",
+                *expected_data,
+            ],
+        )
+    assert result.exit_code == 0
+    assert f"{ID_ARG}={expected_id}" in result.stdout
+    assert f"{NAME_ARG}={expected_name}" in result.stdout
+    assert f"{REPOID_ARG}={expected_repoid}" in result.stdout
+    assert f"{DATA_ARG}=({tuple(expected_data)},)" in result.stdout

core/cdevents/core/branch.py

@@ -0,0 +1,37 @@
+"""branch"""
+
+from enum import Enum
+from cdevents.core.events import Events
+
+class BranchType(Enum):
+    BranchCreatedEventV1: str = "cd.repository.branch.created.v1"
+    BranchDeletedEventV1: str = "cd.repository.branch.deleted.v1"
+
+
+class Branch(Events):
+    """Events."""
+
+    def __init__(self, branch_type: BranchType, id: str, name: str, repoid: str):
+        """Initializes class.
+        """
+        self._event_type = branch_type
+        self._id = id
+        self._name = name
+        self._repoid = repoid
+    
+    def create_extensions(self):
+        """Create extensions.
+        """
+        extensions = {
+            "branchid": self._id,
+            "branchname": self._name,
+            "branchrepositoryid": self._repoid,
+        }
+        return extensions
+    
+    def create_event(self, data: dict={}):
+        """Create branch event.
+        """
+        extensions = self.create_extensions()
+        event = super().create_event(event_type=self._event_type.value, extensions=extensions, data=data)
+        return event

core/tests/test_branch.py

@@ -1,20 +1,21 @@
-from cdevents.core import event_type
 import pytest
 
-from cdevents.core.events import Events
+from cdevents.core.branch import Branch, BranchType
 
 @pytest.mark.unit
 def test_repository_branch_created():
-    event = Events().create_branch_event(event_type.BranchCreatedEventV1, id="_id", name="_name", repoid="_repoid", data={"branch": "_branch"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.BranchCreatedEventV1
-    assert event._attributes["extensions"] == {"branchid": "_id", "branchname": "_name", "branchrepositoryid": "_repoid"}
-    assert event.data == {"branch": "_branch"}
+    branch = Branch(branch_type=BranchType.BranchCreatedEventV1, id="_id", name="_name", repoid="_repoid")
+    branch_event = branch.create_event(data={"key1": "value1"})
+    assert branch_event is not None
+    assert branch_event._attributes["type"] == "cd.repository.branch.created.v1"
+    assert branch_event._attributes["extensions"] == {"branchid": "_id", "branchname": "_name", "branchrepositoryid": "_repoid"}
+    assert branch_event.data == {"key1": "value1"}
 
 @pytest.mark.unit
 def test_repository_branch_deleted():
-    event = Events().create_branch_event(event_type.BranchDeletedEventV1, id="_id", name="_name", repoid="_repoid", data={"branch": "_branch"})
-    assert event is not None
-    assert event._attributes["type"] == event_type.BranchDeletedEventV1
-    assert event._attributes["extensions"] == {"branchid": "_id", "branchname": "_name", "branchrepositoryid": "_repoid"}
-    assert event.data == {"branch": "_branch"}
+    branch = Branch(branch_type=BranchType.BranchDeletedEventV1, id="_id", name="_name", repoid="_repoid")
+    branch_event = branch.create_event(data={"key1": "value1"})
+    assert branch_event is not None
+    assert branch_event._attributes["type"] == "cd.repository.branch.deleted.v1"
+    assert branch_event._attributes["extensions"] == {"branchid": "_id", "branchname": "_name", "branchrepositoryid": "_repoid"}
+    assert branch_event.data == {"key1": "value1"}