open-cmis-tck/tests/test_core.py

from __future__ import annotations

import unittest
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from tempfile import TemporaryDirectory
from pathlib import Path

from guide_board.discovery import discover_extensions
from guide_board.execution import run_assessment
from guide_board.planning import (
    build_run_plan,
    validate_assessment_profile,
    validate_target_profile,
)


ROOT = Path(__file__).resolve().parents[1]


class CoreArchitectureTests(unittest.TestCase):
    def test_discovers_incubating_extensions(self) -> None:
        extensions = {extension.id for extension in discover_extensions(ROOT)}

        self.assertIn("sample-noop", extensions)
        self.assertIn("open-cmis-tck", extensions)

    def test_validates_sample_profiles(self) -> None:
        target = validate_target_profile(ROOT / "profiles" / "targets" / "sample-repository.json")
        assessment = validate_assessment_profile(
            ROOT / "profiles" / "assessments" / "sample-noop.json"
        )

        self.assertEqual(target["id"], "sample-repository")
        self.assertEqual(assessment["target_profile_ref"], "sample-repository")

    def test_builds_sample_run_plan(self) -> None:
        plan = build_run_plan(
            ROOT,
            ROOT / "profiles" / "targets" / "sample-repository.json",
            ROOT / "profiles" / "assessments" / "sample-noop.json",
        )

        self.assertEqual(plan["target_profile_snapshot"]["id"], "sample-repository")
        self.assertEqual(plan["extension_snapshots"][0]["id"], "sample-noop")
        self.assertEqual(
            [step["id"] for step in plan["ordered_steps"]],
            [
                "preflight:sample-noop",
                "check-group:sample-noop:profile-shape",
            ],
        )
        self.assertEqual(
            plan["ordered_steps"][1]["requirement_refs"],
            ["guide-board.sample-readiness.v0.profile-shape"],
        )

    def test_builds_cmis_baseline_plan(self) -> None:
        plan = build_run_plan(
            ROOT,
            ROOT / "profiles" / "targets" / "kontextual-cmis-compat.json",
            ROOT / "profiles" / "assessments" / "cmis-browser-baseline.json",
        )

        self.assertEqual(plan["extension_snapshots"][0]["id"], "open-cmis-tck")
        self.assertEqual(len(plan["ordered_steps"]), 3)

    def test_runs_sample_noop_assessment(self) -> None:
        with TemporaryDirectory() as temporary_directory:
            result = run_assessment(
                ROOT,
                ROOT / "profiles" / "targets" / "sample-repository.json",
                ROOT / "profiles" / "assessments" / "sample-noop.json",
                Path(temporary_directory) / "sample-run",
            )

            run_dir = Path(result["run_dir"])
            self.assertEqual(result["status"], "completed")
            self.assertTrue((run_dir / "run.json").exists())
            self.assertTrue((run_dir / "normalized" / "evidence.json").exists())
            self.assertTrue((run_dir / "reports" / "assessment-package.json").exists())
            self.assertTrue((run_dir / "reports" / "report.md").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"
                target_path.write_text(
                    json.dumps(
                        {
                            "id": "local-cmis-test",
                            "subject_type": "cmis-browser-binding-endpoint",
                            "subject_name": "Local CMIS Test",
                            "environment": "test",
                            "scope": ["preflight"],
                            "endpoints": [
                                {
                                    "id": "browser-binding",
                                    "url": f"http://127.0.0.1:{server.server_port}/cmis/browser",
                                    "binding": "cmis-browser",
                                }
                            ],
                            "artifacts": [],
                            "credentials_ref": None,
                            "declared_capabilities": ["cmis.repository-info"],
                            "known_gaps": [],
                        }
                    ),
                    encoding="utf-8",
                )
                assessment_path.write_text(
                    json.dumps(
                        {
                            "id": "local-cmis-preflight",
                            "framework_refs": ["cmis.browser-binding.compatibility.v1"],
                            "extension_refs": ["open-cmis-tck"],
                            "target_profile_ref": "local-cmis-test",
                            "selected_check_groups": {"open-cmis-tck": []},
                            "expectations_ref": None,
                            "waivers_ref": None,
                            "output_policy": {
                                "report_formats": ["json", "markdown"],
                                "artifact_retention": "summary-only",
                            },
                            "retention_policy": {
                                "summary_days": 365,
                                "raw_artifact_days": 0,
                            },
                            "runtime_policy": {
                                "offline": False,
                                "timeout_seconds": 2,
                            },
                        }
                    ),
                    encoding="utf-8",
                )

                result = run_assessment(
                    ROOT,
                    target_path,
                    assessment_path,
                    temp_root / "run",
                )
                evidence = json.loads(
                    (Path(result["run_dir"]) / "normalized" / "evidence.json").read_text(
                        encoding="utf-8"
                    )
                )

                self.assertEqual(result["status"], "completed")
                self.assertEqual(evidence["evidence"][0]["result"], "pass")
                self.assertEqual(
                    evidence["evidence"][0]["facts"]["repository_ids"],
                    ["local-test-repository"],
                )
        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"
                target_path.write_text(
                    json.dumps(
                        {
                            "id": "local-cmis-command-test",
                            "subject_type": "cmis-browser-binding-endpoint",
                            "subject_name": "Local CMIS Command Test",
                            "environment": "test",
                            "scope": ["preflight", "tck-wrapper"],
                            "endpoints": [
                                {
                                    "id": "browser-binding",
                                    "url": f"http://127.0.0.1:{server.server_port}/cmis/browser",
                                    "binding": "cmis-browser",
                                }
                            ],
                            "artifacts": [],
                            "credentials_ref": None,
                            "declared_capabilities": ["cmis.repository-info"],
                            "known_gaps": [],
                        }
                    ),
                    encoding="utf-8",
                )
                assessment_path.write_text(
                    json.dumps(
                        {
                            "id": "local-cmis-command-boundary",
                            "framework_refs": ["cmis.browser-binding.compatibility.v1"],
                            "extension_refs": ["open-cmis-tck"],
                            "target_profile_ref": "local-cmis-command-test",
                            "selected_check_groups": {
                                "open-cmis-tck": ["repository-type"]
                            },
                            "expectations_ref": None,
                            "waivers_ref": None,
                            "output_policy": {
                                "report_formats": ["json", "markdown"],
                                "artifact_retention": "summary-only",
                            },
                            "retention_policy": {
                                "summary_days": 365,
                                "raw_artifact_days": 0,
                            },
                            "runtime_policy": {
                                "offline": False,
                                "timeout_seconds": 15,
                            },
                        }
                    ),
                    encoding="utf-8",
                )

                result = run_assessment(
                    ROOT,
                    target_path,
                    assessment_path,
                    temp_root / "run",
                )
                evidence = json.loads(
                    (Path(result["run_dir"]) / "normalized" / "evidence.json").read_text(
                        encoding="utf-8"
                    )
                )["evidence"]
                findings = json.loads(
                    (Path(result["run_dir"]) / "normalized" / "findings.json").read_text(
                        encoding="utf-8"
                    )
                )["findings"]

                self.assertEqual(result["status"], "blocked")
                self.assertEqual(evidence[0]["result"], "pass")
                self.assertEqual(evidence[1]["result"], "blocked")
                self.assertEqual(evidence[1]["facts"]["runner_kind"], "command")
                self.assertIn(
                    evidence[1]["facts"]["blocked_reason"],
                    {"missing_dependency", "tck_invocation_not_configured"},
                )
                self.assertEqual(
                    findings[0]["classification"],
                    evidence[1]["facts"]["blocked_reason"],
                )
        finally:
            server.shutdown()
            thread.join(timeout=5)
            server.server_close()


class _CmisHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        body = json.dumps(
            {
                "local-test-repository": {
                    "repositoryId": "local-test-repository",
                    "repositoryName": "Local Test Repository",
                    "cmisVersionSupported": "1.1",
                    "capabilities": {},
                }
            }
        ).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format: str, *args: object) -> None:
        return


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