agentic-mobile-control/apps/device-host-agent/tests/test_cli.py

from __future__ import annotations

import sys

import pytest

from host_agent import cli
from host_agent.instance_lock import InstanceAlreadyRunningError
from host_agent.local_account import LocalAccountStore


class _RecordingApplication:
    def __init__(self) -> None:
        self.ran = False

    def run(self) -> None:
        self.ran = True


def _base_env(tmp_path) -> dict[str, str]:
    return {
        "HOST_AGENT_CONTROL_PLANE_URL": "https://cloud.example",
        "HOST_AGENT_LOCAL_ACCOUNT_PATH": str(tmp_path / "account.json"),
        "HOST_AGENT_IDENTITY_PATH": str(tmp_path / "identity.json"),
    }


def _set_env(monkeypatch, tmp_path) -> None:
    for key, value in _base_env(tmp_path).items():
        monkeypatch.setenv(key, value)


def _patch_create_application(monkeypatch) -> dict:
    captured: dict = {}

    def fake_create_application(*, config=None, **kwargs):
        captured["config"] = config
        app = _RecordingApplication()
        captured["app"] = app
        return app

    monkeypatch.setattr(cli, "create_application", fake_create_application)
    return captured


def test_existing_account_fast_path_skips_prompting(monkeypatch, tmp_path) -> None:
    _set_env(monkeypatch, tmp_path)
    LocalAccountStore(tmp_path / "account.json").create(
        "operator", "correct horse battery staple"
    )

    def fail_input(prompt: str = "") -> str:
        raise AssertionError("must not prompt when a local account already exists")

    monkeypatch.setattr("builtins.input", fail_input)
    captured = _patch_create_application(monkeypatch)

    cli.main([])

    assert captured["app"].ran is True
    assert captured["config"].display_name == "operator"


def test_interactive_first_run_prompts_and_creates_account(
    monkeypatch, tmp_path
) -> None:
    _set_env(monkeypatch, tmp_path)
    monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
    inputs = iter(["operator"])
    monkeypatch.setattr("builtins.input", lambda prompt="": next(inputs))
    passwords = iter(["hunter2", "hunter2"])
    monkeypatch.setattr("getpass.getpass", lambda prompt="": next(passwords))
    captured = _patch_create_application(monkeypatch)

    cli.main([])

    assert captured["app"].ran is True
    account = LocalAccountStore(tmp_path / "account.json").load()
    assert account is not None
    assert account.username == "operator"


def test_non_interactive_without_account_exits_with_clear_error(
    monkeypatch, tmp_path, capsys
) -> None:
    _set_env(monkeypatch, tmp_path)
    monkeypatch.setattr(sys.stdin, "isatty", lambda: False)
    captured = _patch_create_application(monkeypatch)

    with pytest.raises(SystemExit) as exc_info:
        cli.main([])

    assert exc_info.value.code == 1
    assert "setup" in capsys.readouterr().err
    assert "app" not in captured
    assert LocalAccountStore(tmp_path / "account.json").load() is None


def test_setup_subcommand_creates_account(monkeypatch, tmp_path) -> None:
    _set_env(monkeypatch, tmp_path)
    inputs = iter(["operator"])
    monkeypatch.setattr("builtins.input", lambda prompt="": next(inputs))
    passwords = iter(["hunter2", "hunter2"])
    monkeypatch.setattr("getpass.getpass", lambda prompt="": next(passwords))
    captured = _patch_create_application(monkeypatch)

    cli.main(["setup"])

    assert "app" not in captured
    account = LocalAccountStore(tmp_path / "account.json").load()
    assert account is not None
    assert account.username == "operator"


def test_setup_subcommand_refuses_overwrite_without_confirmation(
    monkeypatch, tmp_path
) -> None:
    _set_env(monkeypatch, tmp_path)
    store = LocalAccountStore(tmp_path / "account.json")
    original = store.create("operator", "original-password")
    monkeypatch.setattr("builtins.input", lambda prompt="": "n")

    cli.main(["setup"])

    assert store.load() == original


def test_duplicate_instance_exits_with_clear_error(
    monkeypatch, tmp_path, capsys
) -> None:
    _set_env(monkeypatch, tmp_path)
    LocalAccountStore(tmp_path / "account.json").create(
        "operator", "correct horse battery staple"
    )
    lock_path = tmp_path / "identity_dir" / "host_agent.lock"

    def raising_create_application(*, config=None, **kwargs):
        raise InstanceAlreadyRunningError(lock_path)

    monkeypatch.setattr(cli, "create_application", raising_create_application)

    with pytest.raises(SystemExit) as exc_info:
        cli.main([])

    assert exc_info.value.code == 1
    err = capsys.readouterr().err
    assert "another Host Agent instance" in err
    assert str(lock_path) in err


def test_mcp_token_subcommand_prints_token(tmp_path, capsys, monkeypatch) -> None:
    monkeypatch.setenv("HOST_AGENT_IDENTITY_PATH", str(tmp_path / "host_identity.json"))
    monkeypatch.setenv(
        "HOST_AGENT_LOCAL_ACCOUNT_PATH", str(tmp_path / "host_local_account.json")
    )
    # Also set control plane URL to satisfy config loading
    monkeypatch.setenv("HOST_AGENT_CONTROL_PLANE_URL", "https://cloud.example")
    from host_agent.cli import main

    main(["mcp-token"])
    out = capsys.readouterr().out.strip()
    assert len(out) >= 40  # token is ~43 chars
    # Subsequent invocation prints the same token (idempotent).
    main(["mcp-token"])
    out2 = capsys.readouterr().out.strip()
    assert out == out2