open-cmis-tck/tests/test_open_cmis_tck.py

from __future__ import annotations

import http.client
import json
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
from guide_board.retention import build_trend_summary
from guide_board.service import ServiceHandle, start_service
from open_cmis_tck.profile import validate_cmis_profile_config


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

        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_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_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)
        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


if __name__ == "__main__":
    unittest.main()