Files
guide-board/tests/test_core.py

1094 lines
44 KiB
Python

from __future__ import annotations
import http.client
import json
import shutil
import time
import unittest
from tempfile import TemporaryDirectory
from pathlib import Path
from urllib.parse import quote
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"],
)
assert_valid(plan["source_lock"], "source-lock")
self.assertEqual(plan["source_lock"]["schema_version"], "guide-board.source-lock.v1")
self.assertEqual(plan["source_lock"]["framework_refs"], ["guide-board.sample-readiness.v0"])
self.assertEqual(plan["source_lock"]["extension_refs"], ["sample-noop"])
self.assertEqual(
plan["source_lock"]["profiles"]["target"]["snapshot_ref"],
"target-profile.snapshot.json",
)
self.assertTrue(plan["source_lock"]["profiles"]["target"]["checksum"].startswith("sha256:"))
self.assertEqual(plan["source_lock"]["mapping_sets"][0]["id"], "sample-readiness-map")
self.assertTrue(plan["source_lock"]["mapping_sets"][0]["checksum"].startswith("sha256:"))
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_sdk_fixture_from_external_extension_repo(self) -> None:
with TemporaryDirectory() as temporary_directory:
temp_root = Path(temporary_directory)
extension_dir = temp_root / "sdk-fixture"
shutil.copytree(ROOT / "extensions" / "sdk-fixture", extension_dir)
result = run_assessment(
temp_root,
extension_dir / "profiles" / "targets" / "sdk-fixture-target.json",
extension_dir / "profiles" / "assessments" / "sdk-fixture-assessment.json",
temp_root / "runs" / "sdk-fixture",
[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"]
mappings = json.loads(
(run_dir / "normalized" / "mappings.json").read_text(encoding="utf-8")
)["mappings"]
assessment_package = json.loads(
(run_dir / "reports" / "assessment-package.json").read_text(encoding="utf-8")
)
report = (run_dir / "reports" / "report.md").read_text(encoding="utf-8")
export_manifest = json.loads(
(run_dir / "exports" / "export-manifest.json").read_text(encoding="utf-8")
)
self.assertEqual(result["status"], "completed")
self.assertEqual(plan["extension_snapshots"][0]["source"], "external")
self.assertEqual(plan["target_profile_snapshot"]["subject_type"], "sdk-fixture-target")
self.assertEqual([item["result"] for item in evidence], ["skipped", "pass"])
check_evidence = evidence[1]
self.assertEqual(
check_evidence["facts"]["normalizer_refs"],
["native-probe-normalizer"],
)
self.assertEqual(check_evidence["facts"]["native_score"], 98)
self.assertEqual(
check_evidence["requirement_refs"],
["guide-board.sdk-fixture.v1.native-output"],
)
self.assertEqual(
check_evidence["artifact_refs"],
["artifacts/sdk-fixture/native-result.json"],
)
self.assertEqual(
check_evidence["facts"]["source_metadata"]["runner"]["metadata"]["harness_version"],
"1.0.0",
)
self.assertEqual(
check_evidence["facts"]["source_metadata"]["reported"]["native_result_id"],
"sdk-fixture-native-result",
)
self.assertEqual(mappings[0]["target_id"], "normalizer-plugin")
self.assertEqual(assessment_package["summary"], {"pass": 1, "skipped": 1})
self.assertEqual(
assessment_package["report_fragments"][0]["markdown"],
(ROOT / "tests" / "golden" / "sdk-fixture-report-fragment.md")
.read_text(encoding="utf-8")
.rstrip(),
)
self.assertIn("### SDK Fixture Summary", report)
assert_valid(export_manifest, "export-manifest")
export_shape = load_json(ROOT / "tests" / "golden" / "export-manifest-shape.json")
self.assertEqual(sorted(export_manifest), export_shape["top_level_keys"])
self.assertEqual(export_manifest["counts"]["report_fragments"], 1)
self.assertEqual(
export_manifest["report_fragments"][0]["structured"]["evidence_count"],
2,
)
self.assertEqual(
assessment_package["source_lock"]["metadata_hooks"]["runner_entrypoints"][0][
"metadata"
]["harness_id"],
"sdk-fixture-native-probe",
)
submission_package = json.loads(
(run_dir / "reports" / "submission-package.json").read_text(encoding="utf-8")
)
assert_valid(submission_package, "submission-package")
self.assertEqual(submission_package["source_lock"]["id"], "source-lock:sdk-fixture-assessment:sdk-fixture-target")
self.assertEqual(
submission_package["reported_metadata"][1]["metadata"]["reported"][
"native_result_id"
],
"sdk-fixture-native-result",
)
self.assertEqual(
submission_package["artifact_manifest"][0]["checksum"],
assessment_package["artifact_manifest"][0]["checksum"],
)
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())
self.assertTrue((run_dir / "reports" / "fragments.json").exists())
self.assertTrue((run_dir / "reports" / "submission-package.json").exists())
self.assertTrue((run_dir / "exports" / "export-manifest.json").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(
result["submission_package"],
str(run_dir / "reports" / "submission-package.json"),
)
self.assertEqual(
result["export_manifest"],
str(run_dir / "exports" / "export-manifest.json"),
)
self.assertEqual(retention["summary"]["status"], "completed")
self.assertEqual(retention["summary"]["artifact_count"], 0)
self.assertIn("reports/submission-package.json", retention["report_refs"])
self.assertIn("exports/export-manifest.json", retention["export_refs"])
self.assertEqual(
retention["artifact_retention"]["policy"],
{"raw_artifact_days": 0, "summary_days": 365},
)
submission = json.loads(
(run_dir / "reports" / "submission-package.json").read_text(encoding="utf-8")
)
assert_valid(submission, "submission-package")
self.assertEqual(submission["package_identity"]["target_profile_ref"], "sample-repository")
self.assertEqual(
[entry["path"] for entry in submission["reports"]],
["reports/assessment-package.json", "reports/report.md"],
)
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_applies_challenges_and_exclusions_without_hiding_gate_failures(self) -> None:
with TemporaryDirectory() as temporary_directory:
temp_root = Path(temporary_directory)
extension_dir = temp_root / "review-noop"
_write_review_extension(extension_dir)
target_path = temp_root / "review-target.json"
assessment_path = temp_root / "review-assessment.json"
challenge_path = temp_root / "review-challenges.json"
exclusion_path = temp_root / "review-exclusions.json"
_write_review_target(target_path)
_write_review_assessment(assessment_path)
_write_review_challenges(challenge_path)
_write_review_exclusions(exclusion_path)
result = run_assessment(
ROOT,
target_path,
assessment_path,
temp_root / "runs" / "review",
[extension_dir],
)
run_dir = Path(result["run_dir"])
evidence = json.loads(
(run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8")
)["evidence"]
assessment_package = json.loads(
(run_dir / "reports" / "assessment-package.json").read_text(encoding="utf-8")
)
retention = json.loads(
(run_dir / "retention-summary.json").read_text(encoding="utf-8")
)
report = (run_dir / "reports" / "report.md").read_text(encoding="utf-8")
self.assertEqual(result["status"], "blocked")
finding = assessment_package["findings"][0]
self.assertEqual(finding["challenge_ref"], "challenge-review-blocked")
self.assertEqual(finding["exclusion_ref"], "exclusion-review-blocked")
self.assertEqual(finding["review_status"], "authority_excluded")
self.assertFalse(finding["expected"])
self.assertEqual(assessment_package["policy_summary"]["unexpected_findings"], 1)
self.assertEqual(assessment_package["policy_summary"]["challenged_findings"], 1)
self.assertEqual(assessment_package["policy_summary"]["authority_exclusions"], 1)
self.assertEqual(assessment_package["policy_summary"]["unresolved_defects"], 0)
self.assertEqual(
evidence[1]["review"]["challenge_refs"],
["challenge-review-blocked"],
)
self.assertEqual(
evidence[1]["review"]["exclusion_refs"],
["exclusion-review-blocked"],
)
self.assertEqual(assessment_package["challenges"][0]["owner"], "qa")
self.assertEqual(assessment_package["exclusions"][0]["authority_ref"], "review-authority")
self.assertEqual(retention["summary"]["challenged_findings"], 1)
self.assertEqual(retention["summary"]["authority_exclusions"], 1)
self.assertEqual(retention["summary"]["unresolved_review_items"], 1)
self.assertIn("- authority_excluded: 1", report)
gate = evaluate_trend_gates(build_trend_summary(temp_root / "runs"))
self.assertEqual(gate["status"], "failed")
checks = {check["id"]: check for check in gate["groups"][0]["checks"]}
self.assertEqual(checks["unexpected-findings"]["observed"], 1)
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"],
)
self.assertEqual(
reports["submission_package"]["json"]["run_id"],
status["result"]["run_id"],
)
self.assertEqual(
reports["export_manifest"]["json"]["run_id"],
status["result"]["run_id"],
)
finally:
service.stop()
def test_service_exposes_retained_runs_after_restart(self) -> None:
with TemporaryDirectory() as temporary_directory:
runs_dir = Path(temporary_directory) / "runs"
result = run_assessment(
ROOT,
ROOT / "profiles" / "targets" / "sample-repository.json",
ROOT / "profiles" / "assessments" / "sample-noop.json",
runs_dir / "sample",
)
_write_unsafe_artifact_run(runs_dir / "unsafe-run")
service = start_service(ROOT, host="127.0.0.1", port=0)
try:
query = f"runs_dir={quote(str(runs_dir), safe='')}"
listing = _request_json(service, "GET", f"/retained-runs?{query}")
self.assertEqual(listing["runs_dir"], str(runs_dir))
self.assertIn(result["run_id"], [run["run_id"] for run in listing["runs"]])
latest = _request_json(
service,
"GET",
f"/retained-runs/latest?{query}&target=sample-repository&assessment=sample-noop-assessment",
)
self.assertEqual(latest["run"]["run_id"], result["run_id"])
self.assertIn("submission_package", latest["run"]["paths"])
reports = _request_json(
service,
"GET",
f"/retained-runs/{result['run_id']}/reports?{query}",
)
self.assertEqual(
reports["run"]["paths"]["assessment_package"],
str(runs_dir / "sample" / "reports" / "assessment-package.json"),
)
artifacts = _request_json(
service,
"GET",
f"/retained-runs/{result['run_id']}/artifact-manifest?{query}",
)
self.assertEqual(artifacts["artifact_manifest"], [])
self.assertEqual(artifacts["compatibility"], "current")
unsafe = _request_json(
service,
"GET",
f"/retained-runs/unsafe-run/artifact-manifest?{query}",
expected_status=400,
)
self.assertIn("escapes run directory", unsafe["error"]["message"])
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"]["status_change"],
{"from": "blocked", "to": "completed"},
)
self.assertEqual(group["trend"]["unexpected_findings_delta"], -1)
self.assertEqual(group["trend"]["mapping_target_count_delta"], 0)
self.assertIn("Trend improved", group["trend"]["summary_text"])
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 _write_unsafe_artifact_run(run_dir: Path) -> None:
_write_retention_summary(
run_dir,
"unsafe-run",
"2026-05-07T12:00:00+00:00",
"completed",
{"pass": 1},
0,
1,
)
reports_dir = run_dir / "reports"
reports_dir.mkdir(parents=True, exist_ok=True)
(reports_dir / "assessment-package.json").write_text(
json.dumps(
{
"artifact_manifest": [
{
"id": "artifact:unsafe",
"path": "../outside.txt",
"checksum": "sha256:unsafe",
}
]
}
),
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",
)
def _write_review_extension(extension_dir: Path) -> None:
extension_dir.mkdir(parents=True, exist_ok=True)
(extension_dir / "extension.json").write_text(
json.dumps(
{
"id": "review-noop",
"name": "Review No-op",
"version": "0.1.0",
"extension_type": "repository_quality",
"lifecycle_status": "incubating",
"supported_frameworks": ["review.framework.v1"],
"authorities": ["review-authority"],
"profile_schemas": ["target-profile", "assessment-profile"],
"check_groups": [
{
"id": "review",
"name": "Review",
"check_type": "repository_quality",
"requirement_refs": ["review.requirement"],
"runner_ref": "external-review",
}
],
"preflight_runner": None,
"runner_entrypoints": [
{
"id": "external-review",
"kind": "external",
"module_path": None,
"callable": None,
"command": None,
"metadata": {"test_suite_id": "review-suite"},
"description": "External runner used to produce reviewable blocked evidence.",
}
],
"normalizers": [],
"mappings": [],
"report_fragments": [],
"dependencies": [],
"restricted_assets": [],
"certification_boundary": "Review fixture only.",
}
),
encoding="utf-8",
)
def _write_review_target(path: Path) -> None:
path.write_text(
json.dumps(
{
"id": "review-target",
"subject_type": "repository",
"subject_name": "Review Target",
"environment": "test",
"scope": ["review"],
"endpoints": [],
"artifacts": [],
"credentials_ref": None,
"declared_capabilities": [],
"known_gaps": [],
}
),
encoding="utf-8",
)
def _write_review_assessment(path: Path) -> None:
path.write_text(
json.dumps(
{
"id": "review-assessment",
"framework_refs": ["review.framework.v1"],
"extension_refs": ["review-noop"],
"target_profile_ref": "review-target",
"selected_check_groups": {"review-noop": ["review"]},
"expectations_ref": None,
"waivers_ref": None,
"challenges_ref": "review-challenges.json",
"exclusions_ref": "review-exclusions.json",
"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",
)
def _write_review_challenges(path: Path) -> None:
path.write_text(
json.dumps(
{
"id": "review-challenges",
"target_profile_ref": "review-target",
"challenges": [
{
"id": "challenge-review-blocked",
"requirement_refs": ["review.requirement"],
"check_refs": ["check-group:review-noop:review"],
"evidence_refs": [],
"result_refs": ["blocked"],
"classification_refs": ["runner_not_implemented"],
"authority_source_refs": ["review-authority:rule-1"],
"owner": "qa",
"review_status": "open",
"rationale": "The external suite is not wired in this fixture.",
"created_at": "2026-05-16",
"review_due_at": "2026-06-16",
"expires_at": None,
"native_challenge_id": "native-challenge-1",
"metadata": {"kind": "fixture"},
}
],
}
),
encoding="utf-8",
)
def _write_review_exclusions(path: Path) -> None:
path.write_text(
json.dumps(
{
"id": "review-exclusions",
"target_profile_ref": "review-target",
"exclusions": [
{
"id": "exclusion-review-blocked",
"authority_ref": "review-authority",
"requirement_refs": ["review.requirement"],
"check_refs": ["check-group:review-noop:review"],
"evidence_refs": [],
"result_refs": ["blocked"],
"classification_refs": ["runner_not_implemented"],
"authority_source_refs": ["review-authority:rule-1"],
"owner": "qa",
"approved_by": "authority-reviewer",
"review_status": "approved",
"rationale": "Fixture demonstrates authority exclusion annotation.",
"created_at": "2026-05-16",
"review_due_at": "2026-06-16",
"expires_at": None,
"native_exclusion_id": "native-exclusion-1",
"metadata": {"kind": "fixture"},
}
],
}
),
encoding="utf-8",
)
if __name__ == "__main__":
unittest.main()