from __future__ import annotations import base64 import http.client import json import os import subprocess import sys import threading import time import unittest from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path from tempfile import TemporaryDirectory 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 build_trend_summary from guide_board.service import ServiceHandle, start_service from open_cmis_tck.bootstrap import TCK_COORDINATE, check_runtime from open_cmis_tck.normalization import ( aggregate_case_result, parse_text_report, result_counts, ) from open_cmis_tck.profile import validate_cmis_profile_config from open_cmis_tck.scorecard import build_scorecard, write_scorecard ROOT = Path(__file__).resolve().parents[1] CORE_ROOT = ROOT.parent / "guide-board" class OpenCmisTckExtensionTests(unittest.TestCase): def test_extension_manifest_discovers_from_repo_root(self) -> None: extensions = { extension.id: extension for extension in discover_extensions(CORE_ROOT, [ROOT]) } self.assertIn("open-cmis-tck", extensions) self.assertEqual(extensions["open-cmis-tck"].source, "external") self.assertEqual(extensions["open-cmis-tck"].path, ROOT) def test_builds_cmis_baseline_plan_from_external_extension(self) -> None: assessment = validate_assessment_profile( ROOT / "profiles" / "assessments" / "cmis-browser-baseline.json" ) plan = build_run_plan( CORE_ROOT, ROOT / "profiles" / "targets" / "kontextual-cmis-compat.json", ROOT / "profiles" / "assessments" / "cmis-browser-baseline.json", [ROOT], ) self.assertEqual(assessment["extension_refs"], ["open-cmis-tck"]) self.assertEqual(plan["extension_snapshots"][0]["id"], "open-cmis-tck") self.assertEqual(plan["extension_snapshots"][0]["source"], "external") self.assertEqual(plan["extension_snapshots"][0]["path"], str(ROOT)) self.assertEqual(len(plan["ordered_steps"]), 3) def test_validates_cmis_profile_config_with_actionable_diagnostics(self) -> None: target = json.loads( (ROOT / "profiles" / "targets" / "kontextual-cmis-compat.json").read_text( encoding="utf-8" ) ) assessment = json.loads( (ROOT / "profiles" / "assessments" / "cmis-browser-baseline.json").read_text( encoding="utf-8" ) ) diagnostics = validate_cmis_profile_config(target, assessment) self.assertEqual(diagnostics["status"], "valid") self.assertEqual( diagnostics["cmis_config"]["browser_binding_url"], "http://127.0.0.1:8000/cmis/compat-tck/browser", ) self.assertEqual(diagnostics["cmis_config"]["repository_id"], "compat-tck") self.assertEqual(diagnostics["cmis_config"]["auth_mode"], "anonymous") authenticated = dict(target) authenticated["credentials_ref"] = "env:CMIS_TCK_USER,CMIS_TCK_PASSWORD" authenticated_diagnostics = validate_cmis_profile_config(authenticated, assessment) self.assertEqual(authenticated_diagnostics["cmis_config"]["auth_mode"], "env") broken = dict(target) broken["endpoints"] = [] broken_diagnostics = validate_cmis_profile_config(broken, assessment) self.assertEqual(broken_diagnostics["status"], "invalid") self.assertIn( "Add one endpoint with binding 'cmis-browser'", broken_diagnostics["diagnostics"][0]["message"], ) def test_target_profile_templates_validate(self) -> None: template_dir = ROOT / "profiles" / "targets" / "templates" for name in [ "cmis-browser-anonymous.json", "cmis-browser-basic-auth-env.json", "cmis-browser-basic-auth-file.json", ]: with self.subTest(name=name): profile = validate_target_profile(template_dir / name) self.assertEqual(profile["subject_type"], "cmis-browser-binding-endpoint") def test_inmemory_pilot_profiles_validate(self) -> None: target = validate_target_profile(ROOT / "profiles" / "targets" / "opencmis-inmemory-local.json") assessment = validate_assessment_profile( ROOT / "profiles" / "assessments" / "cmis-browser-inmemory-pilot.json" ) plan = build_run_plan( CORE_ROOT, ROOT / "profiles" / "targets" / "opencmis-inmemory-local.json", ROOT / "profiles" / "assessments" / "cmis-browser-inmemory-pilot.json", [ROOT], ) self.assertEqual(target["credentials_ref"], "env:OPENCMIS_INMEMORY_USER,OPENCMIS_INMEMORY_PASSWORD") self.assertEqual(assessment["target_profile_ref"], "opencmis-inmemory-local") self.assertEqual( [step["id"] for step in plan["ordered_steps"]], ["preflight:open-cmis-tck", "check-group:open-cmis-tck:repository-type"], ) def test_bootstrap_reports_local_tck_runtime_posture(self) -> None: with TemporaryDirectory() as temporary_directory: output = Path(temporary_directory) / "runtime-summary.json" summary = check_runtime(ROOT, output, resolve=False) self.assertIn(summary["status"], {"ready", "blocked"}) self.assertEqual(summary["tck"]["coordinate"], TCK_COORDINATE) self.assertEqual( summary["tck"]["runner_class"], "org.apache.chemistry.opencmis.tck.runner.ConsoleRunner", ) self.assertTrue(output.exists()) def test_bootstrap_ready_path_with_fake_local_toolchain(self) -> None: with TemporaryDirectory() as temporary_directory: temp_root = Path(temporary_directory) bin_dir = temp_root / "bin" bin_dir.mkdir() java = bin_dir / "java" maven = bin_dir / "mvn" java.write_text("#!/usr/bin/env sh\necho 'openjdk version \"17\"' >&2\n", encoding="utf-8") maven.write_text("#!/usr/bin/env sh\necho 'Apache Maven 3.9.0'\n", encoding="utf-8") java.chmod(0o755) maven.chmod(0o755) output = temp_root / "runtime-summary.json" original_path = os.environ.get("PATH", "") os.environ["PATH"] = f"{bin_dir}{os.pathsep}{original_path}" try: summary = check_runtime(ROOT, output, resolve=False) finally: os.environ["PATH"] = original_path self.assertEqual(summary["status"], "ready") self.assertTrue(summary["runtime"]["java"]["available"]) self.assertTrue(summary["runtime"]["maven"]["available"]) self.assertTrue(output.exists()) def test_parses_native_opencmis_text_report_fixture(self) -> None: fixture = (ROOT / "tests" / "fixtures" / "opencmis-text-report-sanitized.txt").read_text( encoding="utf-8" ) cases = parse_text_report( fixture, "repository-type", ["org.apache.chemistry.opencmis.tck.tests.basics.BasicsTestGroup"], ) counts = result_counts(cases) warning = next(case for case in cases if case["status"] == "warning") failure = next(case for case in cases if case["status"] == "fail") self.assertEqual( counts, { "fail": 1, "infrastructure_error": 1, "pass": 2, "skipped": 1, "warning": 1, }, ) self.assertEqual(aggregate_case_result(counts, 0), "infrastructure_error") self.assertEqual(warning["status_native"], "WARNING") self.assertEqual(warning["test_name"], "Repository Info Test") self.assertEqual(warning["source_location"], {"file": "SecurityTest.java", "line": 52}) self.assertEqual(failure["message"], "Test folder could not be created.") def test_console_adapter_dry_run_writes_session_and_group_files(self) -> None: with TemporaryDirectory() as temporary_directory: temp_root = Path(temporary_directory) run_dir = temp_root / "run" artifact_dir = run_dir / "artifacts" / "open-cmis-tck" / "tck" / "repository-type" completed = subprocess.run( [ sys.executable, str(ROOT / "adapters" / "opencmis_console_adapter.py"), "--browser-url", "http://127.0.0.1:8000/cmis/browser", "--repository-id", "local-test-repository", "--check-group", "repository-type", "--artifact-dir", str(artifact_dir), "--run-dir", str(run_dir), "--extension-path", str(ROOT), "--dry-run", ], capture_output=True, text=True, check=False, ) result = json.loads(completed.stdout) self.assertEqual(completed.returncode, 0) self.assertEqual(result["result"], "skipped") self.assertIn( "artifacts/open-cmis-tck/tck/repository-type/session.properties.redacted", result["artifact_refs"], ) self.assertIn( "org.apache.chemistry.opencmis.binding.spi.type=browser", (artifact_dir / "session.properties.redacted").read_text(encoding="utf-8"), ) self.assertIn( "org.apache.chemistry.opencmis.tck.tests.basics.BasicsTestGroup", (artifact_dir / "groups.txt").read_text(encoding="utf-8"), ) self.assertFalse((artifact_dir / "session-private.properties").exists()) def test_console_adapter_uses_env_credentials_without_persisting_secret(self) -> None: with TemporaryDirectory() as temporary_directory: temp_root = Path(temporary_directory) run_dir = temp_root / "run" artifact_dir = run_dir / "artifacts" / "open-cmis-tck" / "tck" / "repository-type" env = { **os.environ, "CMIS_TEST_USER": "alice", "CMIS_TEST_PASSWORD": "secret-password", } completed = subprocess.run( [ sys.executable, str(ROOT / "adapters" / "opencmis_console_adapter.py"), "--browser-url", "http://127.0.0.1:8000/cmis/browser", "--repository-id", "local-test-repository", "--check-group", "repository-type", "--artifact-dir", str(artifact_dir), "--run-dir", str(run_dir), "--extension-path", str(ROOT), "--credentials-ref", "env:CMIS_TEST_USER,CMIS_TEST_PASSWORD", "--dry-run", ], capture_output=True, text=True, check=False, env=env, ) result = json.loads(completed.stdout) redacted = (artifact_dir / "session.properties.redacted").read_text( encoding="utf-8" ) self.assertEqual(completed.returncode, 0) self.assertEqual(result["result"], "skipped") self.assertIn("org.apache.chemistry.opencmis.user=alice", redacted) self.assertIn("org.apache.chemistry.opencmis.password=", redacted) self.assertNotIn("secret-password", redacted) self.assertFalse((artifact_dir / "session-private.properties").exists()) 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" _write_target(target_path, server.server_port, "local-cmis-test") _write_assessment( assessment_path, "local-cmis-preflight", "local-cmis-test", [], None, ) result = run_assessment( CORE_ROOT, target_path, assessment_path, temp_root / "run", [ROOT], ) run_dir = Path(result["run_dir"]) evidence = json.loads( (run_dir / "normalized" / "evidence.json").read_text( encoding="utf-8" ) )["evidence"] package = json.loads( (run_dir / "reports" / "assessment-package.json").read_text( encoding="utf-8" ) ) self.assertEqual(result["status"], "completed") self.assertEqual(evidence[0]["result"], "pass") self.assertEqual( evidence[0]["facts"]["repository_ids"], ["local-test-repository"], ) posture = { item["requirement_ref"]: item["status"] for item in evidence[0]["facts"]["capability_posture"] } self.assertEqual(posture["cmis.repository-info"], "supported") self.assertEqual(len(package["artifact_manifest"]), 2) self.assertTrue( ( run_dir / "artifacts" / "open-cmis-tck" / "preflight" / "response-metadata.json" ).exists() ) finally: server.shutdown() thread.join(timeout=5) server.server_close() def test_preflight_accepts_unsupported_optional_capability_as_known_gap(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" _write_target(target_path, server.server_port, "local-cmis-query-gap") target = json.loads(target_path.read_text(encoding="utf-8")) target["declared_capabilities"].append("cmis.query") target["known_gaps"].append( { "id": "query-not-targeted", "requirement_refs": ["cmis.query"], "reason": "The local fixture deliberately reports no query support.", "status": "unsupported_by_design", } ) target_path.write_text(json.dumps(target), encoding="utf-8") _write_assessment( assessment_path, "local-cmis-known-gap", "local-cmis-query-gap", [], None, ) result = run_assessment( CORE_ROOT, target_path, assessment_path, temp_root / "run", [ROOT], ) evidence = json.loads( (Path(result["run_dir"]) / "normalized" / "evidence.json").read_text( encoding="utf-8" ) )["evidence"] posture = { item["requirement_ref"]: item["status"] for item in evidence[0]["facts"]["capability_posture"] } self.assertEqual(result["status"], "completed") self.assertEqual(posture["cmis.query"], "expected_gap") finally: server.shutdown() thread.join(timeout=5) server.server_close() def test_preflight_uses_env_credentials_for_basic_auth(self) -> None: server = HTTPServer(("127.0.0.1", 0), _BasicAuthCmisHandler) thread = threading.Thread(target=server.serve_forever) thread.daemon = True thread.start() original_user = os.environ.get("CMIS_AUTH_USER") original_password = os.environ.get("CMIS_AUTH_PASSWORD") os.environ["CMIS_AUTH_USER"] = "alice" os.environ["CMIS_AUTH_PASSWORD"] = "secret" try: with TemporaryDirectory() as temporary_directory: temp_root = Path(temporary_directory) target_path = temp_root / "target.json" assessment_path = temp_root / "assessment.json" _write_target(target_path, server.server_port, "local-cmis-auth-preflight") target = json.loads(target_path.read_text(encoding="utf-8")) target["credentials_ref"] = "env:CMIS_AUTH_USER,CMIS_AUTH_PASSWORD" target_path.write_text(json.dumps(target), encoding="utf-8") _write_assessment( assessment_path, "local-cmis-auth-preflight", "local-cmis-auth-preflight", [], None, ) result = run_assessment( CORE_ROOT, target_path, assessment_path, temp_root / "run", [ROOT], ) evidence = json.loads( (Path(result["run_dir"]) / "normalized" / "evidence.json").read_text( encoding="utf-8" ) )["evidence"] self.assertEqual(result["status"], "completed") self.assertEqual(evidence[0]["result"], "pass") self.assertEqual(evidence[0]["facts"]["auth_mode"], "env") finally: if original_user is None: os.environ.pop("CMIS_AUTH_USER", None) else: os.environ["CMIS_AUTH_USER"] = original_user if original_password is None: os.environ.pop("CMIS_AUTH_PASSWORD", None) else: os.environ["CMIS_AUTH_PASSWORD"] = original_password 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" _write_target(target_path, server.server_port, "local-cmis-command-test") _write_assessment( assessment_path, "local-cmis-command-boundary", "local-cmis-command-test", ["repository-type"], str(waiver_path), ) _write_command_waiver(waiver_path, "local-cmis-command-test") result = run_assessment( CORE_ROOT, target_path, assessment_path, temp_root / "run", [ROOT], ) 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"] mappings = json.loads( (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]["waiver_ref"], "local-command-wrapper-bootstrap") self.assertEqual({mapping["target_id"] for mapping in mappings}, {"repository-type"}) finally: server.shutdown() thread.join(timeout=5) server.server_close() def test_runs_configured_tck_command_and_normalizes_json_results(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" fake_tck = temp_root / "fake_tck.py" fake_tck.write_text( "\n".join( [ "import json", "print(json.dumps({", " 'tests': [", " {'id': 'repository-info', 'status': 'pass'},", " {'id': 'type-definitions', 'status': 'pass'}", " ]", "}))", ] ), encoding="utf-8", ) _write_target(target_path, server.server_port, "local-cmis-configured-tck") _write_assessment( assessment_path, "local-cmis-configured-tck", "local-cmis-configured-tck", ["repository-type"], None, { "requires_java_maven": False, "repository_id": "local-test-repository", "command": [ sys.executable, str(fake_tck), "--url", "{browser_url}", "--repository", "{repository_id}", "--group", "{check_group}", ], }, ) result = run_assessment( CORE_ROOT, target_path, assessment_path, temp_root / "run", [ROOT], ) run_dir = Path(result["run_dir"]) evidence = json.loads( (run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8") )["evidence"] retention = json.loads( (run_dir / "retention-summary.json").read_text(encoding="utf-8") ) trend = build_trend_summary(temp_root) self.assertEqual(result["status"], "completed") self.assertEqual(evidence[1]["result"], "pass") self.assertEqual(evidence[1]["facts"]["normalizer"], "json-cases") self.assertEqual(evidence[1]["facts"]["result_counts"], {"pass": 2}) self.assertTrue( ( run_dir / "artifacts" / "open-cmis-tck" / "tck" / "repository-type" / "stdout.log" ).exists() ) self.assertEqual(retention["summary"]["status"], "completed") self.assertGreaterEqual(retention["summary"]["artifact_count"], 4) self.assertEqual(trend["run_count"], 1) scorecard = build_scorecard(run_dir) self.assertEqual(scorecard["run_id"], result["run_id"]) groups = {group["id"]: group for group in scorecard["groups"]} self.assertEqual(groups["repository-type"]["status"], "demonstrated") self.assertEqual(groups["repository-type"]["score"], 4) self.assertEqual(groups["object-content"]["status"], "not_assessed") written = write_scorecard(run_dir) self.assertTrue(Path(written["json"]).exists()) self.assertTrue(Path(written["markdown"]).exists()) finally: server.shutdown() thread.join(timeout=5) server.server_close() def test_runs_configured_tck_command_and_normalizes_text_report_results(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" fake_tck = temp_root / "fake_tck_text.py" fixture = ROOT / "tests" / "fixtures" / "opencmis-text-report-sanitized.txt" fake_tck.write_text( "\n".join( [ "from pathlib import Path", f"print(Path({str(fixture)!r}).read_text(encoding='utf-8'))", ] ), encoding="utf-8", ) _write_target(target_path, server.server_port, "local-cmis-text-tck") _write_assessment( assessment_path, "local-cmis-text-tck", "local-cmis-text-tck", ["repository-type"], None, { "requires_java_maven": False, "repository_id": "local-test-repository", "command": [sys.executable, str(fake_tck)], }, ) result = run_assessment( CORE_ROOT, target_path, assessment_path, temp_root / "run", [ROOT], ) run_dir = Path(result["run_dir"]) evidence = json.loads( (run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8") )["evidence"] cases = evidence[1]["facts"]["cases"] self.assertEqual(result["status"], "infrastructure_error") self.assertEqual(evidence[1]["result"], "infrastructure_error") self.assertEqual(evidence[1]["facts"]["normalizer"], "opencmis-text-report") self.assertEqual( evidence[1]["facts"]["result_counts"], { "fail": 1, "infrastructure_error": 1, "pass": 2, "skipped": 1, "warning": 1, }, ) self.assertEqual(cases[0]["status_native"], "OK") self.assertEqual(cases[0]["group_name"], "Basics Test Group") self.assertEqual(cases[0]["test_name"], "Repository Info Test") self.assertIn( "artifacts/open-cmis-tck/tck/repository-type/stdout.log", evidence[1]["artifact_refs"], ) finally: server.shutdown() thread.join(timeout=5) server.server_close() def test_guide_board_dry_run_invokes_console_adapter_and_captures_artifacts(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" _write_target(target_path, server.server_port, "local-cmis-guide-dry-run") _write_assessment( assessment_path, "local-cmis-guide-dry-run", "local-cmis-guide-dry-run", ["repository-type"], None, { "requires_java_maven": False, "repository_id": "local-test-repository", "command": [ sys.executable, str(ROOT / "adapters" / "opencmis_console_adapter.py"), "--browser-url", "{browser_url}", "--repository-id", "{repository_id}", "--check-group", "{check_group}", "--artifact-dir", "{artifact_dir}", "--run-dir", "{run_dir}", "--extension-path", "{extension_path}", "--credentials-ref", "{credentials_ref}", "--target-profile-dir", "{target_profile_dir}", "--timeout-seconds", "{timeout_seconds}", "--dry-run", ], }, ) result = run_assessment( CORE_ROOT, target_path, assessment_path, temp_root / "run", [ROOT], ) run_dir = Path(result["run_dir"]) evidence = json.loads( (run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8") )["evidence"] package = json.loads( (run_dir / "reports" / "assessment-package.json").read_text( encoding="utf-8" ) ) artifact_paths = { item["path"] for item in package["artifact_manifest"] } self.assertEqual(result["status"], "completed") self.assertEqual(evidence[1]["result"], "skipped") self.assertEqual( evidence[1]["facts"]["adapter"], "opencmis-console-runner", ) self.assertIn( "artifacts/open-cmis-tck/tck/repository-type/session.properties.redacted", artifact_paths, ) self.assertIn( "artifacts/open-cmis-tck/tck/repository-type/groups.txt", artifact_paths, ) self.assertFalse( ( run_dir / "artifacts" / "open-cmis-tck" / "tck" / "repository-type" / "session-private.properties" ).exists() ) finally: server.shutdown() thread.join(timeout=5) server.server_close() def test_guide_board_service_runs_cmis_extension(self) -> None: server = HTTPServer(("127.0.0.1", 0), _CmisHandler) thread = threading.Thread(target=server.serve_forever) thread.daemon = True thread.start() service = start_service(CORE_ROOT, [ROOT], host="127.0.0.1", port=0) try: with TemporaryDirectory() as temporary_directory: temp_root = Path(temporary_directory) target_path = temp_root / "target.json" assessment_path = temp_root / "assessment.json" _write_target(target_path, server.server_port, "local-cmis-service") _write_assessment( assessment_path, "local-cmis-service", "local-cmis-service", [], None, ) extensions = _request_json(service, "GET", "/extensions") self.assertIn( "open-cmis-tck", [extension["id"] for extension in extensions["extensions"]], ) job = _request_json( service, "POST", "/runs", { "target": str(target_path), "assessment": str(assessment_path), "output_dir": str(temp_root / "service-run"), }, expected_status=202, ) status = _wait_for_job(service, job["job_id"]) reports = _request_json(service, "GET", f"/runs/{job['job_id']}/reports") self.assertEqual(status["status"], "succeeded") self.assertEqual(status["result"]["status"], "completed") self.assertEqual( reports["assessment_package"]["json"]["extensions"][0]["id"], "open-cmis-tck", ) finally: service.stop() 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" _write_failing_target(target_path) _write_assessment( assessment_path, "local-cmis-preflight-gate", "local-cmis-preflight-failure", ["repository-type"], None, ) result = run_assessment( CORE_ROOT, target_path, assessment_path, temp_root / "run", [ROOT], ) 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.assertFalse((run_dir / "artifacts" / "runner-contexts").exists()) self.assertEqual(findings[1]["classification"], "preflight_failed") self.assertTrue(findings[1]["expected"]) def _write_target(path: Path, port: int, target_id: str) -> None: path.write_text( json.dumps( { "id": target_id, "subject_type": "cmis-browser-binding-endpoint", "subject_name": "Local CMIS Test", "environment": "test", "scope": ["preflight", "tck-wrapper"], "endpoints": [ { "id": "browser-binding", "url": f"http://127.0.0.1:{port}/cmis/browser", "binding": "cmis-browser", } ], "artifacts": [], "credentials_ref": None, "declared_capabilities": ["cmis.repository-info"], "known_gaps": [], } ), encoding="utf-8", ) def _write_failing_target(path: Path) -> None: 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", ) def _write_assessment( path: Path, assessment_id: str, target_id: str, check_groups: list[str], waiver_ref: str | None, opencmis_policy: dict[str, object] | None = None, ) -> None: runtime_policy: dict[str, object] = { "offline": False, "timeout_seconds": 15, } if opencmis_policy is not None: runtime_policy["opencmis_tck"] = opencmis_policy path.write_text( json.dumps( { "id": assessment_id, "framework_refs": ["cmis.browser-binding.compatibility.v1"], "extension_refs": ["open-cmis-tck"], "target_profile_ref": target_id, "selected_check_groups": {"open-cmis-tck": check_groups}, "expectations_ref": None, "waivers_ref": waiver_ref, "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_command_waiver(path: Path, target_id: str) -> None: path.write_text( json.dumps( { "id": "local-cmis-command-waivers", "target_profile_ref": target_id, "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 stops before invoking the Java/Maven TCK.", "owner": "open-cmis-tck-tests", "approved_by": "open-cmis-tck-tests", "created_at": "2026-05-07", "expires_at": "2099-12-31", "review_status": "approved", } ], } ), 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}") 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": { "capabilityACL": "discover", "capabilityChanges": "none", "capabilityGetDescendants": True, "capabilityGetFolderTree": True, "capabilityQuery": "none", }, } } ).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 class _BasicAuthCmisHandler(_CmisHandler): def do_GET(self) -> None: expected = "Basic " + base64.b64encode(b"alice:secret").decode("ascii") if self.headers.get("Authorization") != expected: body = b"unauthorized" self.send_response(401) self.send_header("Content-Type", "text/plain") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return super().do_GET() if __name__ == "__main__": unittest.main()