Files
open-cmis-tck/tests/test_core.py
2026-05-07 16:01:33 +02:00

474 lines
20 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.planning import (
build_run_plan,
validate_assessment_profile,
validate_target_profile,
)
from guide_board.retention import list_retained_runs
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_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
if __name__ == "__main__":
unittest.main()