525 lines
18 KiB
Python
525 lines
18 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 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 run_in_vm(self, **kwargs: Any) -> dict[str, Any]:
|
|
assert kwargs["vcpu_count"] == 1
|
|
assert kwargs["mem_mib"] == 1024
|
|
return {
|
|
"environment": kwargs["environment"],
|
|
"execution_mode": "guest_vsock",
|
|
"exit_code": 0,
|
|
"duration_ms": 12,
|
|
"stdout": "hi\n",
|
|
"stderr": "",
|
|
}
|
|
|
|
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] 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 run_in_vm(self, **kwargs: Any) -> dict[str, Any]:
|
|
del kwargs
|
|
return {
|
|
"environment": "debian:12",
|
|
"execution_mode": "guest_vsock",
|
|
"exit_code": 7,
|
|
"duration_ms": 5,
|
|
"stdout": "",
|
|
"stderr": "bad\n",
|
|
}
|
|
|
|
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_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
|