from __future__ import annotations import hashlib import hmac import json from typing import Any from fastapi.testclient import TestClient from sqlalchemy import select from gitea_codex_bot.main import app from gitea_codex_bot.db import get_session_factory from gitea_codex_bot.models import ReviewJob 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) monkeypatch.setattr( "gitea_codex_bot.services.gitea.GiteaClient.get_pull_request", lambda *_args, **_kwargs: type("PR", (), {"head_sha": "abcdef123"})(), ) monkeypatch.setattr("gitea_codex_bot.services.gitea.GiteaClient.get_file_content", lambda *_args, **_kwargs: None) 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 session_factory = get_session_factory() with session_factory() as session: queued = session.execute(select(ReviewJob).where(ReviewJob.trigger_comment_id == 111)).scalar_one() assert queued.trigger_comment_body == "@codex review security" def test_webhook_accepts_review_for_bot_username_alias(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) monkeypatch.setattr( "gitea_codex_bot.services.gitea.GiteaClient.get_pull_request", lambda *_args, **_kwargs: type("PR", (), {"head_sha": "abcdef123"})(), ) monkeypatch.setattr("gitea_codex_bot.services.gitea.GiteaClient.get_file_content", lambda *_args, **_kwargs: None) client = TestClient(app) payload_obj = _payload("@codex-bot review security", username="alice", comment_id=311) raw = json.dumps(payload_obj).encode() response = client.post( "/webhook/gitea", content=raw, headers={ "X-Gitea-Event": "issue_comment", "X-Gitea-Delivery": "d-2-alias", "X-Gitea-Signature": _sign(raw), "Content-Type": "application/json", }, ) assert response.status_code == 200 assert response.json()["status"] == "queued" assert posted_comments session_factory = get_session_factory() with session_factory() as session: queued = session.execute(select(ReviewJob).where(ReviewJob.trigger_comment_id == 311)).scalar_one() assert queued.trigger_comment_body == "@codex-bot review security" def test_webhook_uses_latest_pr_head_sha_when_config_lookup_fails(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) monkeypatch.setattr( "gitea_codex_bot.services.gitea.GiteaClient.get_pull_request", lambda *_args, **_kwargs: type("PR", (), {"head_sha": "newsha123"})(), ) monkeypatch.setattr( "gitea_codex_bot.services.gitea.GiteaClient.get_file_content", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("config unavailable")), ) client = TestClient(app) payload_obj = _payload("@codex review", username="alice", comment_id=112) payload_obj["pull_request"]["head"]["sha"] = "oldsha999" raw = json.dumps(payload_obj).encode() response = client.post( "/webhook/gitea", content=raw, headers={ "X-Gitea-Event": "issue_comment", "X-Gitea-Delivery": "d-2b", "X-Gitea-Signature": _sign(raw), "Content-Type": "application/json", }, ) assert response.status_code == 200 assert response.json()["status"] == "queued" assert any("`newsha1`" in body for body in posted_comments) session_factory = get_session_factory() with session_factory() as session: queued = session.execute(select(ReviewJob).where(ReviewJob.trigger_comment_id == 112)).scalar_one() assert queued.head_sha == "newsha123" def test_webhook_logs_when_no_codex_review_command(monkeypatch) -> None: messages: list[str] = [] def _log_info(message: str, *args, **_kwargs) -> None: messages.append(message % args if args else message) monkeypatch.setattr("gitea_codex_bot.main.logger.info", _log_info) client = TestClient(app) payload_obj = _payload("hello world", username="alice", comment_id=222) raw = json.dumps(payload_obj).encode() response = client.post( "/webhook/gitea", content=raw, headers={ "X-Gitea-Event": "issue_comment", "X-Gitea-Delivery": "d-3", "X-Gitea-Signature": _sign(raw), "Content-Type": "application/json", }, ) assert response.status_code == 200 assert response.json()["reason"] == "no codex command" assert any("Webhook ignored: no @codex review command" in item for item in messages) def test_webhook_logs_when_codex_command_is_not_review(monkeypatch) -> None: messages: list[str] = [] def _log_info(message: str, *args, **_kwargs) -> None: messages.append(message % args if args else message) monkeypatch.setattr("gitea_codex_bot.main.logger.info", _log_info) client = TestClient(app) payload_obj = _payload("@codex explain", username="alice", comment_id=223) raw = json.dumps(payload_obj).encode() response = client.post( "/webhook/gitea", content=raw, headers={ "X-Gitea-Event": "issue_comment", "X-Gitea-Delivery": "d-4", "X-Gitea-Signature": _sign(raw), "Content-Type": "application/json", }, ) assert response.status_code == 200 assert response.json()["status"] == "queued" assert any("Webhook without @codex review command" in item for item in messages) def test_webhook_accepts_help_short_flag_and_queues(monkeypatch) -> None: monkeypatch.setattr( "gitea_codex_bot.services.gitea.GiteaClient.get_pull_request", lambda *_args, **_kwargs: type("PR", (), {"head_sha": "abcdef123"})(), ) client = TestClient(app) payload_obj = _payload("@codex -h", username="alice", comment_id=333) raw = json.dumps(payload_obj).encode() response = client.post( "/webhook/gitea", content=raw, headers={ "X-Gitea-Event": "issue_comment", "X-Gitea-Delivery": "d-help-1", "X-Gitea-Signature": _sign(raw), "Content-Type": "application/json", }, ) assert response.status_code == 200 assert response.json()["status"] == "queued" session_factory = get_session_factory() with session_factory() as session: queued = session.execute(select(ReviewJob).where(ReviewJob.trigger_comment_id == 333)).scalar_one() assert queued.command == "help" def test_webhook_logs_when_repo_not_allowed(monkeypatch) -> None: messages: list[str] = [] def _log_info(message: str, *args, **_kwargs) -> None: messages.append(message % args if args else message) monkeypatch.setattr("gitea_codex_bot.main.logger.info", _log_info) client = TestClient(app) payload_obj = _payload("@codex review", username="alice", comment_id=225) payload_obj["repository"]["full_name"] = "acme/not-allowed" raw = json.dumps(payload_obj).encode() response = client.post( "/webhook/gitea", content=raw, headers={ "X-Gitea-Event": "issue_comment", "X-Gitea-Delivery": "d-6", "X-Gitea-Signature": _sign(raw), "Content-Type": "application/json", }, ) assert response.status_code == 200 assert response.json()["reason"] == "repo not allowed" assert any("Webhook ignored: repo not in ALLOWED_REPOS" in item for item in messages) def test_webhook_rejects_review_when_repo_config_disabled(monkeypatch) -> None: posted_comments: list[str] = [] monkeypatch.setattr( "gitea_codex_bot.services.gitea.GiteaClient.get_pull_request", lambda *_args, **_kwargs: type("PR", (), {"head_sha": "abcdef123"})(), ) monkeypatch.setattr( "gitea_codex_bot.services.gitea.GiteaClient.get_file_content", lambda *_args, **_kwargs: "enabled: false\n", ) monkeypatch.setattr( "gitea_codex_bot.services.gitea.GiteaClient.post_issue_comment", lambda _self, _repo, _pr, body: posted_comments.append(body) or 100, ) client = TestClient(app) payload_obj = _payload("@codex review", username="alice", comment_id=224) raw = json.dumps(payload_obj).encode() response = client.post( "/webhook/gitea", content=raw, headers={ "X-Gitea-Event": "issue_comment", "X-Gitea-Delivery": "d-5", "X-Gitea-Signature": _sign(raw), "Content-Type": "application/json", }, ) assert response.status_code == 200 assert response.json()["reason"] == "review disabled by repo config" assert any("Review is disabled" in body for body in posted_comments)