Files
open-cmis-tck/tests/test_core.py
2026-05-07 17:00:10 +02:00

605 lines
25 KiB
Python

from __future__ import annotations
import unittest
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
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.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
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)
self.assertIn("open-cmis-tck", 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_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_builds_cmis_baseline_plan(self) -> None:
plan = build_run_plan(
ROOT,
ROOT / "profiles" / "targets" / "kontextual-cmis-compat.json",
ROOT / "profiles" / "assessments" / "cmis-browser-baseline.json",
)
self.assertEqual(plan["extension_snapshots"][0]["id"], "open-cmis-tck")
self.assertEqual(len(plan["ordered_steps"]), 3)
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_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 test_runs_cmis_preflight_against_local_endpoint(self) -> None:
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
thread = threading.Thread(target=server.serve_forever)
thread.daemon = True
thread.start()
try:
with TemporaryDirectory() as temporary_directory:
temp_root = Path(temporary_directory)
target_path = temp_root / "target.json"
assessment_path = temp_root / "assessment.json"
target_path.write_text(
json.dumps(
{
"id": "local-cmis-test",
"subject_type": "cmis-browser-binding-endpoint",
"subject_name": "Local CMIS Test",
"environment": "test",
"scope": ["preflight"],
"endpoints": [
{
"id": "browser-binding",
"url": f"http://127.0.0.1:{server.server_port}/cmis/browser",
"binding": "cmis-browser",
}
],
"artifacts": [],
"credentials_ref": None,
"declared_capabilities": ["cmis.repository-info"],
"known_gaps": [],
}
),
encoding="utf-8",
)
assessment_path.write_text(
json.dumps(
{
"id": "local-cmis-preflight",
"framework_refs": ["cmis.browser-binding.compatibility.v1"],
"extension_refs": ["open-cmis-tck"],
"target_profile_ref": "local-cmis-test",
"selected_check_groups": {"open-cmis-tck": []},
"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": False,
"timeout_seconds": 2,
},
}
),
encoding="utf-8",
)
result = run_assessment(
ROOT,
target_path,
assessment_path,
temp_root / "run",
)
evidence = json.loads(
(Path(result["run_dir"]) / "normalized" / "evidence.json").read_text(
encoding="utf-8"
)
)
package = json.loads(
(Path(result["run_dir"]) / "reports" / "assessment-package.json").read_text(
encoding="utf-8"
)
)
mappings = json.loads(
(Path(result["run_dir"]) / "normalized" / "mappings.json").read_text(
encoding="utf-8"
)
)["mappings"]
self.assertEqual(result["status"], "completed")
self.assertEqual(evidence["evidence"][0]["result"], "pass")
self.assertEqual(
sorted(evidence["evidence"][0]["artifact_refs"]),
[
"artifacts/open-cmis-tck/preflight/response-body.bin",
"artifacts/open-cmis-tck/preflight/response-metadata.json",
],
)
self.assertEqual(
evidence["evidence"][0]["facts"]["repository_ids"],
["local-test-repository"],
)
self.assertEqual(len(package["artifact_manifest"]), 2)
self.assertEqual(mappings, [])
self.assertTrue(
(
Path(result["run_dir"])
/ "artifacts"
/ "open-cmis-tck"
/ "preflight"
/ "response-metadata.json"
).exists()
)
finally:
server.shutdown()
thread.join(timeout=5)
server.server_close()
def test_runs_cmis_tck_command_wrapper_boundary(self) -> None:
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
thread = threading.Thread(target=server.serve_forever)
thread.daemon = True
thread.start()
try:
with TemporaryDirectory() as temporary_directory:
temp_root = Path(temporary_directory)
target_path = temp_root / "target.json"
assessment_path = temp_root / "assessment.json"
waiver_path = temp_root / "waivers.json"
target_path.write_text(
json.dumps(
{
"id": "local-cmis-command-test",
"subject_type": "cmis-browser-binding-endpoint",
"subject_name": "Local CMIS Command Test",
"environment": "test",
"scope": ["preflight", "tck-wrapper"],
"endpoints": [
{
"id": "browser-binding",
"url": f"http://127.0.0.1:{server.server_port}/cmis/browser",
"binding": "cmis-browser",
}
],
"artifacts": [],
"credentials_ref": None,
"declared_capabilities": ["cmis.repository-info"],
"known_gaps": [],
}
),
encoding="utf-8",
)
assessment_path.write_text(
json.dumps(
{
"id": "local-cmis-command-boundary",
"framework_refs": ["cmis.browser-binding.compatibility.v1"],
"extension_refs": ["open-cmis-tck"],
"target_profile_ref": "local-cmis-command-test",
"selected_check_groups": {
"open-cmis-tck": ["repository-type"]
},
"expectations_ref": None,
"waivers_ref": str(waiver_path),
"output_policy": {
"report_formats": ["json", "markdown"],
"artifact_retention": "summary-only",
},
"retention_policy": {
"summary_days": 365,
"raw_artifact_days": 0,
},
"runtime_policy": {
"offline": False,
"timeout_seconds": 15,
},
}
),
encoding="utf-8",
)
waiver_path.write_text(
json.dumps(
{
"id": "local-cmis-command-waivers",
"target_profile_ref": "local-cmis-command-test",
"waivers": [
{
"id": "local-command-wrapper-bootstrap",
"scope": "test",
"requirement_refs": [],
"check_refs": [
"check-group:open-cmis-tck:repository-type"
],
"result_refs": ["blocked"],
"classification_refs": [],
"reason": "The test intentionally stops before invoking the Java/Maven TCK.",
"owner": "guide-board-tests",
"approved_by": "guide-board-tests",
"created_at": "2026-05-07",
"expires_at": "2099-12-31",
"review_status": "approved",
}
],
}
),
encoding="utf-8",
)
result = run_assessment(
ROOT,
target_path,
assessment_path,
temp_root / "run",
)
evidence = json.loads(
(Path(result["run_dir"]) / "normalized" / "evidence.json").read_text(
encoding="utf-8"
)
)["evidence"]
findings = json.loads(
(Path(result["run_dir"]) / "normalized" / "findings.json").read_text(
encoding="utf-8"
)
)["findings"]
package = json.loads(
(Path(result["run_dir"]) / "reports" / "assessment-package.json").read_text(
encoding="utf-8"
)
)
mappings = json.loads(
(Path(result["run_dir"]) / "normalized" / "mappings.json").read_text(
encoding="utf-8"
)
)["mappings"]
self.assertEqual(result["status"], "blocked")
self.assertEqual(evidence[0]["result"], "pass")
self.assertEqual(evidence[1]["result"], "blocked")
self.assertEqual(evidence[1]["facts"]["runner_kind"], "command")
self.assertIn(
evidence[1]["facts"]["blocked_reason"],
{"missing_dependency", "tck_invocation_not_configured"},
)
self.assertEqual(
findings[0]["classification"],
evidence[1]["facts"]["blocked_reason"],
)
self.assertEqual(findings[0]["waiver_ref"], "local-command-wrapper-bootstrap")
self.assertEqual(package["policy_summary"]["applied_waivers"], 1)
self.assertGreaterEqual(len(package["artifact_manifest"]), 3)
self.assertEqual(len(mappings), 2)
self.assertEqual(
{mapping["target_id"] for mapping in mappings},
{"repository-type"},
)
self.assertEqual(
package["mapping_summary"]["targets"][0]["results"],
{"blocked": 2},
)
finally:
server.shutdown()
thread.join(timeout=5)
server.server_close()
def test_preflight_failure_blocks_downstream_checks(self) -> None:
with TemporaryDirectory() as temporary_directory:
temp_root = Path(temporary_directory)
target_path = temp_root / "target.json"
assessment_path = temp_root / "assessment.json"
target_path.write_text(
json.dumps(
{
"id": "local-cmis-preflight-failure",
"subject_type": "cmis-browser-binding-endpoint",
"subject_name": "Local CMIS Preflight Failure",
"environment": "test",
"scope": ["preflight", "tck-wrapper"],
"endpoints": [
{
"id": "browser-binding",
"url": "http://127.0.0.1:9/cmis/browser",
"binding": "cmis-browser",
}
],
"artifacts": [],
"credentials_ref": None,
"declared_capabilities": ["cmis.repository-info"],
"known_gaps": [],
}
),
encoding="utf-8",
)
assessment_path.write_text(
json.dumps(
{
"id": "local-cmis-preflight-gate",
"framework_refs": ["cmis.browser-binding.compatibility.v1"],
"extension_refs": ["open-cmis-tck"],
"target_profile_ref": "local-cmis-preflight-failure",
"selected_check_groups": {
"open-cmis-tck": ["repository-type"]
},
"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": False,
"timeout_seconds": 1,
},
}
),
encoding="utf-8",
)
result = run_assessment(
ROOT,
target_path,
assessment_path,
temp_root / "run",
)
run_dir = Path(result["run_dir"])
evidence = json.loads(
(run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8")
)["evidence"]
findings = json.loads(
(run_dir / "normalized" / "findings.json").read_text(encoding="utf-8")
)["findings"]
self.assertEqual(result["status"], "infrastructure_error")
self.assertEqual(evidence[0]["result"], "infrastructure_error")
self.assertEqual(evidence[1]["result"], "blocked")
self.assertEqual(evidence[1]["facts"]["blocked_reason"], "preflight_failed")
self.assertEqual(
evidence[1]["facts"]["preflight_evidence_ref"],
evidence[0]["id"],
)
self.assertFalse((run_dir / "artifacts" / "runner-contexts").exists())
self.assertEqual(findings[1]["classification"], "preflight_failed")
self.assertTrue(findings[1]["expected"])
class _CmisHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
body = json.dumps(
{
"local-test-repository": {
"repositoryId": "local-test-repository",
"repositoryName": "Local Test Repository",
"cmisVersionSupported": "1.1",
"capabilities": {},
}
}
).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, format: str, *args: object) -> None:
return
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",
)
if __name__ == "__main__":
unittest.main()