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 "pyro workspace sync push WORKSPACE_ID ./changes" 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 workspace_help = _subparser_choice(parser, "workspace").format_help() assert "pyro workspace create debian:12 --seed-path ./repo" in workspace_help assert "pyro workspace sync push WORKSPACE_ID ./repo --dest src" in workspace_help assert "pyro workspace exec WORKSPACE_ID" in workspace_help assert "pyro workspace diff WORKSPACE_ID" in workspace_help assert "pyro workspace export WORKSPACE_ID src/note.txt --output ./note.txt" in workspace_help assert "pyro workspace shell open WORKSPACE_ID" in workspace_help workspace_create_help = _subparser_choice( _subparser_choice(parser, "workspace"), "create", ).format_help() assert "--seed-path" in workspace_create_help assert "seed into `/workspace`" in workspace_create_help workspace_exec_help = _subparser_choice( _subparser_choice(parser, "workspace"), "exec", ).format_help() assert "persistent `/workspace`" in workspace_exec_help assert "pyro workspace exec WORKSPACE_ID -- cat note.txt" in workspace_exec_help workspace_sync_help = _subparser_choice( _subparser_choice(parser, "workspace"), "sync", ).format_help() assert "Sync is non-atomic." in workspace_sync_help assert "pyro workspace sync push WORKSPACE_ID ./repo" in workspace_sync_help workspace_sync_push_help = _subparser_choice( _subparser_choice(_subparser_choice(parser, "workspace"), "sync"), "push" ).format_help() assert "--dest" in workspace_sync_push_help assert "Import host content into `/workspace`" in workspace_sync_push_help workspace_export_help = _subparser_choice( _subparser_choice(parser, "workspace"), "export" ).format_help() assert "--output" in workspace_export_help assert "Export one file or directory from `/workspace`" in workspace_export_help workspace_diff_help = _subparser_choice( _subparser_choice(parser, "workspace"), "diff" ).format_help() assert "immutable workspace baseline" in workspace_diff_help assert "workspace export" in workspace_diff_help workspace_shell_help = _subparser_choice( _subparser_choice(parser, "workspace"), "shell", ).format_help() assert "pyro workspace shell open WORKSPACE_ID" in workspace_shell_help assert "Use `workspace exec` for one-shot commands." in workspace_shell_help workspace_service_help = _subparser_choice( _subparser_choice(parser, "workspace"), "service", ).format_help() assert "pyro workspace service start WORKSPACE_ID app" in workspace_service_help assert "Use `--ready-file` by default" in workspace_service_help workspace_service_start_help = _subparser_choice( _subparser_choice(_subparser_choice(parser, "workspace"), "service"), "start" ).format_help() assert "--ready-file" in workspace_service_start_help assert "--ready-tcp" in workspace_service_start_help assert "--ready-http" in workspace_service_start_help assert "--ready-command" in workspace_service_start_help workspace_service_logs_help = _subparser_choice( _subparser_choice(_subparser_choice(parser, "workspace"), "service"), "logs" ).format_help() assert "--tail-lines" in workspace_service_logs_help assert "--all" in workspace_service_logs_help workspace_shell_open_help = _subparser_choice( _subparser_choice(_subparser_choice(parser, "workspace"), "shell"), "open" ).format_help() assert "--cwd" in workspace_shell_open_help assert "--cols" in workspace_shell_open_help assert "--rows" in workspace_shell_open_help workspace_shell_read_help = _subparser_choice( _subparser_choice(_subparser_choice(parser, "workspace"), "shell"), "read" ).format_help() assert "Shell output is written to stdout." in workspace_shell_read_help workspace_shell_write_help = _subparser_choice( _subparser_choice(_subparser_choice(parser, "workspace"), "shell"), "write" ).format_help() assert "--input" in workspace_shell_write_help assert "--no-newline" in workspace_shell_write_help workspace_shell_signal_help = _subparser_choice( _subparser_choice(_subparser_choice(parser, "workspace"), "shell"), "signal" ).format_help() assert "--signal" in workspace_shell_signal_help workspace_shell_close_help = _subparser_choice( _subparser_choice(_subparser_choice(parser, "workspace"), "shell"), "close" ).format_help() assert "Close a persistent workspace shell" in workspace_shell_close_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_cli_requires_command_preserves_shell_argument_boundaries() -> None: command = cli._require_command( ["--", "sh", "-lc", 'printf "hello from workspace\\n" > note.txt'] ) assert command == 'sh -lc \'printf "hello from workspace\\n" > note.txt\'' def test_cli_workspace_create_prints_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: class StubPyro: def create_workspace(self, **kwargs: Any) -> dict[str, Any]: assert kwargs["environment"] == "debian:12" assert kwargs["seed_path"] == "./repo" return {"workspace_id": "workspace-123", "state": "started"} class StubParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="create", environment="debian:12", vcpu_count=1, mem_mib=1024, ttl_seconds=600, network=False, allow_host_compat=False, seed_path="./repo", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: StubParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() output = json.loads(capsys.readouterr().out) assert output["workspace_id"] == "workspace-123" def test_cli_workspace_create_prints_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: class StubPyro: def create_workspace(self, **kwargs: Any) -> dict[str, Any]: del kwargs return { "workspace_id": "workspace-123", "environment": "debian:12", "state": "started", "workspace_path": "/workspace", "workspace_seed": { "mode": "directory", "seed_path": "/tmp/repo", "destination": "/workspace", "entry_count": 1, "bytes_written": 6, }, "execution_mode": "guest_vsock", "vcpu_count": 1, "mem_mib": 1024, "command_count": 0, "last_command": None, } class StubParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="create", environment="debian:12", vcpu_count=1, mem_mib=1024, ttl_seconds=600, network=False, allow_host_compat=False, seed_path="/tmp/repo", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: StubParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() output = capsys.readouterr().out assert "Workspace ID: workspace-123" in output assert "Workspace: /workspace" in output assert "Workspace seed: directory from /tmp/repo" in output def test_cli_workspace_exec_prints_human_output( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def exec_workspace( self, workspace_id: str, *, command: str, timeout_seconds: int, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert command == "cat note.txt" assert timeout_seconds == 30 return { "workspace_id": workspace_id, "sequence": 2, "cwd": "/workspace", "execution_mode": "guest_vsock", "exit_code": 0, "duration_ms": 4, "stdout": "hello\n", "stderr": "", } class StubParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="exec", workspace_id="workspace-123", timeout_seconds=30, json=False, command_args=["--", "cat", "note.txt"], ) monkeypatch.setattr(cli, "_build_parser", lambda: StubParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() captured = capsys.readouterr() assert captured.out == "hello\n" assert ( "[workspace-exec] workspace_id=workspace-123 sequence=2 cwd=/workspace" in captured.err ) def test_cli_workspace_export_prints_human_output( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def export_workspace( self, workspace_id: str, path: str, *, output_path: str, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert path == "note.txt" assert output_path == "./note.txt" return { "workspace_id": workspace_id, "workspace_path": "/workspace/note.txt", "output_path": "/tmp/note.txt", "artifact_type": "file", "entry_count": 1, "bytes_written": 6, "execution_mode": "guest_vsock", } class StubParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="export", workspace_id="workspace-123", path="note.txt", output="./note.txt", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: StubParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() output = capsys.readouterr().out assert "[workspace-export] workspace_id=workspace-123" in output assert "artifact_type=file" in output def test_cli_workspace_diff_prints_human_output( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def diff_workspace(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return { "workspace_id": workspace_id, "changed": True, "summary": { "total": 1, "added": 0, "modified": 1, "deleted": 0, "type_changed": 0, "text_patched": 1, "non_text": 0, }, "entries": [ { "path": "note.txt", "status": "modified", "artifact_type": "file", "text_patch": "--- a/note.txt\n+++ b/note.txt\n", } ], "patch": "--- a/note.txt\n+++ b/note.txt\n", } class StubParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="diff", workspace_id="workspace-123", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: StubParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() output = capsys.readouterr().out assert ( "[workspace-diff] workspace_id=workspace-123 total=1 added=0 modified=1" in output ) assert "--- a/note.txt" in output def test_cli_workspace_sync_push_prints_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: class StubPyro: def push_workspace_sync( self, workspace_id: str, source_path: str, *, dest: str, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert source_path == "./repo" assert dest == "src" return { "workspace_id": workspace_id, "execution_mode": "guest_vsock", "workspace_sync": { "mode": "directory", "source_path": "/tmp/repo", "destination": "/workspace/src", "entry_count": 2, "bytes_written": 12, }, } class StubParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="sync", workspace_sync_command="push", workspace_id="workspace-123", source_path="./repo", dest="src", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: StubParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() output = json.loads(capsys.readouterr().out) assert output["workspace_sync"]["destination"] == "/workspace/src" def test_cli_workspace_sync_push_prints_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def push_workspace_sync( self, workspace_id: str, source_path: str, *, dest: str, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert source_path == "./repo" assert dest == "/workspace" return { "workspace_id": workspace_id, "execution_mode": "guest_vsock", "workspace_sync": { "mode": "directory", "source_path": "/tmp/repo", "destination": "/workspace", "entry_count": 2, "bytes_written": 12, }, } class StubParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="sync", workspace_sync_command="push", workspace_id="workspace-123", source_path="./repo", dest="/workspace", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: StubParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() output = capsys.readouterr().out assert "[workspace-sync] workspace_id=workspace-123 mode=directory source=/tmp/repo" in output assert ( "destination=/workspace entry_count=2 bytes_written=12 " "execution_mode=guest_vsock" ) in output def test_cli_workspace_logs_and_delete_print_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def logs_workspace(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return { "workspace_id": workspace_id, "count": 1, "entries": [ { "sequence": 1, "exit_code": 0, "duration_ms": 2, "cwd": "/workspace", "command": "printf 'ok\\n'", "stdout": "ok\n", "stderr": "", } ], } def delete_workspace(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return {"workspace_id": workspace_id, "deleted": True} class LogsParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="logs", workspace_id="workspace-123", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: LogsParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() class DeleteParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="delete", workspace_id="workspace-123", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: DeleteParser()) cli.main() output = capsys.readouterr().out assert "#1 exit_code=0 duration_ms=2 cwd=/workspace" in output assert "Deleted workspace: workspace-123" in output def test_cli_workspace_status_and_delete_print_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def status_workspace(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return {"workspace_id": workspace_id, "state": "started"} def delete_workspace(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return {"workspace_id": workspace_id, "deleted": True} class StatusParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="status", workspace_id="workspace-123", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: StatusParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() status = json.loads(capsys.readouterr().out) assert status["state"] == "started" class DeleteParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="delete", workspace_id="workspace-123", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: DeleteParser()) cli.main() deleted = json.loads(capsys.readouterr().out) assert deleted["deleted"] is True def test_cli_workspace_status_prints_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def status_workspace(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return { "workspace_id": workspace_id, "environment": "debian:12", "state": "started", "workspace_path": "/workspace", "execution_mode": "guest_vsock", "vcpu_count": 1, "mem_mib": 1024, "command_count": 0, "last_command": None, "service_count": 1, "running_service_count": 1, } class StatusParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="status", workspace_id="workspace-123", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: StatusParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() output = capsys.readouterr().out assert "Workspace ID: workspace-123" in output assert "Services: 1/1" in output def test_cli_workspace_logs_prints_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def logs_workspace(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return {"workspace_id": workspace_id, "count": 0, "entries": []} class LogsParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="logs", workspace_id="workspace-123", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: LogsParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() payload = json.loads(capsys.readouterr().out) assert payload["count"] == 0 def test_cli_workspace_delete_prints_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def delete_workspace(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return {"workspace_id": workspace_id, "deleted": True} class DeleteParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="delete", workspace_id="workspace-123", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: DeleteParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() assert "Deleted workspace: workspace-123" in capsys.readouterr().out def test_cli_workspace_exec_prints_json_and_exits_nonzero( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def exec_workspace( self, workspace_id: str, *, command: str, timeout_seconds: int ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert command == "false" assert timeout_seconds == 30 return { "workspace_id": workspace_id, "sequence": 1, "cwd": "/workspace", "execution_mode": "guest_vsock", "exit_code": 2, "duration_ms": 5, "stdout": "", "stderr": "boom\n", } class ExecParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="exec", workspace_id="workspace-123", timeout_seconds=30, json=True, command_args=["--", "false"], ) monkeypatch.setattr(cli, "_build_parser", lambda: ExecParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) with pytest.raises(SystemExit, match="2"): cli.main() payload = json.loads(capsys.readouterr().out) assert payload["exit_code"] == 2 def test_cli_workspace_exec_prints_human_error( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def exec_workspace( self, workspace_id: str, *, command: str, timeout_seconds: int ) -> dict[str, Any]: del workspace_id, command, timeout_seconds raise RuntimeError("exec boom") class ExecParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="exec", workspace_id="workspace-123", timeout_seconds=30, json=False, command_args=["--", "cat", "note.txt"], ) monkeypatch.setattr(cli, "_build_parser", lambda: ExecParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) with pytest.raises(SystemExit, match="1"): cli.main() assert "[error] exec boom" in capsys.readouterr().err def test_cli_workspace_export_and_diff_print_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def export_workspace( self, workspace_id: str, path: str, *, output_path: str ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert path == "note.txt" assert output_path == "./note.txt" return { "workspace_id": workspace_id, "workspace_path": "/workspace/note.txt", "output_path": "/tmp/note.txt", "artifact_type": "file", "entry_count": 1, "bytes_written": 6, "execution_mode": "guest_vsock", } def diff_workspace(self, workspace_id: str) -> dict[str, Any]: return { "workspace_id": workspace_id, "changed": False, "summary": { "total": 0, "added": 0, "modified": 0, "deleted": 0, "type_changed": 0, "text_patched": 0, "non_text": 0, }, "entries": [], "patch": "", } class ExportParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="export", workspace_id="workspace-123", path="note.txt", output="./note.txt", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: ExportParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() export_payload = json.loads(capsys.readouterr().out) assert export_payload["artifact_type"] == "file" class DiffParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="diff", workspace_id="workspace-123", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: DiffParser()) cli.main() diff_payload = json.loads(capsys.readouterr().out) assert diff_payload["changed"] is False @pytest.mark.parametrize( ("command_name", "json_mode", "method_name"), [ ("list", True, "list_services"), ("list", False, "list_services"), ("status", True, "status_service"), ("status", False, "status_service"), ("logs", True, "logs_service"), ("logs", False, "logs_service"), ("stop", True, "stop_service"), ("stop", False, "stop_service"), ], ) def test_cli_workspace_service_error_paths( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], command_name: str, json_mode: bool, method_name: str, ) -> None: class StubPyro: def list_services(self, workspace_id: str) -> dict[str, Any]: del workspace_id raise RuntimeError("service branch boom") def status_service(self, workspace_id: str, service_name: str) -> dict[str, Any]: del workspace_id, service_name raise RuntimeError("service branch boom") def logs_service( self, workspace_id: str, service_name: str, *, tail_lines: int | None, all: bool, ) -> dict[str, Any]: del workspace_id, service_name, tail_lines, all raise RuntimeError("service branch boom") def stop_service(self, workspace_id: str, service_name: str) -> dict[str, Any]: del workspace_id, service_name raise RuntimeError("service branch boom") class Parser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command=command_name, workspace_id="workspace-123", service_name="app", tail_lines=50, all=False, json=json_mode, ) monkeypatch.setattr(cli, "_build_parser", lambda: Parser()) monkeypatch.setattr(cli, "Pyro", StubPyro) with pytest.raises(SystemExit, match="1"): cli.main() captured = capsys.readouterr() if json_mode: payload = json.loads(captured.out) assert payload["error"] == "service branch boom" else: assert "[error] service branch boom" in captured.err assert hasattr(StubPyro, method_name) def test_cli_workspace_shell_open_and_read_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def open_shell( self, workspace_id: str, *, cwd: str, cols: int, rows: int, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert cwd == "/workspace" assert cols == 120 assert rows == 30 return { "workspace_id": workspace_id, "shell_id": "shell-123", "state": "running", "cwd": cwd, "cols": cols, "rows": rows, "started_at": 1.0, "ended_at": None, "exit_code": None, "execution_mode": "guest_vsock", } def read_shell( self, workspace_id: str, shell_id: str, *, cursor: int, max_chars: int, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert shell_id == "shell-123" assert cursor == 0 assert max_chars == 1024 return { "workspace_id": workspace_id, "shell_id": shell_id, "state": "running", "cwd": "/workspace", "cols": 120, "rows": 30, "started_at": 1.0, "ended_at": None, "exit_code": None, "execution_mode": "guest_vsock", "cursor": 0, "next_cursor": 14, "output": "pyro$ pwd\n", "truncated": False, } class OpenParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="open", workspace_id="workspace-123", cwd="/workspace", cols=120, rows=30, json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: OpenParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() class ReadParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="read", workspace_id="workspace-123", shell_id="shell-123", cursor=0, max_chars=1024, json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: ReadParser()) cli.main() captured = capsys.readouterr() assert "pyro$ pwd\n" in captured.out assert "[workspace-shell-open] workspace_id=workspace-123 shell_id=shell-123" in captured.err assert "[workspace-shell-read] workspace_id=workspace-123 shell_id=shell-123" in captured.err def test_cli_workspace_shell_write_signal_close_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def write_shell( self, workspace_id: str, shell_id: str, *, input: str, append_newline: bool, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert shell_id == "shell-123" assert input == "pwd" assert append_newline is False return { "workspace_id": workspace_id, "shell_id": shell_id, "state": "running", "cwd": "/workspace", "cols": 120, "rows": 30, "started_at": 1.0, "ended_at": None, "exit_code": None, "execution_mode": "guest_vsock", "input_length": 3, "append_newline": False, } def signal_shell( self, workspace_id: str, shell_id: str, *, signal_name: str, ) -> dict[str, Any]: assert signal_name == "INT" return { "workspace_id": workspace_id, "shell_id": shell_id, "state": "running", "cwd": "/workspace", "cols": 120, "rows": 30, "started_at": 1.0, "ended_at": None, "exit_code": None, "execution_mode": "guest_vsock", "signal": signal_name, } def close_shell(self, workspace_id: str, shell_id: str) -> dict[str, Any]: return { "workspace_id": workspace_id, "shell_id": shell_id, "state": "stopped", "cwd": "/workspace", "cols": 120, "rows": 30, "started_at": 1.0, "ended_at": 2.0, "exit_code": 0, "execution_mode": "guest_vsock", "closed": True, } class WriteParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="write", workspace_id="workspace-123", shell_id="shell-123", input="pwd", no_newline=True, json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: WriteParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() written = json.loads(capsys.readouterr().out) assert written["append_newline"] is False class SignalParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="signal", workspace_id="workspace-123", shell_id="shell-123", signal="INT", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: SignalParser()) cli.main() signaled = json.loads(capsys.readouterr().out) assert signaled["signal"] == "INT" class CloseParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="close", workspace_id="workspace-123", shell_id="shell-123", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: CloseParser()) cli.main() closed = json.loads(capsys.readouterr().out) assert closed["closed"] is True def test_cli_workspace_shell_open_and_read_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def open_shell( self, workspace_id: str, *, cwd: str, cols: int, rows: int, ) -> dict[str, Any]: return { "workspace_id": workspace_id, "shell_id": "shell-123", "state": "running", "cwd": cwd, "cols": cols, "rows": rows, "started_at": 1.0, "ended_at": None, "exit_code": None, "execution_mode": "guest_vsock", } def read_shell( self, workspace_id: str, shell_id: str, *, cursor: int, max_chars: int, ) -> dict[str, Any]: return { "workspace_id": workspace_id, "shell_id": shell_id, "state": "running", "cwd": "/workspace", "cols": 120, "rows": 30, "started_at": 1.0, "ended_at": None, "exit_code": None, "execution_mode": "guest_vsock", "cursor": cursor, "next_cursor": max_chars, "output": "pyro$ pwd\n", "truncated": False, } class OpenParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="open", workspace_id="workspace-123", cwd="/workspace", cols=120, rows=30, json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: OpenParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() opened = json.loads(capsys.readouterr().out) assert opened["shell_id"] == "shell-123" class ReadParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="read", workspace_id="workspace-123", shell_id="shell-123", cursor=0, max_chars=1024, json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: ReadParser()) cli.main() payload = json.loads(capsys.readouterr().out) assert payload["output"] == "pyro$ pwd\n" def test_cli_workspace_shell_write_signal_close_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def write_shell( self, workspace_id: str, shell_id: str, *, input: str, append_newline: bool, ) -> dict[str, Any]: del input, append_newline return { "workspace_id": workspace_id, "shell_id": shell_id, "state": "running", "cwd": "/workspace", "cols": 120, "rows": 30, "started_at": 1.0, "ended_at": None, "exit_code": None, "execution_mode": "guest_vsock", "input_length": 3, "append_newline": True, } def signal_shell( self, workspace_id: str, shell_id: str, *, signal_name: str, ) -> dict[str, Any]: return { "workspace_id": workspace_id, "shell_id": shell_id, "state": "running", "cwd": "/workspace", "cols": 120, "rows": 30, "started_at": 1.0, "ended_at": None, "exit_code": None, "execution_mode": "guest_vsock", "signal": signal_name, } def close_shell(self, workspace_id: str, shell_id: str) -> dict[str, Any]: return { "workspace_id": workspace_id, "shell_id": shell_id, "state": "stopped", "cwd": "/workspace", "cols": 120, "rows": 30, "started_at": 1.0, "ended_at": 2.0, "exit_code": 0, "execution_mode": "guest_vsock", "closed": True, } class WriteParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="write", workspace_id="workspace-123", shell_id="shell-123", input="pwd", no_newline=False, json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: WriteParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() class SignalParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="signal", workspace_id="workspace-123", shell_id="shell-123", signal="INT", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: SignalParser()) cli.main() class CloseParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command="close", workspace_id="workspace-123", shell_id="shell-123", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: CloseParser()) cli.main() captured = capsys.readouterr() assert "[workspace-shell-write]" in captured.err assert "[workspace-shell-signal]" in captured.err assert "[workspace-shell-close]" in captured.err @pytest.mark.parametrize( ("shell_command", "kwargs"), [ ("open", {"cwd": "/workspace", "cols": 120, "rows": 30}), ("read", {"shell_id": "shell-123", "cursor": 0, "max_chars": 1024}), ("write", {"shell_id": "shell-123", "input": "pwd", "no_newline": False}), ("signal", {"shell_id": "shell-123", "signal": "INT"}), ("close", {"shell_id": "shell-123"}), ], ) def test_cli_workspace_shell_error_paths( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], shell_command: str, kwargs: dict[str, Any], ) -> None: class StubPyro: def open_shell(self, *args: Any, **inner_kwargs: Any) -> dict[str, Any]: del args, inner_kwargs raise RuntimeError("shell boom") def read_shell(self, *args: Any, **inner_kwargs: Any) -> dict[str, Any]: del args, inner_kwargs raise RuntimeError("shell boom") def write_shell(self, *args: Any, **inner_kwargs: Any) -> dict[str, Any]: del args, inner_kwargs raise RuntimeError("shell boom") def signal_shell(self, *args: Any, **inner_kwargs: Any) -> dict[str, Any]: del args, inner_kwargs raise RuntimeError("shell boom") def close_shell(self, *args: Any, **inner_kwargs: Any) -> dict[str, Any]: del args, inner_kwargs raise RuntimeError("shell boom") class Parser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="shell", workspace_shell_command=shell_command, workspace_id="workspace-123", json=False, **kwargs, ) monkeypatch.setattr(cli, "_build_parser", lambda: Parser()) monkeypatch.setattr(cli, "Pyro", StubPyro) with pytest.raises(SystemExit, match="1"): cli.main() assert "[error] shell boom" in capsys.readouterr().err def test_cli_workspace_service_start_prints_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def start_service( self, workspace_id: str, service_name: str, **kwargs: Any ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert service_name == "app" assert kwargs["command"] == "sh -lc 'touch .ready && while true; do sleep 60; done'" assert kwargs["readiness"] == {"type": "file", "path": ".ready"} return { "workspace_id": workspace_id, "service_name": service_name, "state": "running", "cwd": "/workspace", "execution_mode": "guest_vsock", } class StartParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="start", workspace_id="workspace-123", service_name="app", cwd="/workspace", ready_file=".ready", ready_tcp=None, ready_http=None, ready_command=None, ready_timeout_seconds=30, ready_interval_ms=500, json=True, command_args=["--", "sh", "-lc", "touch .ready && while true; do sleep 60; done"], ) monkeypatch.setattr(cli, "_build_parser", lambda: StartParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() payload = json.loads(capsys.readouterr().out) assert payload["state"] == "running" def test_cli_workspace_service_logs_prints_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def logs_service( self, workspace_id: str, service_name: str, *, tail_lines: int, all: bool, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert service_name == "app" assert tail_lines == 200 assert all is False return { "workspace_id": workspace_id, "service_name": service_name, "state": "running", "cwd": "/workspace", "execution_mode": "guest_vsock", "stdout": "ready\n", "stderr": "", "tail_lines": 200, "truncated": False, } class LogsParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="logs", workspace_id="workspace-123", service_name="app", tail_lines=200, all=False, json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: LogsParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() captured = capsys.readouterr() assert captured.out == "ready\n" assert "service_name=app" in captured.err def test_cli_workspace_service_list_prints_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def list_services(self, workspace_id: str) -> dict[str, Any]: assert workspace_id == "workspace-123" return { "workspace_id": workspace_id, "count": 2, "running_count": 1, "services": [ { "workspace_id": workspace_id, "service_name": "app", "state": "running", "cwd": "/workspace", "execution_mode": "guest_vsock", "readiness": {"type": "file", "path": "/workspace/.ready"}, }, { "workspace_id": workspace_id, "service_name": "worker", "state": "stopped", "cwd": "/workspace", "execution_mode": "guest_vsock", "readiness": None, }, ], } class ListParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="list", workspace_id="workspace-123", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: ListParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() captured = capsys.readouterr() assert "app [running] cwd=/workspace" in captured.out assert "worker [stopped] cwd=/workspace" in captured.out def test_cli_workspace_service_status_prints_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def status_service(self, workspace_id: str, service_name: str) -> dict[str, Any]: assert workspace_id == "workspace-123" assert service_name == "app" return { "workspace_id": workspace_id, "service_name": service_name, "state": "running", "cwd": "/workspace", "execution_mode": "guest_vsock", "readiness": {"type": "file", "path": "/workspace/.ready"}, } class StatusParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="status", workspace_id="workspace-123", service_name="app", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: StatusParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() payload = json.loads(capsys.readouterr().out) assert payload["state"] == "running" def test_cli_workspace_service_stop_prints_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def stop_service(self, workspace_id: str, service_name: str) -> dict[str, Any]: assert workspace_id == "workspace-123" assert service_name == "app" return { "workspace_id": workspace_id, "service_name": service_name, "state": "stopped", "cwd": "/workspace", "execution_mode": "guest_vsock", "stop_reason": "sigterm", } class StopParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="stop", workspace_id="workspace-123", service_name="app", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: StopParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() captured = capsys.readouterr() assert "service_name=app" in captured.err assert "state=stopped" in captured.err def test_cli_workspace_service_start_rejects_multiple_readiness_flags( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def start_service(self, *args: Any, **kwargs: Any) -> dict[str, Any]: raise AssertionError("start_service should not be called") class StartParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="start", workspace_id="workspace-123", service_name="app", cwd="/workspace", ready_file=".ready", ready_tcp="127.0.0.1:8080", ready_http=None, ready_command=None, ready_timeout_seconds=30, ready_interval_ms=500, json=False, command_args=["--", "sh", "-lc", "touch .ready && while true; do sleep 60; done"], ) monkeypatch.setattr(cli, "_build_parser", lambda: StartParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) with pytest.raises(SystemExit, match="1"): cli.main() captured = capsys.readouterr() assert "choose at most one" in captured.err def test_cli_workspace_service_start_prints_human_with_ready_command( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def start_service( self, workspace_id: str, service_name: str, **kwargs: Any ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert service_name == "app" assert kwargs["readiness"] == {"type": "command", "command": "test -f .ready"} return { "workspace_id": workspace_id, "service_name": service_name, "state": "running", "cwd": "/workspace", "execution_mode": "guest_vsock", } class StartParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="start", workspace_id="workspace-123", service_name="app", cwd="/workspace", ready_file=None, ready_tcp=None, ready_http=None, ready_command="test -f .ready", ready_timeout_seconds=30, ready_interval_ms=500, json=False, command_args=["--", "sh", "-lc", "touch .ready && while true; do sleep 60; done"], ) monkeypatch.setattr(cli, "_build_parser", lambda: StartParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() captured = capsys.readouterr() assert "service_name=app" in captured.err assert "state=running" in captured.err def test_cli_workspace_service_start_prints_json_error( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def start_service( self, workspace_id: str, service_name: str, **kwargs: Any ) -> dict[str, Any]: del workspace_id, service_name, kwargs raise RuntimeError("service boom") class StartParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="start", workspace_id="workspace-123", service_name="app", cwd="/workspace", ready_file=None, ready_tcp="127.0.0.1:8080", ready_http=None, ready_command=None, ready_timeout_seconds=30, ready_interval_ms=500, json=True, command_args=["--", "sh", "-lc", "while true; do sleep 60; done"], ) monkeypatch.setattr(cli, "_build_parser", lambda: StartParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) with pytest.raises(SystemExit, match="1"): cli.main() payload = json.loads(capsys.readouterr().out) assert payload["error"] == "service boom" def test_cli_workspace_service_list_prints_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def list_services(self, workspace_id: str) -> dict[str, Any]: return {"workspace_id": workspace_id, "count": 0, "running_count": 0, "services": []} class ListParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="list", workspace_id="workspace-123", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: ListParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() payload = json.loads(capsys.readouterr().out) assert payload["count"] == 0 def test_cli_workspace_service_status_prints_human( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def status_service(self, workspace_id: str, service_name: str) -> dict[str, Any]: del workspace_id return { "workspace_id": "workspace-123", "service_name": service_name, "state": "running", "cwd": "/workspace", "execution_mode": "guest_vsock", "readiness": {"type": "tcp", "address": "127.0.0.1:8080"}, } class StatusParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="status", workspace_id="workspace-123", service_name="app", json=False, ) monkeypatch.setattr(cli, "_build_parser", lambda: StatusParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() captured = capsys.readouterr() assert "service_name=app" in captured.err assert "state=running" in captured.err def test_cli_workspace_service_logs_prints_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def logs_service( self, workspace_id: str, service_name: str, *, tail_lines: int | None, all: bool, ) -> dict[str, Any]: assert workspace_id == "workspace-123" assert service_name == "app" assert tail_lines is None assert all is True return { "workspace_id": workspace_id, "service_name": service_name, "state": "running", "cwd": "/workspace", "execution_mode": "guest_vsock", "stdout": "ready\n", "stderr": "", "tail_lines": None, "truncated": False, } class LogsParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="logs", workspace_id="workspace-123", service_name="app", tail_lines=None, all=True, json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: LogsParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() payload = json.loads(capsys.readouterr().out) assert payload["tail_lines"] is None def test_cli_workspace_service_stop_prints_json( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: class StubPyro: def stop_service(self, workspace_id: str, service_name: str) -> dict[str, Any]: return { "workspace_id": workspace_id, "service_name": service_name, "state": "stopped", "cwd": "/workspace", "execution_mode": "guest_vsock", } class StopParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="service", workspace_service_command="stop", workspace_id="workspace-123", service_name="app", json=True, ) monkeypatch.setattr(cli, "_build_parser", lambda: StopParser()) monkeypatch.setattr(cli, "Pyro", StubPyro) cli.main() payload = json.loads(capsys.readouterr().out) assert payload["state"] == "stopped" def test_cli_workspace_exec_json_error_exits_nonzero( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: class StubPyro: def exec_workspace( self, workspace_id: str, *, command: str, timeout_seconds: int, ) -> dict[str, Any]: del workspace_id, command, timeout_seconds raise RuntimeError("workspace is unavailable") class StubParser: def parse_args(self) -> argparse.Namespace: return argparse.Namespace( command="workspace", workspace_command="exec", workspace_id="workspace-123", timeout_seconds=30, json=True, command_args=["--", "true"], ) 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_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