tests/test_plugin_contract.py

"""Lean plugin packaging and workflow contract tests."""

from __future__ import annotations

import json
import subprocess
import sys
import unittest
from pathlib import Path

PLUGIN_ROOT = Path(__file__).resolve().parent.parent
MANIFEST = PLUGIN_ROOT / ".codex-plugin" / "plugin.json"
README = PLUGIN_ROOT / "README.md"
SKILL = PLUGIN_ROOT / "skills" / "codex-replay" / "SKILL.md"
AGENT = PLUGIN_ROOT / "skills" / "codex-replay" / "agents" / "openai.yaml"
BASELINE_HANDLING = PLUGIN_ROOT / "BASELINE_HANDLING.md"
RUNNER = PLUGIN_ROOT / "scripts" / "historical_replay.py"
MCP_SERVER = PLUGIN_ROOT / "mcp" / "server.py"
MCP_CONFIGURATION = PLUGIN_ROOT / "mcp" / "replay_configuration.py"
MCP_CONTROLLER = PLUGIN_ROOT / "mcp" / "controller.html"
MCP_CONTROLLER_CSS = PLUGIN_ROOT / "mcp" / "controller.css"
MCP_WORKER = PLUGIN_ROOT / "mcp" / "codex-worker.mjs"


class PluginContractTests(unittest.TestCase):
    def test_manifest_is_valid_and_points_inside_plugin(self) -> None:
        payload = json.loads(MANIFEST.read_text(encoding="utf-8"))
        self.assertEqual(payload["name"], "codex-replay")
        self.assertEqual(payload["skills"], "./skills/")
        self.assertTrue((PLUGIN_ROOT / payload["skills"]).is_dir())
        self.assertEqual(payload["interface"]["displayName"], "Codex Replay")
        self.assertIn("in-app or system browser", payload["interface"]["shortDescription"])
        self.assertIn("in-app browser", payload["description"])
        self.assertIn("system browser", payload["description"])
        self.assertIn("Codex CLI", payload["interface"]["longDescription"])
        self.assertIn("available loopback port", payload["interface"]["longDescription"])
        self.assertIn(
            "Multiple replay sessions can run in parallel",
            payload["interface"]["longDescription"],
        )
        self.assertNotIn("embedded", json.dumps(payload).lower())
        self.assertNotIn("MCP App", json.dumps(payload))

    def test_beginning_and_end_state_document_contract(self) -> None:
        text = BASELINE_HANDLING.read_text(encoding="utf-8")
        flat = " ".join(text.split())
        self.assertIn("Git     -> Git", text)
        self.assertIn("Non-Git -> Git", text)
        self.assertIn("Non-Git -> Non-Git", text)
        self.assertIn("`Git -> Non-Git` is invalid", text)
        self.assertIn("--confirm-empty-beginning", text)
        self.assertIn("does not confirm the beginning state", text)
        self.assertIn("diffing the empty Git tree", text)
        self.assertIn("All committed files come from the ending commit", text)
        self.assertIn("live source paths", flat)
        self.assertIn("completed Codex workspace", flat)
        self.assertIn("unchanged and available through `complete-run`", flat)
        self.assertIn("excluded symmetrically from both candidate", flat)
        self.assertIn("comparison limitation", flat)
        self.assertIn("If any file existed before Claude, stop", flat)
        self.assertNotIn("--registered-baseline-project", text)
        self.assertNotIn("--registered-baseline-project-id", text)
        self.assertIn("original Claude directory is never modified", text)

    def test_packaged_engine_preserves_baseline_and_completion_contract(self) -> None:
        prepare_help = subprocess.run(
            [sys.executable, str(RUNNER), "prepare", "--help"],
            cwd=PLUGIN_ROOT,
            check=True,
            capture_output=True,
            text=True,
            timeout=15,
        ).stdout
        complete_help = subprocess.run(
            [sys.executable, str(RUNNER), "complete-run", "--help"],
            cwd=PLUGIN_ROOT,
            check=True,
            capture_output=True,
            text=True,
            timeout=15,
        ).stdout

        for option in (
            "--beginning-kind {git,non_git}",
            "--ending-kind {git,non_git}",
            "--baseline-commit",
            "--ending-commit",
            "--confirm-empty-beginning",
            "--confirm-file-selection",
            "--claude-output-file",
            "--created-by-claude",
            "--exclude-file",
        ):
            with self.subTest(option=option):
                self.assertIn(option, prepare_help)
        self.assertNotIn("--registered-baseline-project", prepare_help)
        self.assertIn("--run-dir", complete_help)
        self.assertIn("--native-result", complete_help)

        runner = RUNNER.read_text(encoding="utf-8")
        self.assertIn('beginning_kind == "git" and ending_kind == "non_git"', runner)
        self.assertIn("A Git beginning state requires a Git end state.", runner)
        self.assertIn("excluded symmetrically from both candidates", runner)
        self.assertIn('commands.add_parser("complete-run")', runner)

        source = MCP_CONFIGURATION.read_text(encoding="utf-8")
        for field in ("beginning_kind", "ending_kind", "baseline_commit", "ending_commit"):
            with self.subTest(mcp_configuration_field=field):
                self.assertIn(f'"{field}"', source)
        self.assertIn("A Git beginning state requires a Git end state.", source)

    def test_agent_prompt_opens_the_available_browser(self) -> None:
        text = AGENT.read_text(encoding="utf-8")
        self.assertIn("independent session on an automatically available loopback port", text)
        self.assertIn("Do not run the workflow in chat", text)
        self.assertNotIn("embedded", text)

    def test_readme_describes_desktop_and_cli_browser_controllers(self) -> None:
        text = " ".join(README.read_text(encoding="utf-8").split())
        self.assertIn("local browser controller", text)
        self.assertIn("Codex in-app browser", text)
        self.assertIn("independent controller on an automatically available loopback port", text)
        self.assertIn("Multiple replay sessions can run in parallel", text)
        self.assertIn("each controller resumes only its own runs", text)
        self.assertIn("Inactive controllers exit automatically", text)
        self.assertIn("codex_replay_early_access", text)
        self.assertNotIn("embedded", text)
        self.assertNotIn("MCP App", text)

    def test_runner_and_assets_are_packaged(self) -> None:
        self.assertTrue(RUNNER.is_file())
        self.assertTrue((PLUGIN_ROOT / "assets" / "model-pricing.json").is_file())
        self.assertTrue((PLUGIN_ROOT / "assets" / "icon.svg").is_file())

    def test_browser_controller_is_packaged_with_direct_http_transport(self) -> None:
        self.assertTrue(MCP_SERVER.is_file())
        self.assertTrue(MCP_CONTROLLER.is_file())
        self.assertTrue(MCP_CONTROLLER_CSS.is_file())
        self.assertLessEqual(
            MCP_CONTROLLER.stat().st_size,
            150_000,
            "Browser controller must satisfy the monorepo file-size limit.",
        )
        self.assertTrue(MCP_WORKER.is_file())
        controller = MCP_CONTROLLER.read_text(encoding="utf-8")
        self.assertIn('href="/controller.css"', controller)
        self.assertIn("/api/call", controller)
        self.assertIn("/api/download", controller)
        self.assertIn("localStorage", controller)
        self.assertNotIn("X-Codex-Replay-Session", controller)
        self.assertNotIn("controller-session", controller)
        self.assertNotIn('"ui/initialize"', controller)
        self.assertNotIn('"tools/call"', controller)
        self.assertNotIn("window.openai", controller)

    def test_removed_state_machine_modules_are_absent(self) -> None:
        for name in (
            "historical_execution_core.py",
            "historical_execution_filesystem.py",
            "historical_execution_review.py",
            "historical_execution_runtime.py",
            "historical_file_attribution.py",
        ):
            self.assertFalse((PLUGIN_ROOT / "scripts" / name).exists(), name)
        self.assertTrue((PLUGIN_ROOT / "scripts" / "historical_file_selection.py").is_file())


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