Merge branch 'feature/18' into develop

TODO

@@ -219,7 +219,7 @@ Content-Type: application/issue
 ID: 18
 Type: feature
 Title: implement saas wrapper for Forgejo
-Status: in-progress
+Status: done
 Priority: medium
 Created: 2026-06-06
 Relationships: 

src/byteb4rb1e/utils/saas/forgejo.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+"""Forgejo REST API v1 wrapper.
+
+Thin layer over http.py for Forgejo-specific operations:
+
+- Token authentication
+- Repository existence checks
+- Repository creation under the authenticated user or an organization
+- SSH and HTTPS clone URL construction
+
+Unlike Bitbucket (one global SaaS instance), Forgejo is self-hosted,
+so every operation takes a *host* parameter instead of baking any
+specific instance in.
+"""
+
+import json
+from typing import Any, Dict, Optional
+
+from byteb4rb1e.utils.http import client as http_client
+
+
+def api_url(host: str) -> str:
+    """Return the API base URL for a Forgejo instance."""
+    return f"https://{host}/api/v1"
+
+
+def http_headers(token: str) -> Dict[str, str]:
+    """Construct Forgejo API headers with token auth."""
+    return {
+        "Authorization": f"token {token}",
+        "Accept": "application/json",
+        "Content-Type": "application/json",
+    }
+
+
+def repository_exists(
+    host: str,
+    owner: str,
+    repo_slug: str,
+    token: str,
+) -> bool:
+    """Check whether a repository exists under the owner."""
+    url = f"{api_url(host)}/repos/{owner}/{repo_slug}"
+    resp = http_client.get(url, headers=http_headers(token))
+    return bool(resp.status_code == 200)
+
+
+def create_repository(
+    host: str,
+    repo_slug: str,
+    token: str,
+    org: Optional[str] = None,
+    description: str = "",
+    is_private: bool = True,
+) -> http_client.HttpResponse:
+    """Create a new repository on the Forgejo instance.
+
+    When *org* is given the repository is created in that
+    organization, otherwise under the authenticated user.
+
+    Returns the API response.  Caller should check status_code == 201
+    for success.
+    """
+    if org:
+        url = f"{api_url(host)}/orgs/{org}/repos"
+    else:
+        url = f"{api_url(host)}/user/repos"
+    body: Dict[str, Any] = {
+        "name": repo_slug,
+        "private": is_private,
+        "description": description,
+    }
+    return http_client.post(
+        url,
+        data=json.dumps(body).encode("utf-8"),
+        headers=http_headers(token),
+    )
+
+
+def ssh_clone_url(
+    host: str,
+    owner: str,
+    repo_slug: str,
+) -> str:
+    """Return the SSH clone URL for a Forgejo repository."""
+    return f"git@{host}:{owner}/{repo_slug}.git"
+
+
+def https_clone_url(
+    host: str,
+    owner: str,
+    repo_slug: str,
+) -> str:
+    """Return the HTTPS clone URL for a Forgejo repository.
+
+    Preferred in CI environments without SSH host keys.
+    """
+    return f"https://{host}/{owner}/{repo_slug}.git"

tests/unit/byteb4rb1e/utils/saas/test_forgejo.py

@@ -0,0 +1,133 @@
+"""Tests for the Forgejo API wrapper."""
+
+import json
+from typing import Any, Dict, List, Optional, Tuple
+
+import pytest
+
+from byteb4rb1e.utils.http.client import HttpResponse
+from byteb4rb1e.utils.saas import forgejo
+
+HOST = "git.example.com"
+
+
+class _Recorder:
+    """Records http_client calls and replays a canned response."""
+
+    def __init__(self, response: HttpResponse) -> None:
+        self.calls: List[Tuple[str, Dict[str, Any]]] = []
+        self._response = response
+
+    def __call__(self, url: str, **kwargs: Any) -> HttpResponse:
+        self.calls.append((url, kwargs))
+        return self._response
+
+
+class TestApiUrl:
+
+    def test_host_only(self) -> None:
+        assert forgejo.api_url(HOST) == "https://git.example.com/api/v1"
+
+
+class TestHttpHeaders:
+
+    def test_token_header(self) -> None:
+        headers = forgejo.http_headers("s3cret")
+        assert headers["Authorization"] == "token s3cret"
+        assert headers["Accept"] == "application/json"
+        assert headers["Content-Type"] == "application/json"
+
+
+class TestRepositoryExists:
+
+    def test_exists(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        recorder = _Recorder(HttpResponse(200, {}, b"{}"))
+        monkeypatch.setattr(forgejo.http_client, "get", recorder)
+
+        assert forgejo.repository_exists(HOST, "tiara", "repo", "t") is True
+        url, kwargs = recorder.calls[0]
+        assert url == "https://git.example.com/api/v1/repos/tiara/repo"
+        assert kwargs["headers"]["Authorization"] == "token t"
+
+    def test_missing(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        recorder = _Recorder(HttpResponse(404, {}, b""))
+        monkeypatch.setattr(forgejo.http_client, "get", recorder)
+
+        assert forgejo.repository_exists(HOST, "tiara", "repo", "t") is False
+
+
+class TestCreateRepository:
+
+    def _create(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        org: Optional[str] = None,
+        **kwargs: Any,
+    ) -> _Recorder:
+        recorder = _Recorder(HttpResponse(201, {}, b"{}"))
+        monkeypatch.setattr(forgejo.http_client, "post", recorder)
+        forgejo.create_repository(HOST, "repo", "t", org=org, **kwargs)
+        return recorder
+
+    def test_user_repo_endpoint(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        recorder = self._create(monkeypatch)
+        url, _ = recorder.calls[0]
+        assert url == "https://git.example.com/api/v1/user/repos"
+
+    def test_org_repo_endpoint(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        recorder = self._create(monkeypatch, org="byteb4rb1e")
+        url, _ = recorder.calls[0]
+        assert url == "https://git.example.com/api/v1/orgs/byteb4rb1e/repos"
+
+    def test_body(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        recorder = self._create(
+            monkeypatch, description="demo", is_private=False,
+        )
+        _, kwargs = recorder.calls[0]
+        body = json.loads(kwargs["data"].decode("utf-8"))
+        assert body == {
+            "name": "repo",
+            "private": False,
+            "description": "demo",
+        }
+
+    def test_defaults_to_private(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        recorder = self._create(monkeypatch)
+        _, kwargs = recorder.calls[0]
+        body = json.loads(kwargs["data"].decode("utf-8"))
+        assert body["private"] is True
+
+    def test_auth_header(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        recorder = self._create(monkeypatch)
+        _, kwargs = recorder.calls[0]
+        assert kwargs["headers"]["Authorization"] == "token t"
+
+    def test_returns_response(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        response = HttpResponse(201, {}, b'{"id": 1}')
+        recorder = _Recorder(response)
+        monkeypatch.setattr(forgejo.http_client, "post", recorder)
+
+        resp = forgejo.create_repository(HOST, "repo", "t")
+
+        assert resp is response
+
+
+class TestCloneUrls:
+
+    def test_ssh(self) -> None:
+        assert forgejo.ssh_clone_url(HOST, "tiara", "repo") == (
+            "git@git.example.com:tiara/repo.git"
+        )
+
+    def test_https(self) -> None:
+        assert forgejo.https_clone_url(HOST, "tiara", "repo") == (
+            "https://git.example.com/tiara/repo.git"
+        )