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.errors import ValidationError 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, retained_run_report_paths, select_retained_run, ) 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_applies_external_extension_profile_schemas(self) -> None: with TemporaryDirectory() as temporary_directory: temp_root = Path(temporary_directory) extension_dir = temp_root / "schema-noop" _write_schema_extension(extension_dir) extensions = discover_extensions(ROOT, [extension_dir]) target_path = temp_root / "target.json" assessment_path = temp_root / "assessment.json" _write_schema_target(target_path, endpoints=[{ "id": "api", "url": "http://127.0.0.1:8080", "binding": "example", }]) _write_schema_assessment(assessment_path, runtime_policy={"offline": True}) target = validate_target_profile(target_path, extensions) assessment = validate_assessment_profile(assessment_path, extensions) plan = build_run_plan(ROOT, target_path, assessment_path, [extension_dir]) self.assertEqual(target["subject_type"], "schema-subject") self.assertEqual(assessment["runtime_policy"], {"offline": True}) self.assertEqual(plan["extension_snapshots"][0]["id"], "schema-noop") _write_schema_target(target_path, endpoints=[]) with self.assertRaisesRegex( ValidationError, "schema-noop:schema-target profile schema validation failed", ): validate_target_profile(target_path, extensions) def test_rejects_extension_profile_schema_paths_outside_extension_root(self) -> None: with TemporaryDirectory() as temporary_directory: temp_root = Path(temporary_directory) extension_dir = temp_root / "schema-noop" _write_schema_extension(extension_dir, target_schema_path="../outside.schema.json") target_path = temp_root / "target.json" _write_schema_target(target_path, endpoints=[{ "id": "api", "url": "http://127.0.0.1:8080", "binding": "example", }]) extensions = discover_extensions(ROOT, [extension_dir]) with self.assertRaisesRegex( ValidationError, "profile schema path escapes extension root", ): validate_target_profile(target_path, extensions) 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) latest = select_retained_run( runs_dir, target_profile_ref="sample-repository", assessment_profile_ref="sample-noop-assessment", ) self.assertIsNotNone(latest) assert latest is not None self.assertEqual(latest["run_id"], "run-new") self.assertEqual( retained_run_report_paths(latest)["report"], str(runs_dir / "run-new" / "reports" / "report.md"), ) 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", ) def _write_schema_extension( extension_dir: Path, target_schema_path: str = "schemas/schema-target.schema.json", ) -> None: extension_dir.mkdir(parents=True, exist_ok=True) schema_dir = extension_dir / "schemas" schema_dir.mkdir() (schema_dir / "schema-target.schema.json").write_text( json.dumps( { "type": "object", "required": ["subject_type", "endpoints"], "properties": { "subject_type": {"enum": ["schema-subject"]}, "endpoints": {"type": "array", "minItems": 1}, }, } ), encoding="utf-8", ) (schema_dir / "schema-assessment.schema.json").write_text( json.dumps( { "type": "object", "required": ["runtime_policy"], "properties": { "runtime_policy": { "type": "object", "required": ["offline"], "properties": {"offline": {"type": "boolean"}}, } }, } ), encoding="utf-8", ) (extension_dir / "extension.json").write_text( json.dumps( { "id": "schema-noop", "name": "Schema No-op", "version": "0.1.0", "extension_type": "repository_quality", "lifecycle_status": "incubating", "supported_frameworks": ["schema.readiness.v1"], "authorities": [], "profile_schemas": [ "target-profile", "assessment-profile", { "id": "schema-target", "profile_kind": "target", "path": target_schema_path, "subject_type": "schema-subject", }, { "id": "schema-assessment", "profile_kind": "assessment", "path": "schemas/schema-assessment.schema.json", }, ], "check_groups": [ { "id": "shape", "name": "Shape", "check_type": "repository_quality", "requirement_refs": ["schema.shape"], "runner_ref": None, } ], "preflight_runner": None, "runner_entrypoints": [], "normalizers": [], "mappings": [], "report_fragments": [], "dependencies": [], "restricted_assets": [], "certification_boundary": "Test fixture only.", } ), encoding="utf-8", ) def _write_schema_target(path: Path, endpoints: list[dict[str, str]]) -> None: path.write_text( json.dumps( { "id": "schema-target", "subject_type": "schema-subject", "subject_name": "Schema Target", "environment": "test", "scope": ["schema"], "endpoints": endpoints, "artifacts": [], "credentials_ref": None, "declared_capabilities": [], "known_gaps": [], } ), encoding="utf-8", ) def _write_schema_assessment(path: Path, runtime_policy: dict[str, object]) -> None: path.write_text( json.dumps( { "id": "schema-assessment", "framework_refs": ["schema.readiness.v1"], "extension_refs": ["schema-noop"], "target_profile_ref": "schema-target", "selected_check_groups": {"schema-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": runtime_policy, } ), encoding="utf-8", ) if __name__ == "__main__": unittest.main()