gitea-codex/tests/test_dispatcher.py

from __future__ import annotations

from types import SimpleNamespace

import httpx
from sqlalchemy import select

from gitea_codex_bot.config import get_settings
from gitea_codex_bot.db import get_session_factory
from gitea_codex_bot.models import ReviewJob
from gitea_codex_bot.services.comments import get_persistent_review_comment_id, upsert_persistent_review_comment_id
from gitea_codex_bot.services.jobs import enqueue_job
from gitea_codex_bot.services.repo_config import RepoReviewConfig
from gitea_codex_bot.types import ParsedCommand
from gitea_codex_bot.workers.dispatcher import process_one_job


def test_process_one_job_recreates_persistent_comment_when_edit_returns_404(monkeypatch) -> None:
    session_factory = get_session_factory()
    with session_factory() as session:
        job = enqueue_job(
            session,
            repo="acme/repo",
            pr_number=9,
            head_sha="deadbeef",
            trigger_comment_id=111,
            trigger_comment_body="@codex review",
            requested_by="alice",
            command=ParsedCommand(name="review", raw="@codex review"),
        )
        upsert_persistent_review_comment_id(
            session,
            repo=job.repo,
            pr_number=job.pr_number,
            head_sha=job.head_sha,
            comment_id=289,
        )

    monkeypatch.setattr(
        "gitea_codex_bot.workers.dispatcher.run_review_ephemeral",
        lambda *_args, **_kwargs: (
            {
                "verdict": "has_issues",
                "confidence": 0.7,
                "summary": "runner error",
                "findings": [],
            },
            RepoReviewConfig(configured=True, enabled=True),
        ),
    )

    class _FakeGiteaClient:
        def __init__(self, _settings) -> None:
            self.posted_comment_id = 0

        def get_pull_request(self, _repo: str, _pr_number: int):
            return SimpleNamespace(is_fork=False)

        def edit_issue_comment(self, _repo: str, _comment_id: int, _body: str) -> int:
            request = httpx.Request("PATCH", "https://gitea.test/api/v1/repos/acme/repo/issues/comments/289")
            response = httpx.Response(404, request=request, text='{"message":"not found"}')
            raise httpx.HTTPStatusError("not found", request=request, response=response)

        def post_issue_comment(self, _repo: str, _pr_number: int, _body: str) -> int:
            self.posted_comment_id = 990
            return self.posted_comment_id

    monkeypatch.setattr("gitea_codex_bot.workers.dispatcher.GiteaClient", _FakeGiteaClient)

    settings = get_settings()
    processed = process_one_job(settings)
    assert processed is True

    with session_factory() as session:
        persisted_comment_id = get_persistent_review_comment_id(session, "acme/repo", 9)
        assert persisted_comment_id == 990
        stored_job = session.execute(select(ReviewJob).where(ReviewJob.id == job.id)).scalar_one()
        assert stored_job.status.value == "succeeded"


def test_process_one_job_passes_full_trigger_message_to_runner(monkeypatch) -> None:
    captured: dict[str, str] = {}
    session_factory = get_session_factory()
    with session_factory() as session:
        enqueue_job(
            session,
            repo="acme/repo",
            pr_number=10,
            head_sha="cafebabe",
            trigger_comment_id=112,
            trigger_comment_body="@codex review security --full\nFocus auth/session handling.",
            requested_by="alice",
            command=ParsedCommand(name="review", raw="@codex review security --full", arguments=["security", "--full"]),
        )

    def _fake_run_review_ephemeral(_settings, *, repo: str, pr_number: int, command: ParsedCommand):
        captured["raw"] = command.raw
        return {"verdict": "correct", "confidence": 0.9, "summary": "ok", "findings": []}, RepoReviewConfig(configured=True, enabled=True)

    class _FakeGiteaClient:
        def __init__(self, _settings) -> None:
            pass

        def get_pull_request(self, _repo: str, _pr_number: int):
            return SimpleNamespace(is_fork=False)

        def post_issue_comment(self, _repo: str, _pr_number: int, _body: str) -> int:
            return 901

        def edit_issue_comment(self, _repo: str, _comment_id: int, _body: str) -> int:
            return _comment_id

    monkeypatch.setattr("gitea_codex_bot.workers.dispatcher.run_review_ephemeral", _fake_run_review_ephemeral)
    monkeypatch.setattr("gitea_codex_bot.workers.dispatcher.GiteaClient", _FakeGiteaClient)

    settings = get_settings()
    processed = process_one_job(settings)
    assert processed is True
    assert captured["raw"] == "@codex review security --full\nFocus auth/session handling."


def test_process_one_job_skips_review_when_repo_config_disabled(monkeypatch) -> None:
    posted_comments: list[str] = []
    session_factory = get_session_factory()
    with session_factory() as session:
        job = enqueue_job(
            session,
            repo="acme/repo",
            pr_number=11,
            head_sha="badc0de",
            trigger_comment_id=113,
            trigger_comment_body="@codex review",
            requested_by="alice",
            command=ParsedCommand(name="review", raw="@codex review"),
        )

    monkeypatch.setattr(
        "gitea_codex_bot.workers.dispatcher.run_review_ephemeral",
        lambda *_args, **_kwargs: (
            {"verdict": "correct", "confidence": 1.0, "summary": "ok", "findings": []},
            RepoReviewConfig(configured=True, enabled=False),
        ),
    )

    class _FakeGiteaClient:
        def __init__(self, _settings) -> None:
            pass

        def get_pull_request(self, _repo: str, _pr_number: int):
            return SimpleNamespace(is_fork=False)

        def post_issue_comment(self, _repo: str, _pr_number: int, body: str) -> int:
            posted_comments.append(body)
            return 902

    monkeypatch.setattr("gitea_codex_bot.workers.dispatcher.GiteaClient", _FakeGiteaClient)

    settings = get_settings()
    processed = process_one_job(settings)
    assert processed is True
    assert any("Review is disabled" in body for body in posted_comments)

    with session_factory() as session:
        stored_job = session.execute(select(ReviewJob).where(ReviewJob.id == job.id)).scalar_one()
        assert stored_job.status.value == "skipped"