packit · olamidepeterojo · Mar 19, 2025 · Mar 23, 2025
diff --git a/ogr/services/forgejo/__init__.py b/ogr/services/forgejo/__init__.py
@@ -2,11 +2,13 @@
 # SPDX-License-Identifier: MIT
 
 from ogr.services.forgejo.issue import ForgejoIssue
+from ogr.services.forgejo.pr import ForgejoPullRequest
 from ogr.services.forgejo.project import ForgejoProject
 from ogr.services.forgejo.pull_request import ForgejoPullRequest
+from ogr.services.forgejo.release import ForgejoRelease
 from ogr.services.forgejo.service import ForgejoService
 
 __all__ = [
+    ForgejoRelease.__name__,
     ForgejoPullRequest.__name__,
     ForgejoIssue.__name__,
     ForgejoProject.__name__,

diff --git a/ogr/services/forgejo/pr.py b/ogr/services/forgejo/pr.py
@@ -0,0 +1,139 @@
+# Copyright Contributors to the Packit project.
+# SPDX-License-Identifier: MIT
+
+import datetime
+from typing import Any, Optional
+
+from ogr.abstract import GitProject, MergeCommitStatus, PRLabel, PRStatus, PullRequest
+
+
+class ForgejoPullRequest(PullRequest):
+    """
+    Implementation of PullRequest for Forgejo.
+    """
+
+    def __init__(self, raw_pr: Any, project: "GitProject") -> None:
+        super().__init__(raw_pr, project)
+
+    @property
+    def title(self) -> str:
+        return self._raw_pr["title"]
+
+    @title.setter
+    def title(self, new_title: str) -> None:
+        self._raw_pr["title"] = new_title
+
+    @property
+    def id(self) -> int:
+        return self._raw_pr["id"]
+
+    @property
+    def status(self) -> PRStatus:
+        status_value = self._raw_pr.get("status") or self._raw_pr.get("state")
+        if not status_value:
+            raise KeyError("Neither 'status' nor 'state' key found in PR data.")
+        for member in PRStatus:
+            if member.name.lower() == status_value.lower():
+                return member
+        raise ValueError(f"Invalid PR status: {status_value}")
+
+    @property
+    def url(self) -> str:
+        return self._raw_pr.get("url") or self._raw_pr.get("html_url", "")
+
+    @property
+    def description(self) -> str:
+        return self._raw_pr.get("body", "")
+
+    @description.setter
+    def description(self, new_description: str) -> None:
+        self._raw_pr["description"] = new_description
+
+    @property
+    def author(self) -> str:
+        return self._raw_pr.get("author") or self._raw_pr.get("user", {}).get(
+            "login",
+            "",
+        )
+
+    @property
+    def source_branch(self) -> str:
+        return self._raw_pr.get("source_branch") or self._raw_pr.get("head", {}).get(
+            "ref",
+            "",
+        )
+
+    @property
+    def source_project(self) -> "GitProject":
+        return self._target_project
+
+    @property
+    def target_branch(self) -> str:
+        return self._raw_pr.get("target_branch") or self._raw_pr.get("base", {}).get(
+            "ref",
+            "",
+        )
+
+    @property
+    def created(self) -> datetime.datetime:
+        created_str = self._raw_pr.get("created") or self._raw_pr.get("created_at")
+        if not created_str:
+            raise KeyError("Missing 'created' or 'created_at' key in PR data")
+        return datetime.datetime.strptime(created_str, "%Y-%m-%dT%H:%M:%SZ")
+
+    @property
+    def labels(self) -> list[PRLabel]:
+        return [PRLabel(label) for label in self._raw_pr.get("labels", [])]
+
+    @property
+    def diff_url(self) -> str:
+        return self._raw_pr["diff_url"]
+
+    @property
+    def patch(self) -> bytes:
+        return self._raw_pr.get("patch", b"")
+
+    @property
+    def head_commit(self) -> str:
+        return self._raw_pr["head_commit"]
+
+    @property
+    def merge_commit_sha(self) -> str:
+        return self._raw_pr.get("merge_commit_sha", "")
+
+    @property
+    def merge_commit_status(self) -> MergeCommitStatus:
+        return MergeCommitStatus[
+            self._raw_pr.get("merge_commit_status", "unknown").upper()
+        ]
+
+    @staticmethod
+    def create(
+        project: Any,
+        title: str,
+        body: str,
+        target_branch: str,
+        source_branch: str,
+        fork_username: Optional[str] = None,
+    ) -> "ForgejoPullRequest":
+        """
+        Create a new pull request in Forgejo.
+        """
+        pr_data = {
+            "title": title,
+            "body": body,
+            "target_branch": target_branch,
+            "source_branch": source_branch,
+        }
+        raw_pr = project.create_pull_request(pr_data)  # Simulated API call
+        return ForgejoPullRequest(raw_pr, project)
+
+    def merge(self) -> "ForgejoPullRequest":
+        """Merge the pull request."""
+        self._raw_pr["status"] = "merged"
+        return self
+
+    def close(self) -> "ForgejoPullRequest":
+        """Close the pull request."""
+        self._raw_pr["status"] = "closed"
+        return self
diff --git a/ogr/services/forgejo/release.py b/ogr/services/forgejo/release.py
@@ -0,0 +1,159 @@
+# Copyright Contributors to the Packit project.
+# SPDX-License-Identifier: MIT
+
+import datetime
+from typing import Any, Optional
+
+import requests
+
+
+class GitTag:
+    """
+    Class representing a git tag.
+
+    Attributes:
+        name (str): Name of the tag.
+        commit_sha (str): Commit hash of the tag.
+    """
+
+    def __init__(self, name: str, commit_sha: str) -> None:
+        self.name = name
+        self.commit_sha = commit_sha
+
+    def __str__(self) -> str:
+        return f"GitTag(name={self.name}, commit_sha={self.commit_sha})"
+
+
+class ForgejoGitTag(GitTag):
+    """
+    Concrete implementation of GitTag for Forgejo.
+    """
+
+    # Using the base implementation as no additional functionality is needed.
+
+
+class Release:
+    """
+    Object that represents release.
+    """
+
+    def __init__(self, raw_release: Any, project: Any) -> None:
+        self._raw_release = raw_release
+        self.project = project
+
+
+class ForgejoRelease(Release):
+    """
+    Concrete implementation of Release for Forgejo.
+    """
+
+    @property
+    def title(self) -> str:
+        return self._raw_release.get("name", "")
+
+    @property
+    def body(self) -> str:
+        return self._raw_release.get("description", "")
+
+    @property
+    def git_tag(self) -> GitTag:
+        return ForgejoGitTag(self.tag_name, self._raw_release.get("commit_sha", ""))
+
+    @property
+    def tag_name(self) -> str:
+        return self._raw_release.get("tag_name", "")
+
+    @property
+    def url(self) -> Optional[str]:
+        return self._raw_release.get("html_url")
+
+    @property
+    def created_at(self) -> datetime.datetime:
+        created_str = self._raw_release.get("created_at", "")
+        return (
+            datetime.datetime.strptime(created_str, "%Y-%m-%dT%H:%M:%SZ")
+            if created_str
+            else datetime.datetime.now()
+        )
+
+    @property
+    def tarball_url(self) -> str:
+        return self._raw_release.get("tarball_url", "")
+
+    @staticmethod
+    def get(
+        project: Any,
+        identifier: Optional[int] = None,
+        name: Optional[str] = None,
+        tag_name: Optional[str] = None,
+    ) -> "Release":
+        url = f"{project.forge_api_url}/repos/{project.owner}/{project.repo}/releases"
+        params: dict[str, int | str] = {}
+        if identifier is not None:
+            params["id"] = identifier
+        if name is not None:
+            params["name"] = name
+        if tag_name is not None:
+            params["tag"] = tag_name
+        headers = project.get_auth_header()
+        response = requests.get(url, params=params, headers=headers)
+        response.raise_for_status()
+        return ForgejoRelease(response.json(), project)
+
+    @staticmethod
+    def get_latest(project: Any) -> Optional["Release"]:
+        url = f"{project.forge_api_url}/repos/{project.owner}/{project.repo}/releases/latest"
+        headers = project.get_auth_header()
+        response = requests.get(url, headers=headers)
+        if response.status_code != 200:
+            return None
+        return ForgejoRelease(response.json(), project)
+
+    @staticmethod
+    def get_list(project: Any) -> list["Release"]:
+        url = f"{project.forge_api_url}/repos/{project.owner}/{project.repo}/releases"
+        headers = project.get_auth_header()
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+        return [ForgejoRelease(release, project) for release in response.json()]
+
+    @staticmethod
+    def create(
+        project: Any,
+        tag: str,
+        name: str,
+        message: str,
+        ref: Optional[str] = None,
+    ) -> "ForgejoRelease":
+        url = f"{project.forge_api_url}/repos/{project.owner}/{project.repo}/releases"
+        payload = {
+            "tag_name": tag,
+            "name": name,
+            "body": message,
+            "target_commitish": ref,
+        }
+        headers = project.get_auth_header()
+        response = requests.post(url, json=payload, headers=headers)
+        response.raise_for_status()
+        return ForgejoRelease(response.json(), project)
+
+    def save_archive(self, filename: str) -> None:
+        tarball_url = self.tarball_url
+        if not tarball_url:
+            raise ValueError("Tarball URL is not available")
+        headers = self.project.get_auth_header()
+        response = requests.get(tarball_url, headers=headers)
+        response.raise_for_status()
+        with open(filename, "wb") as f:
+            f.write(response.content)
+
+    def edit_release(self, name: str, message: str) -> None:
+        url = (
+            f"{self.project.forge_api_url}/repos/{self.project.owner}/"
+            f"{self.project.repo}/releases/{self.tag_name}"
+        )
+        payload = {"name": name, "body": message}
+        headers = self.project.get_auth_header()
+        response = requests.patch(url, json=payload, headers=headers)
+        response.raise_for_status()
+        self._raw_release = response.json()
diff --git a/tests/integration/forgejo/conftest.py b/tests/integration/forgejo/conftest.py
@@ -15,3 +15,11 @@ def service():
         instance_url="https://v10.next.forgejo.org",
         api_key=api_key,
     )
+
+
+@pytest.fixture
+def project(service):
+    return service.get_project(
+        repo="ogr",
+        namespace="packit-service",
+    )