pyro-mcp/tests/test_cli.py
Thales Maciel 6e16e74fd5 Harden default environment pull behavior
Fix the default one-shot install path so empty bundled profile directories no longer win over OCI-backed environment pulls or leave broken cached symlinks behind.

Treat cached installs as valid only when the manifest and boot artifacts are all present, repair invalid installs on the next pull, and add human-mode phase markers for env pull and run without changing JSON output.

Align the Python lifecycle example and public docs with the current exec_vm/vm_exec auto-clean semantics, and validate the slice with focused pytest coverage, make check, make dist-check, and a real default-path pull/inspect/run smoke.
2026-03-11 19:27:09 -03:00

601 lines
21 KiB
Python

from __future__ import annotations
import argparse
import json
import sys
from typing import Any, cast
import pytest
import pyro_mcp.cli as cli
def _subparser_choice(parser: argparse.ArgumentParser, name: str) -> argparse.ArgumentParser:
subparsers = getattr(parser, "_subparsers", None)
if subparsers is None:
raise AssertionError("parser does not define subparsers")
group_actions = cast(list[Any], subparsers._group_actions) # noqa: SLF001
if not group_actions:
raise AssertionError("parser subparsers are empty")
choices = cast(dict[str, argparse.ArgumentParser], group_actions[0].choices)
return choices[name]
def test_cli_help_guides_first_run() -> None:
parser = cli._build_parser()
help_text = parser.format_help()
assert "Suggested first run:" in help_text
assert "pyro doctor" in help_text
assert "pyro env list" in help_text
assert "pyro env pull debian:12" in help_text
assert "pyro run debian:12 -- git --version" in help_text
assert "Use `pyro mcp serve` only after the CLI validation path works." in help_text
def test_cli_subcommand_help_includes_examples_and_guidance() -> None:
parser = cli._build_parser()
run_help = _subparser_choice(parser, "run").format_help()
assert "pyro run debian:12 -- git --version" in run_help
assert "Opt into host-side compatibility execution" in run_help
assert "Enable outbound guest networking" in run_help
assert "may appear in either order" in run_help
assert "Use --json for a deterministic" in run_help
env_help = _subparser_choice(_subparser_choice(parser, "env"), "pull").format_help()
assert "Environment name from `pyro env list`" in env_help
assert "pyro env pull debian:12" in env_help
assert "downloads from public Docker Hub" in env_help
doctor_help = _subparser_choice(parser, "doctor").format_help()
assert "Check host prerequisites and embedded runtime health" in doctor_help
assert "pyro doctor --json" in doctor_help
demo_help = _subparser_choice(parser, "demo").format_help()
assert "pyro demo ollama --verbose" in demo_help
mcp_help = _subparser_choice(_subparser_choice(parser, "mcp"), "serve").format_help()
assert "Expose pyro tools over stdio for an MCP client." in mcp_help
assert "Use this from an MCP client config after the CLI evaluation path works." in mcp_help
def test_cli_run_prints_json(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def run_in_vm(self, **kwargs: Any) -> dict[str, Any]:
assert kwargs["network"] is True
assert kwargs["command"] == "echo hi"
return {"exit_code": 0, "stdout": "hi\n"}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="run",
environment="debian:12",
vcpu_count=1,
mem_mib=512,
timeout_seconds=30,
ttl_seconds=600,
network=True,
allow_host_compat=False,
json=True,
command_args=["--", "echo", "hi"],
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
output = json.loads(capsys.readouterr().out)
assert output["exit_code"] == 0
def test_cli_doctor_prints_json(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(command="doctor", platform="linux-x86_64", json=True)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(
cli,
"doctor_report",
lambda platform: {"platform": platform, "runtime_ok": True},
)
cli.main()
output = json.loads(capsys.readouterr().out)
assert output["runtime_ok"] is True
def test_cli_demo_ollama_prints_summary(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="demo",
demo_command="ollama",
base_url="http://localhost:11434/v1",
model="llama3.2:3b",
verbose=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(
cli,
"run_ollama_tool_demo",
lambda **kwargs: {
"exec_result": {"exit_code": 0, "execution_mode": "guest_vsock", "stdout": "true\n"},
"fallback_used": False,
},
)
cli.main()
output = capsys.readouterr().out
assert "[summary] exit_code=0 fallback_used=False execution_mode=guest_vsock" in output
def test_cli_env_list_prints_json(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubPyro:
def list_environments(self) -> list[dict[str, object]]:
return [{"name": "debian:12", "installed": False}]
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(command="env", env_command="list", json=True)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
output = json.loads(capsys.readouterr().out)
assert output["environments"][0]["name"] == "debian:12"
def test_cli_run_prints_human_output(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def create_vm(self, **kwargs: Any) -> dict[str, Any]:
assert kwargs["vcpu_count"] == 1
assert kwargs["mem_mib"] == 1024
return {"vm_id": "vm-123"}
def start_vm(self, vm_id: str) -> dict[str, Any]:
assert vm_id == "vm-123"
return {"vm_id": vm_id, "state": "started"}
def exec_vm(self, vm_id: str, *, command: str, timeout_seconds: int) -> dict[str, Any]:
assert vm_id == "vm-123"
assert command == "echo hi"
assert timeout_seconds == 30
return {
"environment": "debian:12",
"execution_mode": "guest_vsock",
"exit_code": 0,
"duration_ms": 12,
"stdout": "hi\n",
"stderr": "",
}
@property
def manager(self) -> Any:
raise AssertionError("manager cleanup should not be used on a successful run")
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="run",
environment="debian:12",
vcpu_count=1,
mem_mib=1024,
timeout_seconds=30,
ttl_seconds=600,
network=False,
allow_host_compat=False,
json=False,
command_args=["--", "echo", "hi"],
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
captured = capsys.readouterr()
assert captured.out == "hi\n"
assert "[run] phase=create environment=debian:12" in captured.err
assert "[run] phase=start vm_id=vm-123" in captured.err
assert "[run] phase=execute vm_id=vm-123" in captured.err
assert "[run] environment=debian:12 execution_mode=guest_vsock exit_code=0" in captured.err
def test_cli_run_exits_with_command_status(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def create_vm(self, **kwargs: Any) -> dict[str, Any]:
del kwargs
return {"vm_id": "vm-456"}
def start_vm(self, vm_id: str) -> dict[str, Any]:
assert vm_id == "vm-456"
return {"vm_id": vm_id, "state": "started"}
def exec_vm(self, vm_id: str, *, command: str, timeout_seconds: int) -> dict[str, Any]:
assert vm_id == "vm-456"
assert command == "false"
assert timeout_seconds == 30
return {
"environment": "debian:12",
"execution_mode": "guest_vsock",
"exit_code": 7,
"duration_ms": 5,
"stdout": "",
"stderr": "bad\n",
}
@property
def manager(self) -> Any:
raise AssertionError("manager cleanup should not be used when exec_vm returns normally")
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="run",
environment="debian:12",
vcpu_count=1,
mem_mib=1024,
timeout_seconds=30,
ttl_seconds=600,
network=False,
allow_host_compat=False,
json=False,
command_args=["--", "false"],
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
with pytest.raises(SystemExit, match="7"):
cli.main()
captured = capsys.readouterr()
assert "bad\n" in captured.err
def test_cli_env_pull_prints_human_progress(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def pull_environment(self, environment: str) -> dict[str, Any]:
assert environment == "debian:12"
return {
"name": "debian:12",
"version": "1.0.0",
"distribution": "debian",
"distribution_version": "12",
"installed": True,
"cache_dir": "/tmp/cache",
"default_packages": ["bash", "git"],
"install_dir": "/tmp/cache/linux-x86_64/debian_12-1.0.0",
"install_manifest": "/tmp/cache/linux-x86_64/debian_12-1.0.0/environment.json",
"kernel_image": "/tmp/cache/linux-x86_64/debian_12-1.0.0/vmlinux",
"rootfs_image": "/tmp/cache/linux-x86_64/debian_12-1.0.0/rootfs.ext4",
"oci_registry": "registry-1.docker.io",
"oci_repository": "thalesmaciel/pyro-environment-debian-12",
"oci_reference": "1.0.0",
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="env",
env_command="pull",
environment="debian:12",
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
captured = capsys.readouterr()
assert "[pull] phase=install environment=debian:12" in captured.err
assert "[pull] phase=ready environment=debian:12" in captured.err
assert "Pulled: debian:12" in captured.out
def test_cli_requires_run_command() -> None:
with pytest.raises(ValueError, match="command is required"):
cli._require_command([])
def test_print_env_helpers_render_human_output(capsys: pytest.CaptureFixture[str]) -> None:
cli._print_env_list_human(
{
"catalog_version": "2.0.0",
"environments": [
{"name": "debian:12", "installed": True, "description": "Git environment"},
"ignored",
],
}
)
cli._print_env_detail_human(
{
"name": "debian:12",
"version": "1.0.0",
"distribution": "debian",
"distribution_version": "12",
"installed": True,
"cache_dir": "/cache",
"default_packages": ["bash", "git"],
"description": "Git environment",
"install_dir": "/cache/linux-x86_64/debian_12-1.0.0",
"install_manifest": "/cache/linux-x86_64/debian_12-1.0.0/environment.json",
"kernel_image": "/cache/vmlinux",
"rootfs_image": "/cache/rootfs.ext4",
"oci_registry": "registry-1.docker.io",
"oci_repository": "thalesmaciel/pyro-environment-debian-12",
"oci_reference": "1.0.0",
},
action="Environment",
)
cli._print_prune_human({"count": 2, "deleted_environment_dirs": ["a", "b"]})
cli._print_doctor_human(
{
"platform": "linux-x86_64",
"runtime_ok": False,
"issues": ["broken"],
"kvm": {"exists": True, "readable": True, "writable": False},
"runtime": {
"cache_dir": "/cache",
"capabilities": {
"supports_vm_boot": True,
"supports_guest_exec": False,
"supports_guest_network": True,
},
},
"networking": {"tun_available": True, "ip_forward_enabled": False},
}
)
captured = capsys.readouterr().out
assert "Catalog version: 2.0.0" in captured
assert "debian:12 [installed] Git environment" in captured
assert "Install manifest: /cache/linux-x86_64/debian_12-1.0.0/environment.json" in captured
assert "Deleted 2 cached environment entries." in captured
assert "Runtime: FAIL" in captured
assert "Issues:" in captured
def test_print_env_list_human_handles_empty(capsys: pytest.CaptureFixture[str]) -> None:
cli._print_env_list_human({"catalog_version": "2.0.0", "environments": []})
output = capsys.readouterr().out
assert "No environments found." in output
def test_write_stream_skips_empty(capsys: pytest.CaptureFixture[str]) -> None:
cli._write_stream("", stream=sys.stdout)
cli._write_stream("x", stream=sys.stdout)
captured = capsys.readouterr()
assert captured.out == "x"
def test_cli_env_pull_prints_human(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubPyro:
def pull_environment(self, environment: str) -> dict[str, object]:
assert environment == "debian:12"
return {
"name": "debian:12",
"version": "1.0.0",
"distribution": "debian",
"distribution_version": "12",
"installed": True,
"cache_dir": "/cache",
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="env",
env_command="pull",
environment="debian:12",
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
output = capsys.readouterr().out
assert "Pulled: debian:12" in output
def test_cli_env_inspect_and_prune_print_human(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubPyro:
def inspect_environment(self, environment: str) -> dict[str, object]:
assert environment == "debian:12"
return {
"name": "debian:12",
"version": "1.0.0",
"distribution": "debian",
"distribution_version": "12",
"installed": False,
"cache_dir": "/cache",
}
def prune_environments(self) -> dict[str, object]:
return {"count": 1, "deleted_environment_dirs": ["stale"]}
class InspectParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="env",
env_command="inspect",
environment="debian:12",
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: InspectParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
class PruneParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(command="env", env_command="prune", json=False)
monkeypatch.setattr(cli, "_build_parser", lambda: PruneParser())
cli.main()
output = capsys.readouterr().out
assert "Environment: debian:12" in output
assert "Deleted 1 cached environment entry." in output
def test_cli_doctor_prints_human(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(command="doctor", platform="linux-x86_64", json=False)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(
cli,
"doctor_report",
lambda platform: {
"platform": platform,
"runtime_ok": True,
"issues": [],
"kvm": {"exists": True, "readable": True, "writable": True},
},
)
cli.main()
output = capsys.readouterr().out
assert "Runtime: PASS" in output
def test_cli_run_json_error_exits_nonzero(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubPyro:
def run_in_vm(self, **kwargs: Any) -> dict[str, Any]:
del kwargs
raise RuntimeError("guest boot is unavailable")
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="run",
environment="debian:12",
vcpu_count=1,
mem_mib=1024,
timeout_seconds=30,
ttl_seconds=600,
network=False,
allow_host_compat=False,
json=True,
command_args=["--", "echo", "hi"],
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
with pytest.raises(SystemExit, match="1"):
cli.main()
payload = json.loads(capsys.readouterr().out)
assert payload["ok"] is False
def test_cli_mcp_runs_stdio_transport(monkeypatch: pytest.MonkeyPatch) -> None:
observed: dict[str, str] = {}
class StubPyro:
def create_server(self) -> Any:
return type(
"StubServer",
(),
{"run": staticmethod(lambda transport: observed.update({"transport": transport}))},
)()
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(command="mcp", mcp_command="serve")
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
assert observed == {"transport": "stdio"}
def test_cli_demo_default_prints_json(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(command="demo", demo_command=None, network=False)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "run_demo", lambda network: {"exit_code": 0, "network": network})
cli.main()
output = json.loads(capsys.readouterr().out)
assert output["exit_code"] == 0
def test_cli_demo_ollama_verbose_and_error_paths(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class VerboseParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="demo",
demo_command="ollama",
base_url="http://localhost:11434/v1",
model="llama3.2:3b",
verbose=True,
)
monkeypatch.setattr(cli, "_build_parser", lambda: VerboseParser())
monkeypatch.setattr(
cli,
"run_ollama_tool_demo",
lambda **kwargs: {
"exec_result": {"exit_code": 0, "execution_mode": "guest_vsock", "stdout": "true\n"},
"fallback_used": False,
},
)
cli.main()
output = capsys.readouterr().out
assert "[summary] stdout=true" in output
class ErrorParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="demo",
demo_command="ollama",
base_url="http://localhost:11434/v1",
model="llama3.2:3b",
verbose=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: ErrorParser())
monkeypatch.setattr(
cli,
"run_ollama_tool_demo",
lambda **kwargs: (_ for _ in ()).throw(RuntimeError("tool loop failed")),
)
with pytest.raises(SystemExit, match="1"):
cli.main()
assert "[error] tool loop failed" in capsys.readouterr().out