Add job handler to process jobs

Author: gerardsegarra

.github/workflows/tests.yaml

@@ -29,4 +29,4 @@ jobs:
           flake8
       - name: Test with pytest
         run: |
-          pytest --cov=src
+          pytest --cov=src tests/

src/app.py

@@ -9,6 +9,7 @@
 
 from const import GithubHeaders, LOGGING_CONFIG
 from github import GithubJob
+from jobs import JobEventsHandler
 from utils import dict_to_logfmt
 
 dictConfig(LOGGING_CONFIG)
@@ -26,6 +27,7 @@
 logging.getLogger('apscheduler.executors.default').setLevel(logging.WARNING)
 
 jobs = dict()
+job_handler = JobEventsHandler()
 
 
 # check all calls are valid
@@ -50,7 +52,10 @@ def validate_origin_github():
 
 
 def process_workflow_job():
-    job = GithubJob(request.get_json())
+    event = request.get_json()
+    job_handler.process_event(event)
+
+    job = GithubJob(event)
 
     context_details = {
         "action": job.action,

src/jobs.py

@@ -0,0 +1,57 @@
+from github import GithubJob
+
+
+class Job:
+    def __init__(self, github_job: GithubJob) -> None:
+        self.github_job = github_job
+
+
+class JobEventsHandler:
+    def __init__(self) -> None:
+        self.queued = dict()
+        self.in_progress = dict()
+
+    def process_event(self, event: dict):
+        status = event["action"]
+
+        if status == "queued":
+            self._process_queued_event(event)
+
+        elif status == "in_progress":
+            self._process_in_progress_event(event)
+
+        elif status == "completed":
+            self._process_completed_event(event)
+
+        else:
+            pass
+
+    def _get_event_job_id(self, event: dict):
+        return event["workflow_job"]["id"]
+
+    def _create_job(self, githubJob: GithubJob) -> Job:
+        return Job(github_job=githubJob)
+
+    def _process_queued_event(self, event: dict):
+        job = self._create_job(GithubJob(event))
+        self.queued[self._get_event_job_id(event)] = job
+
+    def _process_in_progress_event(self, event: dict):
+        job_id = self._get_event_job_id(event)
+        job = self.queued.pop(job_id, None)
+
+        if not job:
+            job = self._create_job(GithubJob(event))
+        else:
+            # Update github job event from job
+            job.github_job = GithubJob(event)
+
+        self.in_progress[job_id] = job
+
+        # TODO send final time in queue
+
+    def _process_completed_event(self, event: dict):
+        job_id = self._get_event_job_id(event)
+        self.in_progress.pop(job_id, None)
+
+        # TODO send final time in progress

tests/test_jobs.py

@@ -0,0 +1,56 @@
+import pytest
+
+from unittest.mock import Mock
+
+from jobs import JobEventsHandler
+
+
+@pytest.fixture
+def new_job_event():
+    return {"workflow_job": {"id": "workflow_id"}, "action": "queued"}
+
+
+@pytest.fixture
+def in_progress_job_event():
+    return {"workflow_job": {"id": "workflow_id"},  "action": "in_progress"}
+
+
+@pytest.fixture
+def completed_job_event():
+    return {"workflow_job": {"id": "workflow_id"},  "action": "completed"}
+
+
+def test_new_job(new_job_event):
+    handler = JobEventsHandler()
+
+    handler.process_event(new_job_event)
+
+    assert handler.queued.get("workflow_id")
+
+
+def test_in_progress_job(in_progress_job_event):
+    handler = JobEventsHandler()
+    job = Mock()
+    handler.queued["workflow_id"] = job
+
+    handler.process_event(in_progress_job_event)
+
+    assert not handler.queued.get("workflow_id")
+    assert handler.in_progress.get("workflow_id") == job
+
+
+def test_unprocessed_in_progress_job(in_progress_job_event):
+    handler = JobEventsHandler()
+    handler.process_event(in_progress_job_event)
+
+    assert handler.in_progress.get("workflow_id")
+
+
+def test_completed_job(completed_job_event):
+    handler = JobEventsHandler()
+    handler.in_progress["workflow_id"] = Mock()
+
+    handler.process_event(completed_job_event)
+
+    assert not handler.queued.get("workflow_id")
+    assert not handler.in_progress.get("workflow_id")