api: Remove cloudevents dependency

kernelci/api/__init__.py

@@ -13,7 +13,6 @@
 from typing import Dict, Optional, Sequence
 
 import requests
-from cloudevents.http import CloudEvent
 from requests.adapters import HTTPAdapter
 from urllib3.util import Retry
 
@@ -527,15 +526,15 @@ def send_event(self, channel: str, data):
         """Send an event to a given pub/sub channel"""
 
     @abc.abstractmethod
-    def receive_event(self, sub_id: int) -> CloudEvent:
+    def receive_event(self, sub_id: int) -> dict:
         """Listen and receive an event from a given subscription id"""
 
     @abc.abstractmethod
     def push_event(self, list_name: str, data):
         """Push an event to a given Redis List"""
 
     @abc.abstractmethod
-    def pop_event(self, list_name: str) -> CloudEvent:
+    def pop_event(self, list_name: str) -> dict:
         """Listen and pop an event from a given List"""
 
     @abc.abstractmethod

kernelci/api/helper.py

@@ -92,19 +92,19 @@ def unsubscribe_filters(self, sub_id):
         self.api.unsubscribe(sub_id)
 
     def receive_event_data(self, sub_id, block=True):
-        """Receive CloudEvent from Pub/Sub and return its data payload
+        """Receive an event from Pub/Sub and return its data payload
         If block is False, on receiving an "keep-alive" event,
         such as "BEEP" ping, it will return None instead of the data.
         Without this, it will block until an event is received.
         """
         event = self.api.receive_event(sub_id, block=block)
         if event is None:
             return None
-        return event.data
+        return event["data"]
 
     def pop_event_data(self, list_name):
-        """Receive CloudEvent from Redis list and return its data payload"""
-        return self.api.pop_event(list_name).data
+        """Receive an event from Redis list and return its data payload"""
+        return self.api.pop_event(list_name)["data"]
 
     def get_node_from_event(self, event_data):
         """Listen for an event and get the matching node object from it"""

kernelci/api/latest.py

@@ -10,8 +10,6 @@
 import json
 from typing import Dict, Optional, Sequence
 
-from cloudevents.http import from_json
-
 from . import API
 
 
@@ -188,8 +186,8 @@ def receive_event(self, sub_id: int, block: bool = True):
             data = resp.json().get("data")
             if not data:
                 continue
-            event = from_json(data)
-            if event.data == "BEEP":
+            event = json.loads(data)
+            if event.get("data") == "BEEP":
                 if not block:
                     # If block is False, return None
                     # for semi-nonblocking operation
@@ -202,11 +200,8 @@ def push_event(self, list_name: str, data):
 
     def pop_event(self, list_name: str):
         path = "/".join(["pop", str(list_name)])
-        while True:
-            resp = self._get(path)
-            data = json.dumps(resp.json())
-            event = from_json(data)
-            return event
+        resp = self._get(path)
+        return resp.json()
 
     def subscription_stats(self):
         return self._get("stats/subscriptions").json()

pyproject.toml

@@ -19,7 +19,6 @@ dependencies = [
   "azure-storage-file-share==12.24.0",
   "bson==0.5.10",
   "click==8.3.1",
-  "cloudevents==1.12.1",
   "docker==7.1.0",
   "jinja2==3.1.6",
   "kubernetes==35.0.0",

requirements.txt

@@ -2,7 +2,6 @@ azure-storage-blob==12.28.0
 azure-storage-file-share==12.24.0
 bson==0.5.10
 click==8.3.1
-cloudevents==1.12.1
 docker==7.1.0
 jinja2==3.1.6
 kubernetes==35.0.0

tests/api/conftest.py

@@ -7,10 +7,10 @@
 """pytest fixtures for APIHelper unit tests"""
 
 import json
+import uuid
+from datetime import datetime, timezone
 
 import pytest
-from cloudevents.conversion import to_json
-from cloudevents.http import CloudEvent
 from requests import Response
 
 import kernelci.config
@@ -180,16 +180,18 @@ def update_kunit_child_node(self):
         return self._kunit_child_node
 
     def get_test_cloud_event(self):
-        """Get test CloudEvent instance"""
-        attributes = {
-            "type": "api.kernelci.org",
+        """Get test CloudEvents JSON envelope as a dict"""
+        return {
+            "specversion": "1.0",
+            "id": str(uuid.uuid4()),
             "source": "https://api.kernelci.org/",
+            "type": "api.kernelci.org",
+            "time": datetime.now(timezone.utc).isoformat(),
+            "data": {
+                "op": "created",
+                "id": self.checkout_node["id"],
+            },
         }
-        data = {
-            "op": "created",
-            "id": self.checkout_node["id"],
-        }
-        return CloudEvent(attributes=attributes, data=data)
 
 
 @pytest.fixture
@@ -258,12 +260,12 @@ def mock_api_put_nodes(mocker):
 @pytest.fixture
 def mock_receive_event(mocker):
     """
-    Mocks call to LatestAPI class method used to receive CloudEvent
+    Mocks call to LatestAPI class method used to receive an event
     """
     resp = Response()
     resp.status_code = 200
     event = APIHelperTestData().get_test_cloud_event()
-    resp._content = to_json(event)
+    resp._content = json.dumps(event).encode("utf-8")
     mocker.patch(
         "kernelci.api.latest.LatestAPI.receive_event",
         return_value=event,

tests/api/test_api.py

@@ -52,12 +52,12 @@ def test_unsubscribe(get_api_config, mock_api_unsubscribe):
 
 
 def test_get_node_from_event(get_api_config, mock_api_get_node_from_id):
-    "Test method to get node from CloudEvent data"
+    "Test method to get node from event data"
     for _, api_config in get_api_config.items():
         api = kernelci.api.get_api(api_config)
         helper = kernelci.api.helper.APIHelper(api)
         node = helper.get_node_from_event(
-            APIHelperTestData().get_test_cloud_event()
+            APIHelperTestData().get_test_cloud_event()["data"]
         )
         assert node.keys() == {
             "id",