First MVP

tests/conftest.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+import os
+
+import pytest
+
+from gitea_codex_bot.config import get_settings
+from gitea_codex_bot.db import Base, get_engine, get_session_factory
+
+
+@pytest.fixture(autouse=True)
+def _env_defaults(monkeypatch: pytest.MonkeyPatch, tmp_path, request: pytest.FixtureRequest) -> Generator[None, None, None]:
+    monkeypatch.setenv("GITEA_BASE_URL", "https://gitea.test")
+    monkeypatch.setenv("GITEA_TOKEN", "token")
+    monkeypatch.setenv("GITEA_BOT_USERNAME", "codex-bot")
+    monkeypatch.setenv("GITEA_WEBHOOK_SECRET", "secret")
+    monkeypatch.setenv("OPENAI_API_KEY", "openai-key")
+    monkeypatch.setenv("ALLOWED_REPOS", "acme/repo")
+    monkeypatch.setenv("COOLDOWN_SECONDS", "60")
+    monkeypatch.setenv("WEBHOOK_MODE", "repo")
+    monkeypatch.setenv("DB_HOST", "localhost")
+    monkeypatch.setenv("DB_PORT", "3306")
+    monkeypatch.setenv("DB_NAME", "ignored")
+    monkeypatch.setenv("DB_USER", "ignored")
+    monkeypatch.setenv("DB_PASSWORD", "ignored")
+    database_url = os.getenv("TEST_DATABASE_URL", "").strip() or f"sqlite+pysqlite:///{tmp_path / 'test.db'}"
+    monkeypatch.setenv("DATABASE_URL", database_url)
+    monkeypatch.setenv("WORKDIR", str(tmp_path / "work"))
+
+    get_settings.cache_clear()
+    get_engine.cache_clear()
+    get_session_factory.cache_clear()
+
+    engine = get_engine()
+    skip_schema = request.node.get_closest_marker("no_schema") is not None
+    if not skip_schema:
+        Base.metadata.create_all(bind=engine)
+    yield
+    if not skip_schema:
+        Base.metadata.drop_all(bind=engine)
+    get_settings.cache_clear()
+    get_engine.cache_clear()
+    get_session_factory.cache_clear()

tests/test_commands.py

@@ -0,0 +1,20 @@
+from gitea_codex_bot.services.commands import parse_command
+
+
+def test_parse_review_command_modes() -> None:
+    cmd = parse_command("@codex review security --full")
+    assert cmd is not None
+    assert cmd.name == "review"
+    assert cmd.mode == "security"
+    assert cmd.full is True
+
+
+def test_parse_fix_branch() -> None:
+    cmd = parse_command("@codex fix --branch finding 2")
+    assert cmd is not None
+    assert cmd.name == "fix"
+    assert cmd.branch_fix is True
+
+
+def test_invalid_command_returns_none() -> None:
+    assert parse_command("hello") is None

tests/test_config.py

@@ -0,0 +1,6 @@
+from gitea_codex_bot.config import get_settings
+
+
+def test_openai_api_key_required() -> None:
+    settings = get_settings()
+    assert settings.openai_api_key.get_secret_value() == "openai-key"

tests/test_jobs.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from sqlalchemy.exc import IntegrityError
+
+from gitea_codex_bot.db import get_session_factory
+from gitea_codex_bot.services.jobs import cooldown_remaining_seconds, enqueue_job, persist_webhook_event
+from gitea_codex_bot.types import ParsedCommand
+
+def test_persist_webhook_dedupe() -> None:
+    session_factory = get_session_factory()
+    with session_factory() as session:
+        first = persist_webhook_event(session, delivery_id="d1", event_name="issue_comment", repo="acme/repo", comment_id=1, payload=b"{}")
+        second = persist_webhook_event(session, delivery_id="d1", event_name="issue_comment", repo="acme/repo", comment_id=1, payload=b"{}")
+    assert first is True
+    assert second is False
+
+
+def test_enqueue_and_cooldown() -> None:
+    session_factory = get_session_factory()
+    with session_factory() as session:
+        cmd = ParsedCommand(name="review", raw="@codex review")
+        enqueue_job(session, repo="acme/repo", pr_number=42, head_sha="abc", trigger_comment_id=100, requested_by="user", command=cmd)
+        remaining = cooldown_remaining_seconds(session, "acme/repo", 42, 60)
+    assert remaining >= 0
+
+
+def test_trigger_comment_unique() -> None:
+    session_factory = get_session_factory()
+    with session_factory() as session:
+        cmd = ParsedCommand(name="review", raw="@codex review")
+        enqueue_job(session, repo="acme/repo", pr_number=7, head_sha="x", trigger_comment_id=321, requested_by="user", command=cmd)
+        try:
+            enqueue_job(session, repo="acme/repo", pr_number=7, head_sha="x", trigger_comment_id=321, requested_by="user", command=cmd)
+            duplicate_raised = False
+        except IntegrityError:
+            duplicate_raised = True
+            session.rollback()
+    assert duplicate_raised is True

tests/test_migrations.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from alembic import command
+from alembic.config import Config
+
+
+import pytest
+
+
+@pytest.mark.no_schema
+def test_alembic_upgrade_and_downgrade() -> None:
+    cfg = Config("alembic.ini")
+    command.upgrade(cfg, "head")
+    command.downgrade(cfg, "base")
+    command.upgrade(cfg, "head")

tests/test_security.py

@@ -0,0 +1,15 @@
+import hmac
+import hashlib
+
+from gitea_codex_bot.services.security import verify_gitea_signature
+
+
+def test_verify_signature_success() -> None:
+    payload = b'{"a":1}'
+    secret = "abc"
+    signature = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
+    assert verify_gitea_signature(payload, secret, signature)
+
+
+def test_verify_signature_failure() -> None:
+    assert not verify_gitea_signature(b"x", "abc", "deadbeef")

tests/test_transitions.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from gitea_codex_bot.db import get_session_factory
+from gitea_codex_bot.models import JobStatus, ReviewJob
+from gitea_codex_bot.services.jobs import claim_next_job, enqueue_job, finish_job
+from gitea_codex_bot.types import ParsedCommand
+
+
+def test_claim_and_transition() -> None:
+    session_factory = get_session_factory()
+    with session_factory() as session:
+        job = enqueue_job(
+            session,
+            repo="acme/repo",
+            pr_number=314,
+            head_sha="deadbeef",
+            trigger_comment_id=9901,
+            requested_by="alice",
+            command=ParsedCommand(name="review", raw="@codex review"),
+        )
+
+    with session_factory() as session:
+        claimed = claim_next_job(session)
+        assert claimed is not None
+        assert claimed.id == job.id
+        assert claimed.status == JobStatus.running
+
+    with session_factory() as session:
+        finish_job(session, job_id=job.id, success=True, skipped=False, result={"summary": "ok"}, error_message=None)
+
+    with session_factory() as session:
+        loaded = session.execute(select(ReviewJob).where(ReviewJob.id == job.id)).scalar_one()
+        assert loaded.status == JobStatus.succeeded
+        assert loaded.result_json is not None

tests/test_webhook.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+from typing import Any
+
+from fastapi.testclient import TestClient
+
+from gitea_codex_bot.main import app
+
+
+def _sign(payload: bytes) -> str:
+    return hmac.new(b"secret", payload, hashlib.sha256).hexdigest()
+
+
+def _payload(comment_body: str, *, username: str = "alice", comment_id: int = 11) -> dict[str, Any]:
+    return {
+        "repository": {"full_name": "acme/repo"},
+        "sender": {"username": username},
+        "comment": {"id": comment_id, "body": comment_body},
+        "issue": {"number": 9, "pull_request": {"url": "x"}},
+        "pull_request": {"head": {"sha": "abcdef123"}},
+    }
+
+
+def test_webhook_rejects_bad_signature() -> None:
+    client = TestClient(app)
+    payload = b"{}"
+    response = client.post(
+        "/webhook/gitea",
+        content=payload,
+        headers={"X-Gitea-Event": "issue_comment", "X-Gitea-Signature": "bad"},
+    )
+    assert response.status_code == 401
+
+
+def test_webhook_ignores_bot_comment(monkeypatch) -> None:
+    client = TestClient(app)
+    payload = _payload("@codex review", username="codex-bot")
+    raw = json.dumps(payload).encode()
+    response = client.post(
+        "/webhook/gitea",
+        content=raw,
+        headers={
+            "X-Gitea-Event": "issue_comment",
+            "X-Gitea-Delivery": "d-1",
+            "X-Gitea-Signature": _sign(raw),
+            "Content-Type": "application/json",
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["reason"] == "bot comment ignored"
+
+
+def test_webhook_accepts_review_and_queues(monkeypatch) -> None:
+    posted_comments: list[str] = []
+
+    def _post_issue_comment(self, repo: str, pr_number: int, body: str) -> int:
+        posted_comments.append(body)
+        return 100
+
+    monkeypatch.setattr("gitea_codex_bot.services.gitea.GiteaClient.post_issue_comment", _post_issue_comment)
+
+    client = TestClient(app)
+    payload_obj = _payload("@codex review security", username="alice", comment_id=111)
+    raw = json.dumps(payload_obj).encode()
+
+    response = client.post(
+        "/webhook/gitea",
+        content=raw,
+        headers={
+            "X-Gitea-Event": "issue_comment",
+            "X-Gitea-Delivery": "d-2",
+            "X-Gitea-Signature": _sign(raw),
+            "Content-Type": "application/json",
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["status"] == "queued"
+    assert posted_comments