generated from coulomb/repo-seed
447 lines
17 KiB
Python
447 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import http.client
|
|
import json
|
|
import time
|
|
import unittest
|
|
from tempfile import TemporaryDirectory
|
|
from pathlib import Path
|
|
|
|
from guide_board.discovery import discover_extensions
|
|
from guide_board.execution import run_assessment
|
|
from guide_board.gates import evaluate_trend_gates
|
|
from guide_board.io import load_json
|
|
from guide_board.planning import (
|
|
build_run_plan,
|
|
validate_assessment_profile,
|
|
validate_target_profile,
|
|
)
|
|
from guide_board.retention import build_trend_summary, list_retained_runs
|
|
from guide_board.schema import assert_valid
|
|
from guide_board.service import ServiceHandle, start_service
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class CoreArchitectureTests(unittest.TestCase):
|
|
def test_discovers_incubating_extensions(self) -> None:
|
|
extensions = {extension.id for extension in discover_extensions(ROOT)}
|
|
|
|
self.assertIn("sample-noop", extensions)
|
|
|
|
def test_validates_sample_profiles(self) -> None:
|
|
target = validate_target_profile(ROOT / "profiles" / "targets" / "sample-repository.json")
|
|
assessment = validate_assessment_profile(
|
|
ROOT / "profiles" / "assessments" / "sample-noop.json"
|
|
)
|
|
|
|
self.assertEqual(target["id"], "sample-repository")
|
|
self.assertEqual(assessment["target_profile_ref"], "sample-repository")
|
|
|
|
def test_validates_evidence_request_template(self) -> None:
|
|
request_set = load_json(ROOT / "extensions" / "_template" / "evidence-request-set.json")
|
|
|
|
assert_valid(request_set, "evidence-request-set")
|
|
self.assertEqual(
|
|
request_set["source_boundary"]["certification_boundary"],
|
|
"This evidence request set supports assessment preparation only.",
|
|
)
|
|
|
|
def test_builds_sample_run_plan(self) -> None:
|
|
plan = build_run_plan(
|
|
ROOT,
|
|
ROOT / "profiles" / "targets" / "sample-repository.json",
|
|
ROOT / "profiles" / "assessments" / "sample-noop.json",
|
|
)
|
|
|
|
self.assertEqual(plan["target_profile_snapshot"]["id"], "sample-repository")
|
|
self.assertEqual(plan["extension_snapshots"][0]["id"], "sample-noop")
|
|
self.assertEqual(
|
|
[step["id"] for step in plan["ordered_steps"]],
|
|
[
|
|
"preflight:sample-noop",
|
|
"check-group:sample-noop:profile-shape",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
plan["ordered_steps"][1]["requirement_refs"],
|
|
["guide-board.sample-readiness.v0.profile-shape"],
|
|
)
|
|
|
|
def test_runs_external_extension_from_separate_repo(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
extension_dir = temp_root / "external-noop"
|
|
_write_external_extension(extension_dir)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
target_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"id": "external-target",
|
|
"subject_type": "repository",
|
|
"subject_name": "External Target",
|
|
"environment": "test",
|
|
"scope": ["external"],
|
|
"endpoints": [],
|
|
"artifacts": [],
|
|
"credentials_ref": None,
|
|
"declared_capabilities": [],
|
|
"known_gaps": [],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
assessment_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"id": "external-assessment",
|
|
"framework_refs": ["external.readiness.v1"],
|
|
"extension_refs": ["external-noop"],
|
|
"target_profile_ref": "external-target",
|
|
"selected_check_groups": {"external-noop": ["shape"]},
|
|
"expectations_ref": None,
|
|
"waivers_ref": None,
|
|
"output_policy": {
|
|
"report_formats": ["json", "markdown"],
|
|
"artifact_retention": "summary-only",
|
|
},
|
|
"retention_policy": {
|
|
"summary_days": 365,
|
|
"raw_artifact_days": 0,
|
|
},
|
|
"runtime_policy": {
|
|
"offline": True,
|
|
"timeout_seconds": 2,
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_assessment(
|
|
ROOT,
|
|
target_path,
|
|
assessment_path,
|
|
temp_root / "run",
|
|
[extension_dir],
|
|
)
|
|
run_dir = Path(result["run_dir"])
|
|
plan = json.loads((run_dir / "plan.json").read_text(encoding="utf-8"))
|
|
evidence = json.loads(
|
|
(run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8")
|
|
)["evidence"]
|
|
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertEqual(plan["extension_snapshots"][0]["source"], "external")
|
|
self.assertEqual(plan["extension_snapshots"][0]["path"], str(extension_dir))
|
|
self.assertEqual([item["result"] for item in evidence], ["skipped", "manual"])
|
|
|
|
def test_runs_sample_noop_assessment(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
result = run_assessment(
|
|
ROOT,
|
|
ROOT / "profiles" / "targets" / "sample-repository.json",
|
|
ROOT / "profiles" / "assessments" / "sample-noop.json",
|
|
Path(temporary_directory) / "sample-run",
|
|
)
|
|
|
|
run_dir = Path(result["run_dir"])
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertTrue((run_dir / "run.json").exists())
|
|
self.assertTrue((run_dir / "retention-summary.json").exists())
|
|
self.assertTrue((run_dir / "normalized" / "evidence.json").exists())
|
|
self.assertTrue((run_dir / "reports" / "assessment-package.json").exists())
|
|
self.assertTrue((run_dir / "reports" / "report.md").exists())
|
|
retention = json.loads(
|
|
(run_dir / "retention-summary.json").read_text(encoding="utf-8")
|
|
)
|
|
self.assertEqual(
|
|
result["retention_summary"],
|
|
str(run_dir / "retention-summary.json"),
|
|
)
|
|
self.assertEqual(retention["summary"]["status"], "completed")
|
|
self.assertEqual(retention["summary"]["artifact_count"], 0)
|
|
self.assertEqual(
|
|
retention["artifact_retention"]["policy"],
|
|
{"raw_artifact_days": 0, "summary_days": 365},
|
|
)
|
|
self.assertEqual(
|
|
[run["run_id"] for run in list_retained_runs(Path(temporary_directory))],
|
|
[result["run_id"]],
|
|
)
|
|
mappings = json.loads(
|
|
(run_dir / "normalized" / "mappings.json").read_text(encoding="utf-8")
|
|
)["mappings"]
|
|
self.assertEqual(len(mappings), 1)
|
|
self.assertEqual(mappings[0]["target_id"], "profile-readiness")
|
|
|
|
def test_serves_local_api_run_lifecycle(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
service = start_service(ROOT, host="127.0.0.1", port=0)
|
|
try:
|
|
health = _request_json(service, "GET", "/health")
|
|
self.assertEqual(health["status"], "ok")
|
|
|
|
extensions = _request_json(service, "GET", "/extensions")
|
|
self.assertIn(
|
|
"sample-noop",
|
|
[extension["id"] for extension in extensions["extensions"]],
|
|
)
|
|
|
|
target_validation = _request_json(
|
|
service,
|
|
"POST",
|
|
"/profiles/validate",
|
|
{
|
|
"kind": "target",
|
|
"path": "profiles/targets/sample-repository.json",
|
|
},
|
|
)
|
|
self.assertEqual(target_validation["profile_id"], "sample-repository")
|
|
|
|
plan = _request_json(
|
|
service,
|
|
"POST",
|
|
"/assessments/plan",
|
|
{
|
|
"target": "profiles/targets/sample-repository.json",
|
|
"assessment": "profiles/assessments/sample-noop.json",
|
|
},
|
|
)
|
|
self.assertEqual(plan["target_profile_snapshot"]["id"], "sample-repository")
|
|
|
|
job = _request_json(
|
|
service,
|
|
"POST",
|
|
"/runs",
|
|
{
|
|
"target": "profiles/targets/sample-repository.json",
|
|
"assessment": "profiles/assessments/sample-noop.json",
|
|
"output_dir": str(Path(temporary_directory) / "api-run"),
|
|
},
|
|
expected_status=202,
|
|
)
|
|
status = _wait_for_job(service, job["job_id"])
|
|
|
|
self.assertEqual(status["status"], "succeeded")
|
|
self.assertEqual(status["result"]["status"], "completed")
|
|
|
|
reports = _request_json(
|
|
service,
|
|
"GET",
|
|
f"/runs/{job['job_id']}/reports",
|
|
)
|
|
self.assertIn("Guide Board Assessment Report", reports["report"]["markdown"])
|
|
self.assertEqual(
|
|
reports["assessment_package"]["json"]["run_id"],
|
|
status["result"]["run_id"],
|
|
)
|
|
finally:
|
|
service.stop()
|
|
|
|
def test_builds_retained_run_trends(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
runs_dir = Path(temporary_directory)
|
|
_write_retention_summary(
|
|
runs_dir / "run-old",
|
|
"run-old",
|
|
"2026-05-07T10:00:00+00:00",
|
|
"blocked",
|
|
{"blocked": 1},
|
|
1,
|
|
1,
|
|
)
|
|
_write_retention_summary(
|
|
runs_dir / "run-new",
|
|
"run-new",
|
|
"2026-05-07T11:00:00+00:00",
|
|
"completed",
|
|
{"manual": 1, "skipped": 1},
|
|
0,
|
|
2,
|
|
)
|
|
|
|
trend = build_trend_summary(runs_dir)
|
|
assert_valid(trend, "trend-summary")
|
|
|
|
self.assertEqual(trend["run_count"], 2)
|
|
self.assertEqual(len(trend["groups"]), 1)
|
|
group = trend["groups"][0]
|
|
self.assertEqual(group["latest_run"]["run_id"], "run-new")
|
|
self.assertEqual(group["previous_run"]["run_id"], "run-old")
|
|
self.assertEqual(group["trend"]["direction"], "improved")
|
|
self.assertTrue(group["trend"]["status_changed"])
|
|
self.assertEqual(group["trend"]["unexpected_findings_delta"], -1)
|
|
self.assertEqual(
|
|
group["trend"]["evidence_result_deltas"],
|
|
{"blocked": -1, "manual": 1, "skipped": 1},
|
|
)
|
|
|
|
gate = evaluate_trend_gates(
|
|
trend,
|
|
target_profile_ref="sample-repository",
|
|
assessment_profile_ref="sample-noop-assessment",
|
|
)
|
|
assert_valid(gate, "gate-summary")
|
|
self.assertEqual(gate["status"], "passed")
|
|
self.assertEqual(gate["passed_groups"], 1)
|
|
|
|
missing_gate = evaluate_trend_gates(
|
|
trend,
|
|
target_profile_ref="missing-target",
|
|
)
|
|
self.assertEqual(missing_gate["status"], "failed")
|
|
self.assertEqual(missing_gate["groups"][0]["checks"][0]["id"], "history-present")
|
|
|
|
def test_fails_gate_for_regressed_run_history(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
runs_dir = Path(temporary_directory)
|
|
_write_retention_summary(
|
|
runs_dir / "run-old",
|
|
"run-old",
|
|
"2026-05-07T10:00:00+00:00",
|
|
"completed",
|
|
{"manual": 1},
|
|
0,
|
|
1,
|
|
)
|
|
_write_retention_summary(
|
|
runs_dir / "run-new",
|
|
"run-new",
|
|
"2026-05-07T11:00:00+00:00",
|
|
"blocked",
|
|
{"blocked": 1},
|
|
2,
|
|
1,
|
|
)
|
|
|
|
gate = evaluate_trend_gates(build_trend_summary(runs_dir))
|
|
assert_valid(gate, "gate-summary")
|
|
|
|
self.assertEqual(gate["status"], "failed")
|
|
checks = {check["id"]: check for check in gate["groups"][0]["checks"]}
|
|
self.assertEqual(checks["latest-status"]["status"], "failed")
|
|
self.assertEqual(checks["unexpected-findings"]["status"], "failed")
|
|
self.assertEqual(checks["trend-regression"]["status"], "failed")
|
|
|
|
def _write_retention_summary(
|
|
run_dir: Path,
|
|
run_id: str,
|
|
created_at: str,
|
|
status: str,
|
|
evidence_results: dict[str, int],
|
|
unexpected_findings: int,
|
|
artifact_count: int,
|
|
) -> None:
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
(run_dir / "retention-summary.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"id": f"retention-summary:{run_id}",
|
|
"run_id": run_id,
|
|
"target_profile_ref": "sample-repository",
|
|
"assessment_profile_ref": "sample-noop-assessment",
|
|
"created_at": created_at,
|
|
"summary": {
|
|
"status": status,
|
|
"evidence_results": evidence_results,
|
|
"finding_count": unexpected_findings,
|
|
"unexpected_findings": unexpected_findings,
|
|
"expected_findings": 0,
|
|
"waived_findings": 0,
|
|
"mapping_target_count": 1,
|
|
"artifact_count": artifact_count,
|
|
},
|
|
"report_refs": [
|
|
"reports/assessment-package.json",
|
|
"reports/report.md",
|
|
],
|
|
"artifact_retention": {
|
|
"policy": {"raw_artifact_days": 0, "summary_days": 365},
|
|
"output_artifact_retention": "summary-only",
|
|
"retention_class_counts": {"raw": artifact_count},
|
|
"raw_artifact_count": artifact_count,
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _request_json(
|
|
service: ServiceHandle,
|
|
method: str,
|
|
path: str,
|
|
payload: dict[str, object] | None = None,
|
|
expected_status: int = 200,
|
|
) -> dict[str, object]:
|
|
connection = http.client.HTTPConnection(service.host, service.port, timeout=5)
|
|
body = None
|
|
headers = {}
|
|
if payload is not None:
|
|
body = json.dumps(payload).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
try:
|
|
connection.request(method, path, body=body, headers=headers)
|
|
response = connection.getresponse()
|
|
data = response.read().decode("utf-8")
|
|
finally:
|
|
connection.close()
|
|
if response.status != expected_status:
|
|
raise AssertionError(f"expected HTTP {expected_status}, got {response.status}: {data}")
|
|
value = json.loads(data)
|
|
if not isinstance(value, dict):
|
|
raise AssertionError(f"expected JSON object response, got {type(value).__name__}")
|
|
return value
|
|
|
|
|
|
def _wait_for_job(service: ServiceHandle, job_id: str) -> dict[str, object]:
|
|
for _ in range(50):
|
|
status = _request_json(service, "GET", f"/runs/{job_id}")
|
|
if status["status"] in {"succeeded", "failed"}:
|
|
return status
|
|
time.sleep(0.05)
|
|
raise AssertionError(f"job did not finish: {job_id}")
|
|
|
|
|
|
def _write_external_extension(extension_dir: Path) -> None:
|
|
extension_dir.mkdir(parents=True, exist_ok=True)
|
|
(extension_dir / "extension.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"id": "external-noop",
|
|
"name": "External No-op",
|
|
"version": "0.1.0",
|
|
"extension_type": "repository_quality",
|
|
"lifecycle_status": "incubating",
|
|
"supported_frameworks": ["external.readiness.v1"],
|
|
"authorities": [],
|
|
"profile_schemas": ["target-profile", "assessment-profile"],
|
|
"check_groups": [
|
|
{
|
|
"id": "shape",
|
|
"name": "Shape",
|
|
"check_type": "repository_quality",
|
|
"requirement_refs": ["external.shape"],
|
|
"runner_ref": None,
|
|
}
|
|
],
|
|
"preflight_runner": None,
|
|
"runner_entrypoints": [],
|
|
"normalizers": [],
|
|
"mappings": [],
|
|
"report_fragments": [],
|
|
"dependencies": [],
|
|
"restricted_assets": [],
|
|
"certification_boundary": "Test fixture only.",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|