Cobb Hayes
b335405c02
URLs, mount paths, and LAN host bindings parameterized via env or relative paths
so the repo stands up from a clean clone anywhere. Drop cross-codebase refs
("mirrors clawdforge's pattern"), Sulkta-Coop client/merchant test fixtures,
and audit-changelog scaffolding from comments. README terser, technical content
preserved.
289 lines
8.6 KiB
Python
289 lines
8.6 KiB
Python
"""HTTP API tests for the wave-3 patches surface.
|
|
|
|
Covers POST /jobs/{id}/patches (manual trigger), GET /patches (list with
|
|
filters), GET /patches/{id} (detail with cross-token guards). The patcher
|
|
itself is stubbed so we don't make real clawdforge / Gitea calls.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from crafting_table.patcher import PatchAttempt
|
|
from tests.conftest import sample_project_payload
|
|
|
|
|
|
def _install_stub_patcher(server, *, attempt: PatchAttempt | None = None):
|
|
"""Replace server.patcher with a stub that returns the given attempt.
|
|
|
|
``attempt=None`` simulates the "no actionable finding" code path.
|
|
Returns the stub for assertion-side access.
|
|
"""
|
|
stub = MagicMock()
|
|
stub.maybe_draft = AsyncMock(return_value=attempt)
|
|
server.patcher = stub
|
|
return stub
|
|
|
|
|
|
def _register_demo_project(tc, bearer: str, *, name: str = "demo") -> None:
|
|
payload = sample_project_payload(name=name)
|
|
r = tc.post(
|
|
"/projects",
|
|
headers={"Authorization": f"Bearer {bearer}"},
|
|
json=payload,
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
def _seed_job_row(server, *, project_name: str = "demo", job_id: str = "j-1") -> None:
|
|
snapshot = {
|
|
"git_url": "/dev/null",
|
|
"default_branch": "main",
|
|
"subprojects": [{
|
|
"path": ".", "language": "python", "lint": "echo x"
|
|
}],
|
|
"languages": ["python"],
|
|
}
|
|
server.db.insert_job(
|
|
job_id=job_id,
|
|
project_name=project_name,
|
|
subproject_path=".",
|
|
recipe="lint",
|
|
branch="main",
|
|
log_path="/tmp/_x.log",
|
|
recipe_snapshot_json=json.dumps(snapshot),
|
|
)
|
|
server.db.mark_job_finished(job_id=job_id, status="failed", exit_code=1)
|
|
|
|
|
|
def test_post_patches_with_finding_id(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
|
|
fake = PatchAttempt(
|
|
finding_id=42,
|
|
job_id="j-1",
|
|
project_name="demo",
|
|
attempt_number=1,
|
|
status="pr_opened",
|
|
branch_name="crafting-table/auto/j-1-42",
|
|
pr_url="http://git.example.com/X/Y/pulls/9",
|
|
diff_excerpt="--- a/x\n+++ b/x",
|
|
session_id="s-1",
|
|
)
|
|
fake.id = 7
|
|
stub = _install_stub_patcher(server, attempt=fake)
|
|
|
|
r = tc.post(
|
|
"/jobs/j-1/patches",
|
|
headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"},
|
|
json={"finding_id": 42},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert body["attempt"]["status"] == "pr_opened"
|
|
assert body["attempt"]["pr_url"].endswith("/9")
|
|
# Patcher was called with the explicit finding_id.
|
|
assert stub.maybe_draft.await_count == 1
|
|
args, kwargs = stub.maybe_draft.call_args
|
|
assert kwargs.get("finding_id") == 42 or (len(args) > 1 and args[1] == 42)
|
|
|
|
|
|
def test_post_patches_without_finding_id_auto_picks(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
|
|
fake = PatchAttempt(
|
|
finding_id=99, job_id="j-1", project_name="demo",
|
|
attempt_number=1, status="drafted",
|
|
)
|
|
fake.id = 1
|
|
_install_stub_patcher(server, attempt=fake)
|
|
|
|
r = tc.post(
|
|
"/jobs/j-1/patches",
|
|
headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"},
|
|
json={},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["attempt"]["finding_id"] == 99
|
|
|
|
|
|
def test_post_patches_no_actionable_returns_attempt_none(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
_install_stub_patcher(server, attempt=None)
|
|
|
|
r = tc.post(
|
|
"/jobs/j-1/patches",
|
|
headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"},
|
|
json={},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["ok"] is True
|
|
assert body["attempt"] is None
|
|
assert body.get("reason") == "no_actionable_finding"
|
|
|
|
|
|
def test_post_patches_503_when_patcher_disabled(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
server.patcher = None
|
|
|
|
r = tc.post(
|
|
"/jobs/j-1/patches",
|
|
headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"},
|
|
json={},
|
|
)
|
|
assert r.status_code == 503
|
|
|
|
|
|
def test_post_patches_cross_token_returns_404(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
_install_stub_patcher(server, attempt=None)
|
|
|
|
# bravo cannot see alpha's job → 404 (existence-leak guard).
|
|
r = tc.post(
|
|
"/jobs/j-1/patches",
|
|
headers={"Authorization": f"Bearer {ctx['bravo_bearer']}"},
|
|
json={},
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_post_patches_rejects_non_int_finding_id(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
_install_stub_patcher(server, attempt=None)
|
|
|
|
r = tc.post(
|
|
"/jobs/j-1/patches",
|
|
headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"},
|
|
json={"finding_id": "not-an-int"},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_get_patches_filtered_by_project_and_status(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
|
|
# Insert two attempts with different statuses, plus one for a different
|
|
# (synthetic) project so the filter actually has to exclude.
|
|
fid = server.db.insert_finding(
|
|
job_id="j-1", kind="lint", severity="warn", message="m",
|
|
fingerprint="f", file="x.py", line=1, code="X",
|
|
)
|
|
server.db.insert_patch_attempt(
|
|
finding_id=fid, job_id="j-1", project_name="demo",
|
|
attempt_number=1, status="pr_opened",
|
|
pr_url="http://gitea/X/Y/pulls/1",
|
|
)
|
|
server.db.insert_patch_attempt(
|
|
finding_id=fid, job_id="j-1", project_name="demo",
|
|
attempt_number=2, status="apply_failed",
|
|
)
|
|
|
|
r = tc.get(
|
|
"/patches?project=demo&status=pr_opened",
|
|
headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
rows = r.json()["patches"]
|
|
assert len(rows) == 1
|
|
assert rows[0]["status"] == "pr_opened"
|
|
|
|
|
|
def test_get_patches_owner_scoped(client):
|
|
"""Bravo's /patches list never includes Alpha's attempts."""
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
fid = server.db.insert_finding(
|
|
job_id="j-1", kind="lint", severity="warn", message="m",
|
|
fingerprint="f", file="x.py", line=1, code="X",
|
|
)
|
|
server.db.insert_patch_attempt(
|
|
finding_id=fid, job_id="j-1", project_name="demo",
|
|
attempt_number=1, status="pr_opened",
|
|
)
|
|
|
|
r = tc.get(
|
|
"/patches",
|
|
headers={"Authorization": f"Bearer {ctx['bravo_bearer']}"},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.json()["patches"] == []
|
|
|
|
|
|
def test_get_patches_detail_owner(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
fid = server.db.insert_finding(
|
|
job_id="j-1", kind="lint", severity="warn", message="m",
|
|
fingerprint="f", file="x.py", line=1, code="X",
|
|
)
|
|
pid = server.db.insert_patch_attempt(
|
|
finding_id=fid, job_id="j-1", project_name="demo",
|
|
attempt_number=1, status="pr_opened",
|
|
)
|
|
|
|
r = tc.get(
|
|
f"/patches/{pid}",
|
|
headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["patch"]["id"] == pid
|
|
|
|
|
|
def test_get_patches_detail_other_token_404(client):
|
|
tc, ctx = client
|
|
server = ctx["server"]
|
|
_register_demo_project(tc, ctx["alpha_bearer"])
|
|
_seed_job_row(server)
|
|
fid = server.db.insert_finding(
|
|
job_id="j-1", kind="lint", severity="warn", message="m",
|
|
fingerprint="f", file="x.py", line=1, code="X",
|
|
)
|
|
pid = server.db.insert_patch_attempt(
|
|
finding_id=fid, job_id="j-1", project_name="demo",
|
|
attempt_number=1, status="pr_opened",
|
|
)
|
|
|
|
r = tc.get(
|
|
f"/patches/{pid}",
|
|
headers={"Authorization": f"Bearer {ctx['bravo_bearer']}"},
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_get_patches_detail_missing(client):
|
|
tc, ctx = client
|
|
r = tc.get(
|
|
"/patches/999999",
|
|
headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"},
|
|
)
|
|
assert r.status_code == 404
|