143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
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)
|
|
|
|
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_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)
|