open-cmis-tck/tests/test_open_cmis_tck.py

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>", 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()