"""Integration test — runner runs a recipe whose output is parser-shaped, and findings end up in the DB + visible via GET /jobs/{id}/findings. We use python:lint with a stub command that emits a known-good ruff JSON array, then verify: - findings_count on the job row reflects the parsed entries - GET /jobs/{id}/findings returns rows with the right kind/code/file/line - fingerprint is populated and stable across rows for the same locator """ from __future__ import annotations import json import time import shutil import subprocess from pathlib import Path import pytest from tests.conftest import sample_project_payload def _make_local_git_repo(root: Path) -> str: if shutil.which("git") is None: pytest.skip("git binary not present in test environment") repo = root / "findings-fixture-repo" repo.mkdir() subprocess.run(["git", "init", "-q", "-b", "main"], cwd=repo, check=True) subprocess.run(["git", "config", "user.email", "test@example"], cwd=repo, check=True) subprocess.run(["git", "config", "user.name", "test"], cwd=repo, check=True) subprocess.run(["git", "config", "commit.gpgsign", "false"], cwd=repo, check=True) (repo / "README.md").write_text("hello\n") subprocess.run(["git", "add", "README.md"], cwd=repo, check=True) subprocess.run(["git", "commit", "-q", "-m", "initial"], cwd=repo, check=True) return str(repo) def _wait_terminal(tc, bearer: str, job_id: str, timeout: float = 30.0) -> dict: deadline = time.monotonic() + timeout while time.monotonic() < deadline: r = tc.get( f"/jobs/{job_id}", headers={"Authorization": f"Bearer {bearer}"}, ) body = r.json() if body["job"]["status"] in ("succeeded", "failed", "timed_out", "cancelled"): return body time.sleep(0.1) raise AssertionError("job never finished") def test_python_lint_findings_persisted_via_runner(client, tmp_path): """python:lint with a ruff-shaped JSON stub → 2 findings persisted.""" tc, ctx = client git_url = _make_local_git_repo(tmp_path) # Ruff-shaped stub. echo + exit 1 (lint findings → non-zero exit) so # the recipe terminates the way ruff really would. ruff_stub = json.dumps( [ { "code": "F401", "message": "'os' imported but unused", "filename": "src/app.py", "location": {"row": 3, "column": 1}, }, { "code": "E501", "message": "Line too long", "filename": "src/app.py", "location": {"row": 42, "column": 89}, }, ] ) # Single-quote the JSON so the shell doesn't interpret double-quotes. lint_cmd = f"echo '{ruff_stub}'; exit 1" payload = sample_project_payload(name="ct-findings-py") payload["git_url"] = git_url payload["subprojects"][0]["language"] = "python" payload["subprojects"][0]["lint"] = lint_cmd payload["subprojects"][0]["timeout_secs"] = 20 r = tc.post( "/projects", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, json=payload, ) assert r.status_code == 200, r.text r2 = tc.post( "/projects/ct-findings-py/jobs", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, json={"recipe": "lint"}, ) assert r2.status_code == 200, r2.text job_id = r2.json()["job_id"] final = _wait_terminal(tc, ctx["alpha_bearer"], job_id) # Recipe exited 1 so the job is "failed" — but parsing still happens. assert final["job"]["status"] == "failed" assert final["job"]["exit_code"] == 1 assert final["job"]["findings_count"] == 2 r3 = tc.get( f"/jobs/{job_id}/findings", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, ) assert r3.status_code == 200, r3.text findings = r3.json()["findings"] assert len(findings) == 2 by_code = {f["code"]: f for f in findings} assert "F401" in by_code and "E501" in by_code f401 = by_code["F401"] assert f401["kind"] == "lint" assert f401["severity"] == "warn" assert f401["file"] == "src/app.py" assert f401["line"] == 3 assert f401["fingerprint"] assert len(f401["fingerprint"]) == 16 def test_unknown_lang_recipe_falls_back_to_generic(client, tmp_path): """A recipe with no parser registered emits exactly one recipe_fail finding when it exits non-zero, and zero findings when it exits 0.""" tc, ctx = client git_url = _make_local_git_repo(tmp_path) # Use `ruby` which has no parser; our PythonParser etc. all decline. payload = sample_project_payload(name="ct-findings-generic") payload["git_url"] = git_url payload["subprojects"][0]["language"] = "ruby" payload["subprojects"][0]["audit"] = "echo audit-output; exit 5" payload["subprojects"][0]["timeout_secs"] = 20 r = tc.post( "/projects", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, json=payload, ) assert r.status_code == 200 r2 = tc.post( "/projects/ct-findings-generic/jobs", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, json={"recipe": "audit"}, ) job_id = r2.json()["job_id"] final = _wait_terminal(tc, ctx["alpha_bearer"], job_id) assert final["job"]["status"] == "failed" assert final["job"]["findings_count"] == 1 r3 = tc.get( f"/jobs/{job_id}/findings", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, ) rows = r3.json()["findings"] assert len(rows) == 1 assert rows[0]["kind"] == "recipe_fail" assert rows[0]["code"] == "exit_5" def test_clean_recipe_produces_zero_findings(client, tmp_path): """Successful run with no parseable signal → no findings rows, count=0.""" tc, ctx = client git_url = _make_local_git_repo(tmp_path) payload = sample_project_payload(name="ct-findings-clean") payload["git_url"] = git_url payload["subprojects"][0]["language"] = "python" # Empty ruff JSON array → 0 findings, exit 0. payload["subprojects"][0]["lint"] = "echo '[]'; exit 0" payload["subprojects"][0]["timeout_secs"] = 20 r = tc.post( "/projects", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, json=payload, ) assert r.status_code == 200 r2 = tc.post( "/projects/ct-findings-clean/jobs", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, json={"recipe": "lint"}, ) job_id = r2.json()["job_id"] final = _wait_terminal(tc, ctx["alpha_bearer"], job_id) assert final["job"]["status"] == "succeeded" assert final["job"]["findings_count"] == 0 r3 = tc.get( f"/jobs/{job_id}/findings", headers={"Authorization": f"Bearer {ctx['alpha_bearer']}"}, ) assert r3.json()["findings"] == []