tests/test_cli/test_wrap_openclaw.py

"""Tests for `headroom wrap openclaw` command."""

from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest
from click.testing import CliRunner

from headroom.cli import wrap as wrap_cli
from headroom.cli.main import main


@pytest.fixture
def runner() -> CliRunner:
    return CliRunner()


@pytest.fixture
def plugin_dir(tmp_path: Path) -> Path:
    """Create a minimal OpenClaw plugin directory fixture."""
    plugin = tmp_path / "plugins" / "openclaw"
    plugin.mkdir(parents=True)
    (plugin / "package.json").write_text('{"name":"headroom-openclaw"}\n')
    (plugin / "openclaw.plugin.json").write_text('{"id":"headroom"}\n')
    return plugin


def _make_successful_run(calls: list[dict]) -> object:
    def run(cmd, **kwargs):  # noqa: ANN001
        calls.append({"cmd": list(cmd), **kwargs})
        return MagicMock(returncode=0, stdout="", stderr="")

    return run


def test_wrap_openclaw_default_installs_from_npm_and_restarts(runner: CliRunner) -> None:
    calls: list[dict] = []

    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": "npm",
        }
        return mapping.get(name)

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        with patch("headroom.cli.wrap.subprocess.run", side_effect=_make_successful_run(calls)):
            result = runner.invoke(main, ["wrap", "openclaw"])

    assert result.exit_code == 0, result.output

    cmds = [c["cmd"] for c in calls]
    assert [
        "openclaw",
        "plugins",
        "install",
        "--dangerously-force-unsafe-install",
        "headroom-ai/openclaw",
    ] in cmds
    assert ["openclaw", "config", "validate"] in cmds
    assert ["openclaw", "gateway", "restart"] in cmds
    assert ["openclaw", "plugins", "inspect", "headroom"] in cmds

    # Verify plugin install in npm mode does not set cwd
    install_call = next(
        c
        for c in calls
        if c["cmd"][:4] == ["openclaw", "plugins", "install", "--dangerously-force-unsafe-install"]
    )
    assert install_call["cwd"] is None

    # No local build in npm mode
    assert ["npm", "install"] not in cmds
    assert ["npm", "run", "build"] not in cmds

    # Verify config payload includes enabled + expected defaults
    set_entry = next(
        c
        for c in calls
        if c["cmd"][:4] == ["openclaw", "config", "set", "plugins.entries.headroom"]
    )
    payload = json.loads(set_entry["cmd"][4])
    assert payload["enabled"] is True
    assert payload["config"]["proxyPort"] == 8787
    assert payload["config"]["autoStart"] is True
    assert payload["config"]["startupTimeoutMs"] == 20000


def test_wrap_openclaw_skip_build_and_no_restart(runner: CliRunner, plugin_dir: Path) -> None:
    calls: list[dict] = []

    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": "npm",
        }
        return mapping.get(name)

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        with patch("headroom.cli.wrap.subprocess.run", side_effect=_make_successful_run(calls)):
            result = runner.invoke(
                main,
                [
                    "wrap",
                    "openclaw",
                    "--plugin-path",
                    str(plugin_dir),
                    "--skip-build",
                    "--no-restart",
                ],
            )

    assert result.exit_code == 0, result.output
    cmds = [c["cmd"] for c in calls]
    assert ["npm", "install"] not in cmds
    assert ["npm", "run", "build"] not in cmds
    assert ["openclaw", "gateway", "restart"] not in cmds


def test_wrap_openclaw_local_source_mode_builds_and_links(
    runner: CliRunner, plugin_dir: Path
) -> None:
    calls: list[dict] = []

    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": "npm",
        }
        return mapping.get(name)

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        with patch("headroom.cli.wrap.subprocess.run", side_effect=_make_successful_run(calls)):
            result = runner.invoke(
                main,
                ["wrap", "openclaw", "--plugin-path", str(plugin_dir)],
            )

    assert result.exit_code == 0, result.output
    cmds = [c["cmd"] for c in calls]
    assert ["npm", "install"] in cmds
    assert ["npm", "run", "build"] in cmds
    assert [
        "openclaw",
        "plugins",
        "install",
        "--dangerously-force-unsafe-install",
        "--link",
        ".",
    ] in cmds


def test_wrap_openclaw_fails_when_openclaw_missing(runner: CliRunner, plugin_dir: Path) -> None:
    def which(name: str) -> str | None:
        return None if name == "openclaw" else "npm"

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        result = runner.invoke(main, ["wrap", "openclaw", "--plugin-path", str(plugin_dir)])

    assert result.exit_code != 0
    assert "'openclaw' not found in PATH" in result.output


def test_wrap_openclaw_fails_when_plugin_path_invalid(runner: CliRunner, tmp_path: Path) -> None:
    invalid = tmp_path / "missing-plugin"

    with patch("headroom.cli.wrap.shutil.which", return_value="openclaw"):
        result = runner.invoke(main, ["wrap", "openclaw", "--plugin-path", str(invalid)])

    assert result.exit_code != 0
    assert "Plugin path not found" in result.output


def test_wrap_openclaw_uses_extension_fallback_on_linked_install_bug(
    runner: CliRunner, plugin_dir: Path
) -> None:
    calls: list[dict] = []

    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": "npm",
        }
        return mapping.get(name)

    def run(cmd, **kwargs):  # noqa: ANN001
        calls.append({"cmd": list(cmd), **kwargs})
        if cmd[:3] == ["openclaw", "plugins", "install"]:
            return MagicMock(
                returncode=1,
                stdout="Also not a valid hook pack",
                stderr='Plugin installation blocked despite "--dangerously-force-unsafe-install"',
            )
        return MagicMock(returncode=0, stdout="", stderr="")

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        with patch("headroom.cli.wrap.subprocess.run", side_effect=run):
            with patch(
                "headroom.cli.wrap._copy_openclaw_plugin_into_extensions",
                return_value=Path("C:/Users/test/.openclaw/extensions/headroom"),
            ) as copy_fallback:
                result = runner.invoke(main, ["wrap", "openclaw", "--plugin-path", str(plugin_dir)])

    assert result.exit_code == 0, result.output
    copy_fallback.assert_called_once()


def test_wrap_openclaw_continues_when_plugin_already_exists(
    runner: CliRunner,
) -> None:
    calls: list[dict] = []

    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": "npm",
        }
        return mapping.get(name)

    def run(cmd, **kwargs):  # noqa: ANN001
        calls.append({"cmd": list(cmd), **kwargs})
        if cmd[:3] == ["openclaw", "plugins", "install"]:
            return MagicMock(
                returncode=1,
                stdout="plugin already exists: C:\\Users\\test\\.openclaw\\extensions\\headroom",
                stderr="",
            )
        return MagicMock(returncode=0, stdout="", stderr="")

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        with patch("headroom.cli.wrap.subprocess.run", side_effect=run):
            result = runner.invoke(main, ["wrap", "openclaw", "--no-restart"])

    assert result.exit_code == 0, result.output
    cmds = [c["cmd"] for c in calls]
    assert ["openclaw", "config", "validate"] in cmds
    assert ["openclaw", "plugins", "inspect", "headroom"] in cmds


def test_wrap_openclaw_verbose_prints_install_restart_and_inspect_output(
    runner: CliRunner,
) -> None:
    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": "npm",
        }
        return mapping.get(name)

    def run(cmd, **kwargs):  # noqa: ANN001
        if cmd[:3] == ["openclaw", "plugins", "install"]:
            return MagicMock(returncode=0, stdout="install-ok", stderr="")
        if cmd[:3] == ["openclaw", "gateway", "restart"]:
            return MagicMock(returncode=0, stdout="restart-ok", stderr="")
        if cmd[:3] == ["openclaw", "plugins", "inspect"]:
            return MagicMock(returncode=0, stdout="inspect-ok", stderr="")
        return MagicMock(returncode=0, stdout="", stderr="")

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        with patch("headroom.cli.wrap.subprocess.run", side_effect=run):
            result = runner.invoke(main, ["wrap", "openclaw", "--verbose"])

    assert result.exit_code == 0, result.output
    assert "install-ok" in result.output
    assert "restart-ok" in result.output
    assert "inspect-ok" in result.output


def test_wrap_openclaw_fails_for_npm_mode_hook_pack_bug_without_local_fallback(
    runner: CliRunner,
) -> None:
    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": "npm",
        }
        return mapping.get(name)

    def run(cmd, **kwargs):  # noqa: ANN001
        if cmd[:3] == ["openclaw", "plugins", "install"]:
            return MagicMock(
                returncode=1,
                stdout="Also not a valid hook pack",
                stderr='Blocked despite "--dangerously-force-unsafe-install"',
            )
        return MagicMock(returncode=0, stdout="", stderr="")

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        with patch("headroom.cli.wrap.subprocess.run", side_effect=run):
            result = runner.invoke(main, ["wrap", "openclaw"])

    assert result.exit_code != 0
    assert "openclaw plugins install failed" in result.output


def test_wrap_openclaw_copy_mode_uses_path_install(runner: CliRunner, plugin_dir: Path) -> None:
    calls: list[dict] = []

    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": "npm",
        }
        return mapping.get(name)

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        with patch("headroom.cli.wrap.subprocess.run", side_effect=_make_successful_run(calls)):
            result = runner.invoke(
                main,
                [
                    "wrap",
                    "openclaw",
                    "--plugin-path",
                    str(plugin_dir),
                    "--copy",
                    "--skip-build",
                    "--no-restart",
                ],
            )

    assert result.exit_code == 0, result.output
    cmds = [c["cmd"] for c in calls]
    assert [
        "openclaw",
        "plugins",
        "install",
        "--dangerously-force-unsafe-install",
        str(plugin_dir),
    ] in cmds


def test_wrap_openclaw_fails_when_npm_missing_for_local_build(
    runner: CliRunner, plugin_dir: Path
) -> None:
    def which(name: str) -> str | None:
        mapping = {
            "openclaw": "openclaw",
            "npm": None,
        }
        return mapping.get(name)

    with patch("headroom.cli.wrap.shutil.which", side_effect=which):
        result = runner.invoke(main, ["wrap", "openclaw", "--plugin-path", str(plugin_dir)])

    assert result.exit_code != 0
    assert "'npm' not found in PATH" in result.output


def test_wrap_openclaw_fails_when_local_path_missing_manifest_files(
    runner: CliRunner, tmp_path: Path
) -> None:
    plugin = tmp_path / "plugins" / "openclaw"
    plugin.mkdir(parents=True)

    with patch("headroom.cli.wrap.shutil.which", return_value="openclaw"):
        result = runner.invoke(main, ["wrap", "openclaw", "--plugin-path", str(plugin)])
    assert result.exit_code != 0
    assert "missing package.json" in result.output

    (plugin / "package.json").write_text("{}\n")
    with patch("headroom.cli.wrap.shutil.which", return_value="openclaw"):
        result = runner.invoke(main, ["wrap", "openclaw", "--plugin-path", str(plugin)])
    assert result.exit_code != 0
    assert "missing openclaw.plugin.json" in result.output


def test_run_checked_raises_click_exception_on_command_errors() -> None:
    with patch("headroom.cli.wrap.subprocess.run", side_effect=FileNotFoundError()):
        with pytest.raises(Exception, match="command not found"):
            wrap_cli._run_checked(["missing"], action="demo")

    cpe_stderr = wrap_cli.subprocess.CalledProcessError(
        returncode=2,
        cmd=["x"],
        stderr="bad-stderr",
    )
    with patch("headroom.cli.wrap.subprocess.run", side_effect=cpe_stderr):
        with pytest.raises(Exception, match="bad-stderr"):
            wrap_cli._run_checked(["x"], action="demo")

    cpe_stdout = wrap_cli.subprocess.CalledProcessError(
        returncode=3,
        cmd=["x"],
        output="bad-stdout",
        stderr="",
    )
    with patch("headroom.cli.wrap.subprocess.run", side_effect=cpe_stdout):
        with pytest.raises(Exception, match="bad-stdout"):
            wrap_cli._run_checked(["x"], action="demo")


def test_resolve_openclaw_extensions_dir_empty_output_raises() -> None:
    with patch(
        "headroom.cli.wrap._run_checked",
        return_value=MagicMock(stdout="   \n", stderr="", returncode=0),
    ):
        with pytest.raises(Exception, match="Unable to resolve OpenClaw config path"):
            wrap_cli._resolve_openclaw_extensions_dir("openclaw")


def test_copy_openclaw_plugin_into_extensions_handles_missing_and_existing_dist(
    tmp_path: Path,
) -> None:
    plugin = tmp_path / "plugin"
    plugin.mkdir()

    with pytest.raises(Exception, match="Plugin dist folder missing"):
        wrap_cli._copy_openclaw_plugin_into_extensions(plugin_dir=plugin, openclaw_bin="openclaw")

    dist = plugin / "dist"
    dist.mkdir()
    (dist / "index.js").write_text("x\n")
    (plugin / "package.json").write_text("{}\n")
    (plugin / "openclaw.plugin.json").write_text("{}\n")

    ext_root = tmp_path / ".openclaw" / "extensions"
    target_headroom = ext_root / "headroom"
    target_dist = target_headroom / "dist"
    target_dist.mkdir(parents=True)
    (target_dist / "old.js").write_text("old\n")

    with patch("headroom.cli.wrap._resolve_openclaw_extensions_dir", return_value=ext_root):
        out = wrap_cli._copy_openclaw_plugin_into_extensions(
            plugin_dir=plugin, openclaw_bin="openclaw"
        )

    assert out == target_headroom
    assert (target_dist / "index.js").exists()
    assert not (target_dist / "old.js").exists()