146 lines
4.8 KiB
Python
146 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from gitea_codex_bot.db import get_session_factory
|
|
from gitea_codex_bot.main import app
|
|
from gitea_codex_bot.models import JobStatus, ReviewJob
|
|
|
|
|
|
def test_root_returns_tailwind_landing_page() -> None:
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].startswith("text/html")
|
|
assert "Gitea Codex Review Bot" in response.text
|
|
assert "cdn.tailwindcss.com" in response.text
|
|
assert 'id="health-button"' in response.text
|
|
assert 'id="failure-button"' in response.text
|
|
assert 'id="health-modal"' in response.text
|
|
assert 'fetch("/healthz"' in response.text
|
|
assert 'fetch("/healthz/latest-failure"' in response.text
|
|
|
|
|
|
def test_404_returns_tailwind_page_for_browser_requests() -> None:
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/missing", headers={"Accept": "text/html"})
|
|
|
|
assert response.status_code == 404
|
|
assert response.headers["content-type"].startswith("text/html")
|
|
assert "Error 404" in response.text
|
|
assert "cdn.tailwindcss.com" in response.text
|
|
|
|
|
|
def test_404_returns_json_for_non_browser_requests() -> None:
|
|
client = TestClient(app)
|
|
|
|
response = client.get("/missing", headers={"Accept": "application/json"})
|
|
|
|
assert response.status_code == 404
|
|
assert response.headers["content-type"].startswith("application/json")
|
|
assert response.json() == {"detail": "Not Found"}
|
|
|
|
|
|
def test_healthz_latest_failure_returns_empty_when_no_failed_jobs() -> None:
|
|
client = TestClient(app)
|
|
response = client.get("/healthz/latest-failure")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok", "has_failed_job": False}
|
|
|
|
|
|
def test_healthz_latest_failure_returns_latest_failed_job() -> None:
|
|
session_factory = get_session_factory()
|
|
with session_factory() as session:
|
|
first = ReviewJob(
|
|
repo="acme/repo",
|
|
pr_number=1,
|
|
head_sha="1111111",
|
|
trigger_comment_id=3001,
|
|
trigger_comment_body="@codex review",
|
|
command="review",
|
|
requested_by="alice",
|
|
status=JobStatus.failed,
|
|
last_error="first error",
|
|
)
|
|
second = ReviewJob(
|
|
repo="acme/repo",
|
|
pr_number=2,
|
|
head_sha="2222222",
|
|
trigger_comment_id=3002,
|
|
trigger_comment_body="@codex rerun",
|
|
command="rerun",
|
|
requested_by="bob",
|
|
status=JobStatus.failed,
|
|
last_error="second error",
|
|
)
|
|
session.add(first)
|
|
session.add(second)
|
|
session.commit()
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/healthz/latest-failure")
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["status"] == "ok"
|
|
assert payload["has_failed_job"] is True
|
|
assert payload["repo"] == "acme/repo"
|
|
assert payload["pr_number"] == 2
|
|
assert payload["command"] == "rerun"
|
|
assert payload["head_sha"] == "2222222"
|
|
assert payload["error"] == "second error"
|
|
|
|
|
|
def test_healthz_latest_job_returns_empty_when_no_jobs() -> None:
|
|
client = TestClient(app)
|
|
response = client.get("/healthz/latest-job")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok", "has_job": False}
|
|
|
|
|
|
def test_healthz_latest_job_returns_latest_job_details() -> None:
|
|
session_factory = get_session_factory()
|
|
with session_factory() as session:
|
|
first = ReviewJob(
|
|
repo="acme/repo",
|
|
pr_number=3,
|
|
head_sha="3333333",
|
|
trigger_comment_id=3003,
|
|
trigger_comment_body="@codex review",
|
|
command="review",
|
|
requested_by="alice",
|
|
status=JobStatus.succeeded,
|
|
result_json={"summary": "first summary"},
|
|
)
|
|
second = ReviewJob(
|
|
repo="acme/repo",
|
|
pr_number=4,
|
|
head_sha="4444444",
|
|
trigger_comment_id=3004,
|
|
trigger_comment_body="@codex rerun",
|
|
command="rerun",
|
|
requested_by="bob",
|
|
status=JobStatus.failed,
|
|
last_error="failed later",
|
|
result_json={"summary": "second summary"},
|
|
)
|
|
session.add(first)
|
|
session.add(second)
|
|
session.commit()
|
|
|
|
client = TestClient(app)
|
|
response = client.get("/healthz/latest-job")
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["status"] == "ok"
|
|
assert payload["has_job"] is True
|
|
assert payload["repo"] == "acme/repo"
|
|
assert payload["pr_number"] == 4
|
|
assert payload["command"] == "rerun"
|
|
assert payload["head_sha"] == "4444444"
|
|
assert payload["job_status"] == "failed"
|
|
assert payload["error"] == "failed later"
|
|
assert payload["result_summary"] == "second summary"
|