implement support for forgejo commit statuses

ogr/services/forgejo/__init__.py

@@ -1,12 +1,14 @@
 # Copyright Contributors to the Packit project.
 # SPDX-License-Identifier: MIT
 
+from ogr.services.forgejo.commit_flag import ForgejoCommitFlag
 from ogr.services.forgejo.issue import ForgejoIssue
 from ogr.services.forgejo.project import ForgejoProject
 from ogr.services.forgejo.pull_request import ForgejoPullRequest
 from ogr.services.forgejo.service import ForgejoService
 
 __all__ = [
+    ForgejoCommitFlag.__name__,
     ForgejoPullRequest.__name__,
     ForgejoIssue.__name__,
     ForgejoProject.__name__,

ogr/services/forgejo/commit_flag.py

@@ -0,0 +1,107 @@
+# Copyright Contributors to the Packit project.
+# SPDX-License-Identifier: MIT
+
+from datetime import datetime
+from typing import Any
+
+from pyforgejo import PyforgejoApi
+
+from ogr.abstract import CommitFlag, CommitStatus
+
+
+class ForgejoCommitFlag(CommitFlag):
+    """
+    CommitFlag implementation for Forgejo.
+    """
+
+    @classmethod
+    def _state_from_str(cls, state: str) -> CommitStatus:
+        # Convert a status string returned by Forgejo API into CommitStatus enum.
+        state = state.lower()
+        if state == "success":
+            return CommitStatus.success
+        if state == "failure":
+            return CommitStatus.failure
+        if state == "pending":
+            return CommitStatus.pending
+        raise ValueError(f"Unknown commit state from Forgejo: {state}")
+
+    @classmethod
+    def _validate_state(cls, state: CommitStatus) -> CommitStatus:
+        # Validate that the provided state is acceptable for Forgejo.
+        valid_states = {
+            CommitStatus.success,
+            CommitStatus.failure,
+            CommitStatus.pending,
+        }
+        if state in valid_states:
+            return state
+        raise ValueError(f"Invalid commit state for Forgejo: {state}")
+
+    def _from_raw_commit_flag(self) -> None:
+        """
+        Populate attributes from the raw commit flag data obtained from Forgejo's API.
+        Expected keys in self._raw_commit_flag: 'commit', 'state', 'context', 'comment', 'id',
+        'created', and 'updated'.
+        """
+        raw = self._raw_commit_flag
+        self.commit = raw.get("commit")
+        self.state = self._state_from_str(raw.get("state", "pending"))
+        self.context = raw.get("context")
+        self.comment = raw.get("comment")
+        self.uid = raw.get("id")
+        self.url = raw.get("url")
+        self._created = raw.get("created")
+        self._edited = raw.get("updated")
+
+    @staticmethod
+    def get(project: Any, commit: str) -> list["CommitFlag"]:
+        """
+        Retrieve commit statuses for the given commit from Forgejo using the pyforgejo SDK.
+        """
+        client = PyforgejoApi(api_url=project.forge_api_url, token=project.token)
+        raw_flags = client.get_commit_statuses(
+            owner=project.owner,
+            repo=project.repo,
+            commit=commit,
+        )
+        return [
+            ForgejoCommitFlag(raw_commit_flag=raw_flag, project=project, commit=commit)
+            for raw_flag in raw_flags
+        ]
+
+    @staticmethod
+    def set(
+        project: Any,
+        commit: str,
+        state: CommitStatus,
+        target_url: str,
+        description: str,
+        context: str,
+    ) -> "CommitFlag":
+        """
+        Set a new commit status on Forgejo via the pyforgejo SDK.
+        """
+        client = PyforgejoApi(api_url=project.forge_api_url, token=project.token)
+        raw_response = client.set_commit_status(
+            owner=project.owner,
+            repo=project.repo,
+            commit=commit,
+            state=state.name.lower(),
+            target_url=target_url,
+            description=description,
+            context=context,
+        )
+        return ForgejoCommitFlag(
+            raw_commit_flag=raw_response,
+            project=project,
+            commit=commit,
+        )
+
+    @property
+    def created(self) -> datetime:
+        return self._created
+
+    @property
+    def edited(self) -> datetime:
+        return self._edited

tests/integration/forgejo/conftest.py

@@ -15,3 +15,13 @@ def service():
         instance_url="https://v10.next.forgejo.org",
         api_key=api_key,
     )
+
+
+@pytest.fixture
+def project(service):
+    repo = os.environ.get("FORGEJO_REPO", "existing_repo_name")
+    namespace = os.environ.get("FORGEJO_NAMESPACE", "existing_namespace")
+    project = service.get_project(repo=repo, namespace=namespace)
+    if not project:
+        pytest.skip(f"Project {namespace}/{repo} does not exist.")
+    return project

tests/integration/forgejo/test_commit_flag.py

@@ -0,0 +1,163 @@
+# Copyright Contributors to the Packit project.
+# SPDX-License-Identifier: MIT
+
+import datetime
+
+import pytest
+
+from ogr.abstract import CommitFlag, CommitStatus
+from ogr.services.forgejo.commit_flag import ForgejoCommitFlag
+
+
+def fake_get_commit_statuses(self, owner, repo, commit):
+    return [
+        {
+            "commit": commit,
+            "state": "success",
+            "context": "CI",
+            "comment": "All tests passed",
+            "id": "123",
+            "url": f"http://dummy-forgejo/commit/{commit}/status",
+            "created": datetime.datetime.strptime(
+                "2023-01-01T12:00:00Z",
+                "%Y-%m-%dT%H:%M:%SZ",
+            ),
+            "updated": datetime.datetime.strptime(
+                "2023-01-01T12:30:00Z",
+                "%Y-%m-%dT%H:%M:%SZ",
+            ),
+        },
+    ]
+
+
+def fake_set_commit_status(
+    self,
+    owner,
+    repo,
+    commit,
+    state,
+    target_url,
+    description,
+    context,
+):
+    return {
+        "commit": commit,
+        "state": state,
+        "context": context,
+        "comment": description,
+        "id": "456",
+        "url": f"http://dummy-forgejo/commit/{commit}/status",
+        "created": datetime.datetime.strptime(
+            "2023-02-01T12:00:00Z",
+            "%Y-%m-%dT%H:%M:%SZ",
+        ),
+        "updated": datetime.datetime.strptime(
+            "2023-02-01T12:30:00Z",
+            "%Y-%m-%dT%H:%M:%SZ",
+        ),
+    }
+
+
+class FakePyforgejoApi:
+    def __init__(self, api_url, token):
+        self.api_url = api_url
+        self.token = token
+
+    def get_commit_statuses(self, owner, repo, commit):
+        return fake_get_commit_statuses(self, owner, repo, commit)
+
+    def set_commit_status(
+        self,
+        owner,
+        repo,
+        commit,
+        state,
+        target_url,
+        description,
+        context,
+    ):
+        return fake_set_commit_status(
+            self,
+            owner,
+            repo,
+            commit,
+            state,
+            target_url,
+            description,
+            context,
+        )
+
+
+class MockProject:
+    forge_api_url = "http://dummy-forgejo/api/v1"
+    owner = "dummy_owner"
+    repo = "dummy_repo"
+    token = "dummy_token"
+
+    def get_auth_header(self) -> dict[str, str]:
+        return {"Authorization": "Bearer dummy_token"}
+
+
+@pytest.fixture(autouse=True)
+def patch_pyforgejo_api(monkeypatch):
+    monkeypatch.setattr(
+        "ogr.services.forgejo.commit_flag.PyforgejoApi",
+        FakePyforgejoApi,
+    )
+
+
+def test_get_commit_flag_integration():
+    project = MockProject()
+    commit = "abcdef123456"
+
+    flags: list[CommitFlag] = ForgejoCommitFlag.get(project, commit)
+
+    assert len(flags) == 1
+    flag = flags[0]
+    assert flag.commit == commit
+    assert flag.state == CommitStatus.success
+    assert flag.context == "CI"
+    assert flag.comment == "All tests passed"
+    assert flag.uid == "123"
+
+    expected_created = datetime.datetime.strptime(
+        "2023-01-01T12:00:00Z",
+        "%Y-%m-%dT%H:%M:%SZ",
+    )
+    expected_updated = datetime.datetime.strptime(
+        "2023-01-01T12:30:00Z",
+        "%Y-%m-%dT%H:%M:%SZ",
+    )
+    assert flag.created == expected_created
+    assert flag.edited == expected_updated
+
+
+def test_set_commit_flag_integration():
+    project = MockProject()
+    commit = "abcdef123456"
+
+    flag = ForgejoCommitFlag.set(
+        project=project,
+        commit=commit,
+        state=CommitStatus.success,
+        target_url="http://dummy-target",
+        description="Build succeeded",
+        context="CI",
+    )
+
+    assert flag.commit == commit
+    assert flag.state == CommitStatus.success
+    assert flag.context == "CI"
+    assert flag.comment == "Build succeeded"
+    assert flag.uid == "456"
+
+    expected_created = datetime.datetime.strptime(
+        "2023-02-01T12:00:00Z",
+        "%Y-%m-%dT%H:%M:%SZ",
+    )
+    expected_updated = datetime.datetime.strptime(
+        "2023-02-01T12:30:00Z",
+        "%Y-%m-%dT%H:%M:%SZ",
+    )
+    assert flag.created == expected_created
+    assert flag.edited == expected_updated