thaloco/pyro-mcp
1
Fork 0
Code Pull requests Releases Packages Activity Actions
pyro-mcp/tests/test_cli.py
Thales Maciel 68d8e875e0 Add host-specific MCP onramps for major chat clients 
Ship first-class MCP setup examples for Claude Code, Codex, and OpenCode so new users can copy one exact command or config instead of translating the generic MCP template by hand.

Reposition the docs to surface those host-specific examples before the generic config fallback, keep workspace-core as the recommended profile everywhere user-facing, and retain Claude Desktop/Cursor as secondary fallback examples.

Bump the package and catalog to 3.11.0, mark the roadmap milestone done, and add docs-alignment coverage that pins the new examples to the canonical workspace-core server command and the expected OpenCode config shape.

Validation:
- uv lock
- ./.venv/bin/pytest --no-cov tests/test_cli.py
- UV_CACHE_DIR=.uv-cache make check
- UV_CACHE_DIR=.uv-cache make dist-check
2026-03-13 13:42:45 -03:00

4417 lines
151 KiB
Python
Raw Blame History

from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
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 "Continue into the stable workspace path after that:" in help_text
assert "pyro workspace exec WORKSPACE_ID -- cat note.txt" in help_text
assert "pyro workspace snapshot create WORKSPACE_ID checkpoint" in help_text
assert "pyro workspace reset WORKSPACE_ID --snapshot checkpoint" 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 "--profile" in mcp_help
assert "workspace-core" in mcp_help
assert "workspace-full" in mcp_help
assert "vm-run" in mcp_help
assert "recommended first profile for most chat hosts" in mcp_help
assert "default in 3.x for compatibility" in mcp_help
workspace_help = _subparser_choice(parser, "workspace").format_help()
assert "stable workspace contract" in workspace_help
assert "pyro workspace create debian:12 --seed-path ./repo" in workspace_help
assert "--id-only" in workspace_help
assert "pyro workspace create debian:12 --name repro-fix --label issue=123" in workspace_help
assert "pyro workspace list" in workspace_help
assert (
"pyro workspace update WORKSPACE_ID --name retry-run --label owner=codex"
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 stop WORKSPACE_ID" in workspace_help
assert "pyro workspace disk list WORKSPACE_ID" in workspace_help
assert "pyro workspace disk export WORKSPACE_ID --output ./workspace.ext4" in workspace_help
assert "pyro workspace start WORKSPACE_ID" in workspace_help
assert "pyro workspace snapshot create WORKSPACE_ID checkpoint" in workspace_help
assert "pyro workspace reset WORKSPACE_ID --snapshot checkpoint" in workspace_help
assert "pyro workspace shell open WORKSPACE_ID --id-only" in workspace_help
workspace_create_help = _subparser_choice(
_subparser_choice(parser, "workspace"),
"create",
).format_help()
assert "--id-only" in workspace_create_help
assert "--name" in workspace_create_help
assert "--label" in workspace_create_help
assert "--seed-path" in workspace_create_help
assert "--secret" in workspace_create_help
assert "--secret-file" 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 "--secret-env" in workspace_exec_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_list_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "list"
).format_help()
assert "List persisted workspaces" in workspace_list_help
workspace_update_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "update"
).format_help()
assert "--name" in workspace_update_help
assert "--clear-name" in workspace_update_help
assert "--label" in workspace_update_help
assert "--clear-label" in workspace_update_help
workspace_file_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "file"
).format_help()
assert "model-native tree inspection and text edits" in workspace_file_help
assert "pyro workspace file read WORKSPACE_ID src/app.py" in workspace_file_help
workspace_file_list_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "list"
).format_help()
assert "--recursive" in workspace_file_list_help
workspace_file_read_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "read"
).format_help()
assert "--max-bytes" in workspace_file_read_help
assert "--content-only" in workspace_file_read_help
workspace_file_write_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "file"), "write"
).format_help()
assert "--text" in workspace_file_write_help
assert "--text-file" in workspace_file_write_help
workspace_patch_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "patch"
).format_help()
assert "Apply add/modify/delete unified text patches" in workspace_patch_help
workspace_patch_apply_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "patch"), "apply"
).format_help()
assert "--patch" in workspace_patch_apply_help
assert "--patch-file" in workspace_patch_apply_help
workspace_stop_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "stop"
).format_help()
assert "Stop the backing sandbox" in workspace_stop_help
workspace_start_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "start"
).format_help()
assert "previously stopped workspace" in workspace_start_help
workspace_disk_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "disk"
).format_help()
assert "secondary stopped-workspace disk tools" in workspace_disk_help
assert "pyro workspace disk read WORKSPACE_ID note.txt" in workspace_disk_help
workspace_disk_export_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "disk"), "export"
).format_help()
assert "--output" in workspace_disk_export_help
workspace_disk_list_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "disk"), "list"
).format_help()
assert "--recursive" in workspace_disk_list_help
workspace_disk_read_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "disk"), "read"
).format_help()
assert "--max-bytes" in workspace_disk_read_help
assert "--content-only" in workspace_disk_read_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_snapshot_help = _subparser_choice(
_subparser_choice(parser, "workspace"),
"snapshot",
).format_help()
assert "pyro workspace snapshot create WORKSPACE_ID checkpoint" in workspace_snapshot_help
assert "baseline" in workspace_snapshot_help
workspace_snapshot_create_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "snapshot"), "create"
).format_help()
assert "Capture the current `/workspace` tree" in workspace_snapshot_create_help
workspace_snapshot_list_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "snapshot"), "list"
).format_help()
assert "baseline snapshot plus any named snapshots" in workspace_snapshot_list_help
workspace_snapshot_delete_help = _subparser_choice(
_subparser_choice(_subparser_choice(parser, "workspace"), "snapshot"), "delete"
).format_help()
assert "leaving the implicit baseline intact" in workspace_snapshot_delete_help
workspace_reset_help = _subparser_choice(
_subparser_choice(parser, "workspace"), "reset"
).format_help()
assert "--snapshot" in workspace_reset_help
assert "reset over repair" in workspace_reset_help
workspace_shell_help = _subparser_choice(
_subparser_choice(parser, "workspace"),
"shell",
).format_help()
assert "pyro workspace shell open WORKSPACE_ID --id-only" 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
assert "--secret-env" 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 "--id-only" in workspace_shell_open_help
assert "--cwd" in workspace_shell_open_help
assert "--cols" in workspace_shell_open_help
assert "--rows" in workspace_shell_open_help
assert "--secret-env" 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
assert "--plain" in workspace_shell_read_help
assert "--wait-for-idle-ms" 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_read_utf8_text_file_rejects_non_utf8(tmp_path: Path) -> None:
source_path = tmp_path / "bad.txt"
source_path.write_bytes(b"\xff\xfe")
with pytest.raises(ValueError, match="must contain UTF-8 text"):
cli._read_utf8_text_file(str(source_path), option_name="--text-file")
def test_cli_read_utf8_text_file_rejects_empty_path() -> None:
with pytest.raises(ValueError, match="must not be empty"):
cli._read_utf8_text_file("", option_name="--patch-file")
def test_cli_shortcut_flags_are_mutually_exclusive() -> None:
parser = cli._build_parser()
with pytest.raises(SystemExit):
parser.parse_args(
[
"workspace",
"create",
"debian:12",
"--json",
"--id-only",
]
)
with pytest.raises(SystemExit):
parser.parse_args(
[
"workspace",
"shell",
"open",
"workspace-123",
"--json",
"--id-only",
]
)
with pytest.raises(SystemExit):
parser.parse_args(
[
"workspace",
"file",
"write",
"workspace-123",
"src/app.py",
"--text",
"hello",
"--text-file",
"./app.py",
]
)
with pytest.raises(SystemExit):
parser.parse_args(
[
"workspace",
"patch",
"apply",
"workspace-123",
"--patch",
"--- a/app.py\n+++ b/app.py\n",
"--patch-file",
"./fix.patch",
]
)
with pytest.raises(SystemExit):
parser.parse_args(
[
"workspace",
"file",
"read",
"workspace-123",
"note.txt",
"--json",
"--content-only",
]
)
with pytest.raises(SystemExit):
parser.parse_args(
[
"workspace",
"disk",
"read",
"workspace-123",
"note.txt",
"--json",
"--content-only",
]
)
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"
assert kwargs["network_policy"] == "egress"
assert kwargs["name"] == "repro-fix"
assert kwargs["labels"] == {"issue": "123"}
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_policy="egress",
allow_host_compat=False,
seed_path="./repo",
name="repro-fix",
label=["issue=123"],
secret=[],
secret_file=[],
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_id_only(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
class StubPyro:
def create_workspace(self, **kwargs: Any) -> dict[str, Any]:
assert kwargs["environment"] == "debian:12"
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_policy="off",
allow_host_compat=False,
seed_path=None,
name=None,
label=[],
secret=[],
secret_file=[],
json=False,
id_only=True,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
captured = capsys.readouterr()
assert captured.out == "workspace-123\n"
assert captured.err == ""
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",
"name": "repro-fix",
"labels": {"issue": "123"},
"environment": "debian:12",
"state": "started",
"network_policy": "off",
"workspace_path": "/workspace",
"last_activity_at": 123.0,
"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_policy="off",
allow_host_compat=False,
seed_path="/tmp/repo",
name="repro-fix",
label=["issue=123"],
secret=[],
secret_file=[],
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 "Name: repro-fix" in output
assert "Labels: issue=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,
secret_env: dict[str, str] | None = None,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert command == "cat note.txt"
assert timeout_seconds == 30
assert secret_env is None
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,
secret_env=[],
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_print_workspace_summary_human_handles_last_command_and_secret_filtering(
capsys: pytest.CaptureFixture[str],
) -> None:
cli._print_workspace_summary_human(
{
"workspace_id": "workspace-123",
"labels": {"owner": "codex"},
"environment": "debian:12",
"state": "started",
"workspace_path": "/workspace",
"last_activity_at": 123.0,
"network_policy": "off",
"workspace_seed": {"mode": "directory", "seed_path": "/tmp/repo"},
"secrets": ["ignored", {"name": "API_TOKEN", "source_kind": "literal"}],
"execution_mode": "guest_vsock",
"vcpu_count": 1,
"mem_mib": 1024,
"command_count": 2,
"reset_count": 0,
"service_count": 1,
"running_service_count": 1,
"last_command": {"command": "pytest", "exit_code": 0},
},
action="Workspace",
)
output = capsys.readouterr().out
assert "Secrets: API_TOKEN (literal)" in output
assert "Last command: pytest (exit_code=0)" in output
def test_cli_workspace_list_prints_human(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def list_workspaces(self) -> dict[str, Any]:
return {
"count": 1,
"workspaces": [
{
"workspace_id": "workspace-123",
"name": "repro-fix",
"labels": {"issue": "123", "owner": "codex"},
"environment": "debian:12",
"state": "started",
"created_at": 100.0,
"last_activity_at": 200.0,
"expires_at": 700.0,
"command_count": 2,
"service_count": 1,
"running_service_count": 1,
}
],
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="list",
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 "name='repro-fix'" in output
assert "labels=issue=123,owner=codex" in output
def test_print_workspace_list_human_skips_non_dict_entries(
capsys: pytest.CaptureFixture[str],
) -> None:
cli._print_workspace_list_human(
{
"workspaces": [
"ignored",
{
"workspace_id": "workspace-123",
"state": "started",
"environment": "debian:12",
"last_activity_at": 200.0,
"expires_at": 700.0,
"command_count": 2,
"service_count": 1,
"running_service_count": 1,
},
]
}
)
output = capsys.readouterr().out
assert "workspace_id=workspace-123" in output
assert "ignored" not in output
def test_cli_workspace_list_prints_empty_state(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def list_workspaces(self) -> dict[str, Any]:
return {"count": 0, "workspaces": []}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="list",
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
assert capsys.readouterr().out.strip() == "No workspaces."
def test_cli_workspace_update_prints_json(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def update_workspace(self, workspace_id: str, **kwargs: Any) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert kwargs["name"] == "retry-run"
assert kwargs["clear_name"] is False
assert kwargs["labels"] == {"issue": "124", "owner": "codex"}
assert kwargs["clear_labels"] == ["stale"]
return {
"workspace_id": workspace_id,
"name": "retry-run",
"labels": {"issue": "124", "owner": "codex"},
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="update",
workspace_id="workspace-123",
name="retry-run",
clear_name=False,
label=["issue=124", "owner=codex"],
clear_label=["stale"],
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_update_prints_human(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def update_workspace(self, workspace_id: str, **kwargs: Any) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert kwargs["name"] is None
assert kwargs["clear_name"] is True
assert kwargs["labels"] == {"owner": "codex"}
assert kwargs["clear_labels"] is None
return {
"workspace_id": workspace_id,
"name": None,
"labels": {"owner": "codex"},
"environment": "debian:12",
"state": "started",
"workspace_path": "/workspace",
"last_activity_at": 123.0,
"network_policy": "off",
"workspace_seed": {"mode": "empty", "seed_path": None},
"execution_mode": "guest_vsock",
"vcpu_count": 1,
"mem_mib": 1024,
"command_count": 0,
"reset_count": 0,
"service_count": 0,
"running_service_count": 0,
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="update",
workspace_id="workspace-123",
name=None,
clear_name=True,
label=["owner=codex"],
clear_label=[],
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 "Labels: owner=codex" in output
assert "Last activity at: 123.0" in output
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_file_commands_print_human_and_json(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def list_workspace_files(
self,
workspace_id: str,
*,
path: str,
recursive: bool,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "/workspace/src"
assert recursive is True
return {
"workspace_id": workspace_id,
"path": path,
"recursive": recursive,
"entries": [
{
"path": "/workspace/src/app.py",
"artifact_type": "file",
"size_bytes": 14,
"link_target": None,
}
],
"execution_mode": "guest_vsock",
}
def read_workspace_file(
self,
workspace_id: str,
path: str,
*,
max_bytes: int,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "src/app.py"
assert max_bytes == 4096
return {
"workspace_id": workspace_id,
"path": "/workspace/src/app.py",
"size_bytes": 14,
"max_bytes": max_bytes,
"content": "print('hi')\n",
"truncated": False,
"execution_mode": "guest_vsock",
}
def write_workspace_file(
self,
workspace_id: str,
path: str,
*,
text: str,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "src/app.py"
assert text == "print('hello')\n"
return {
"workspace_id": workspace_id,
"path": "/workspace/src/app.py",
"size_bytes": len(text.encode("utf-8")),
"bytes_written": len(text.encode("utf-8")),
"execution_mode": "guest_vsock",
}
def apply_workspace_patch(
self,
workspace_id: str,
*,
patch: str,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert patch.startswith("--- a/src/app.py")
return {
"workspace_id": workspace_id,
"changed": True,
"summary": {"total": 1, "added": 0, "modified": 1, "deleted": 0},
"entries": [{"path": "/workspace/src/app.py", "status": "modified"}],
"patch": patch,
"execution_mode": "guest_vsock",
}
class ListParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="list",
workspace_id="workspace-123",
path="/workspace/src",
recursive=True,
json=False,
)
class ReadParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="read",
workspace_id="workspace-123",
path="src/app.py",
max_bytes=4096,
json=True,
)
class WriteParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="write",
workspace_id="workspace-123",
path="src/app.py",
text="print('hello')\n",
json=False,
)
class PatchParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="patch",
workspace_patch_command="apply",
workspace_id="workspace-123",
patch=(
"--- a/src/app.py\n"
"+++ b/src/app.py\n"
"@@ -1 +1 @@\n"
"-print('hi')\n"
"+print('hello')\n"
),
json=False,
)
monkeypatch.setattr(cli, "Pyro", StubPyro)
monkeypatch.setattr(cli, "_build_parser", lambda: ListParser())
cli.main()
list_output = capsys.readouterr().out
assert "Workspace path: /workspace/src (recursive=yes)" in list_output
assert "/workspace/src/app.py [file]" in list_output
monkeypatch.setattr(cli, "_build_parser", lambda: ReadParser())
cli.main()
read_payload = json.loads(capsys.readouterr().out)
assert read_payload["path"] == "/workspace/src/app.py"
assert read_payload["content"] == "print('hi')\n"
monkeypatch.setattr(cli, "_build_parser", lambda: WriteParser())
cli.main()
write_output = capsys.readouterr().out
assert "[workspace-file-write] workspace_id=workspace-123" in write_output
monkeypatch.setattr(cli, "_build_parser", lambda: PatchParser())
cli.main()
patch_output = capsys.readouterr().out
assert "[workspace-patch] workspace_id=workspace-123 total=1" in patch_output
def test_cli_workspace_file_write_reads_text_file(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
) -> None:
source_path = tmp_path / "app.py"
source_path.write_text("print('from file')\n", encoding="utf-8")
class StubPyro:
def write_workspace_file(
self,
workspace_id: str,
path: str,
*,
text: str,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "src/app.py"
assert text == "print('from file')\n"
return {
"workspace_id": workspace_id,
"path": "/workspace/src/app.py",
"size_bytes": len(text.encode("utf-8")),
"bytes_written": len(text.encode("utf-8")),
"execution_mode": "guest_vsock",
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="write",
workspace_id="workspace-123",
path="src/app.py",
text=None,
text_file=str(source_path),
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
output = capsys.readouterr().out
assert "[workspace-file-write] workspace_id=workspace-123" in output
def test_cli_workspace_patch_apply_reads_patch_file(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
) -> None:
patch_path = tmp_path / "fix.patch"
patch_text = (
"--- a/src/app.py\n"
"+++ b/src/app.py\n"
"@@ -1 +1 @@\n"
"-print('hi')\n"
"+print('hello')\n"
)
patch_path.write_text(patch_text, encoding="utf-8")
class StubPyro:
def apply_workspace_patch(
self,
workspace_id: str,
*,
patch: str,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert patch == patch_text
return {
"workspace_id": workspace_id,
"changed": True,
"summary": {"total": 1, "added": 0, "modified": 1, "deleted": 0},
"entries": [{"path": "/workspace/src/app.py", "status": "modified"}],
"patch": patch,
"execution_mode": "guest_vsock",
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="patch",
workspace_patch_command="apply",
workspace_id="workspace-123",
patch=None,
patch_file=str(patch_path),
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
output = capsys.readouterr().out
assert "[workspace-patch] workspace_id=workspace-123 total=1" in output
def test_cli_workspace_stop_and_start_print_human_output(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def stop_workspace(self, workspace_id: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
return {
"workspace_id": workspace_id,
"environment": "debian:12",
"state": "stopped",
"workspace_path": "/workspace",
"network_policy": "off",
"execution_mode": "guest_vsock",
"vcpu_count": 1,
"mem_mib": 1024,
"command_count": 2,
"reset_count": 0,
"service_count": 0,
"running_service_count": 0,
}
def start_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",
"network_policy": "off",
"execution_mode": "guest_vsock",
"vcpu_count": 1,
"mem_mib": 1024,
"command_count": 2,
"reset_count": 0,
"service_count": 0,
"running_service_count": 0,
}
class StopParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="stop",
workspace_id="workspace-123",
json=False,
)
class StartParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="start",
workspace_id="workspace-123",
json=False,
)
monkeypatch.setattr(cli, "Pyro", StubPyro)
monkeypatch.setattr(cli, "_build_parser", lambda: StopParser())
cli.main()
stopped_output = capsys.readouterr().out
assert "Stopped workspace ID: workspace-123" in stopped_output
monkeypatch.setattr(cli, "_build_parser", lambda: StartParser())
cli.main()
started_output = capsys.readouterr().out
assert "Started workspace ID: workspace-123" in started_output
def test_cli_workspace_disk_commands_print_human_and_json(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def export_workspace_disk(
self,
workspace_id: str,
*,
output_path: str,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert output_path == "./workspace.ext4"
return {
"workspace_id": workspace_id,
"output_path": "/tmp/workspace.ext4",
"disk_format": "ext4",
"bytes_written": 8192,
}
def list_workspace_disk(
self,
workspace_id: str,
*,
path: str,
recursive: bool,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "/workspace"
assert recursive is True
return {
"workspace_id": workspace_id,
"path": path,
"recursive": recursive,
"entries": [
{
"path": "/workspace/note.txt",
"artifact_type": "file",
"size_bytes": 6,
"link_target": None,
}
],
}
def read_workspace_disk(
self,
workspace_id: str,
path: str,
*,
max_bytes: int,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "note.txt"
assert max_bytes == 4096
return {
"workspace_id": workspace_id,
"path": "/workspace/note.txt",
"size_bytes": 6,
"max_bytes": max_bytes,
"content": "hello\n",
"truncated": False,
}
class ExportParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="disk",
workspace_disk_command="export",
workspace_id="workspace-123",
output="./workspace.ext4",
json=False,
)
class ListParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="disk",
workspace_disk_command="list",
workspace_id="workspace-123",
path="/workspace",
recursive=True,
json=False,
)
class ReadParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="disk",
workspace_disk_command="read",
workspace_id="workspace-123",
path="note.txt",
max_bytes=4096,
json=True,
)
monkeypatch.setattr(cli, "Pyro", StubPyro)
monkeypatch.setattr(cli, "_build_parser", lambda: ExportParser())
cli.main()
export_output = capsys.readouterr().out
assert "[workspace-disk-export] workspace_id=workspace-123" in export_output
monkeypatch.setattr(cli, "_build_parser", lambda: ListParser())
cli.main()
list_output = capsys.readouterr().out
assert "Workspace disk path: /workspace" in list_output
assert "/workspace/note.txt [file]" in list_output
monkeypatch.setattr(cli, "_build_parser", lambda: ReadParser())
cli.main()
read_payload = json.loads(capsys.readouterr().out)
assert read_payload["path"] == "/workspace/note.txt"
assert read_payload["content"] == "hello\n"
def test_cli_workspace_file_read_human_separates_summary_from_content(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def read_workspace_file(
self,
workspace_id: str,
path: str,
*,
max_bytes: int,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "note.txt"
assert max_bytes == 4096
return {
"workspace_id": workspace_id,
"path": "/workspace/note.txt",
"size_bytes": 5,
"max_bytes": max_bytes,
"content": "hello",
"truncated": False,
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="read",
workspace_id="workspace-123",
path="note.txt",
max_bytes=4096,
content_only=False,
json=False,
)
monkeypatch.setattr(cli, "Pyro", StubPyro)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
cli.main()
captured = capsys.readouterr()
assert captured.out == "hello\n"
assert "[workspace-file-read] workspace_id=workspace-123" in captured.err
def test_cli_workspace_file_read_content_only_suppresses_summary(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def read_workspace_file(
self,
workspace_id: str,
path: str,
*,
max_bytes: int,
) -> dict[str, Any]:
return {
"workspace_id": workspace_id,
"path": "/workspace/note.txt",
"size_bytes": 5,
"max_bytes": max_bytes,
"content": "hello",
"truncated": False,
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="file",
workspace_file_command="read",
workspace_id="workspace-123",
path="note.txt",
max_bytes=4096,
content_only=True,
json=False,
)
monkeypatch.setattr(cli, "Pyro", StubPyro)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
cli.main()
captured = capsys.readouterr()
assert captured.out == "hello"
assert captured.err == ""
def test_cli_workspace_disk_read_human_separates_summary_from_content(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def read_workspace_disk(
self,
workspace_id: str,
path: str,
*,
max_bytes: int,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert path == "note.txt"
assert max_bytes == 4096
return {
"workspace_id": workspace_id,
"path": "/workspace/note.txt",
"size_bytes": 5,
"max_bytes": max_bytes,
"content": "hello",
"truncated": False,
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="disk",
workspace_disk_command="read",
workspace_id="workspace-123",
path="note.txt",
max_bytes=4096,
content_only=False,
json=False,
)
monkeypatch.setattr(cli, "Pyro", StubPyro)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
cli.main()
captured = capsys.readouterr()
assert captured.out == "hello\n"
assert "[workspace-disk-read] workspace_id=workspace-123" in captured.err
def test_cli_workspace_disk_read_content_only_suppresses_summary(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def read_workspace_disk(
self,
workspace_id: str,
path: str,
*,
max_bytes: int,
) -> dict[str, Any]:
return {
"workspace_id": workspace_id,
"path": "/workspace/note.txt",
"size_bytes": 5,
"max_bytes": max_bytes,
"content": "hello",
"truncated": False,
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="disk",
workspace_disk_command="read",
workspace_id="workspace-123",
path="note.txt",
max_bytes=4096,
content_only=True,
json=False,
)
monkeypatch.setattr(cli, "Pyro", StubPyro)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
cli.main()
captured = capsys.readouterr()
assert captured.out == "hello"
assert captured.err == ""
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_snapshot_create_list_delete_and_reset_prints_json(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def create_snapshot(self, workspace_id: str, snapshot_name: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert snapshot_name == "checkpoint"
return {
"workspace_id": workspace_id,
"snapshot": {
"snapshot_name": snapshot_name,
"kind": "named",
"entry_count": 3,
"bytes_written": 42,
},
}
def list_snapshots(self, workspace_id: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
return {
"workspace_id": workspace_id,
"count": 2,
"snapshots": [
{
"snapshot_name": "baseline",
"kind": "baseline",
"entry_count": 1,
"bytes_written": 10,
"deletable": False,
},
{
"snapshot_name": "checkpoint",
"kind": "named",
"entry_count": 3,
"bytes_written": 42,
"deletable": True,
},
],
}
def reset_workspace(self, workspace_id: str, *, snapshot: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert snapshot == "checkpoint"
return {
"workspace_id": workspace_id,
"state": "started",
"workspace_path": "/workspace",
"reset_count": 2,
"workspace_reset": {
"snapshot_name": snapshot,
"kind": "named",
"destination": "/workspace",
"entry_count": 3,
"bytes_written": 42,
},
}
def delete_snapshot(self, workspace_id: str, snapshot_name: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert snapshot_name == "checkpoint"
return {
"workspace_id": workspace_id,
"snapshot_name": snapshot_name,
"deleted": True,
}
class SnapshotCreateParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="create",
workspace_id="workspace-123",
snapshot_name="checkpoint",
json=True,
)
monkeypatch.setattr(cli, "_build_parser", lambda: SnapshotCreateParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
created = json.loads(capsys.readouterr().out)
assert created["snapshot"]["snapshot_name"] == "checkpoint"
class SnapshotListParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="list",
workspace_id="workspace-123",
json=True,
)
monkeypatch.setattr(cli, "_build_parser", lambda: SnapshotListParser())
cli.main()
listed = json.loads(capsys.readouterr().out)
assert listed["count"] == 2
class ResetParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="reset",
workspace_id="workspace-123",
snapshot="checkpoint",
json=True,
)
monkeypatch.setattr(cli, "_build_parser", lambda: ResetParser())
cli.main()
reset = json.loads(capsys.readouterr().out)
assert reset["workspace_reset"]["snapshot_name"] == "checkpoint"
class SnapshotDeleteParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="delete",
workspace_id="workspace-123",
snapshot_name="checkpoint",
json=True,
)
monkeypatch.setattr(cli, "_build_parser", lambda: SnapshotDeleteParser())
cli.main()
deleted = json.loads(capsys.readouterr().out)
assert deleted["deleted"] is True
def test_cli_workspace_reset_prints_human_output(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def reset_workspace(self, workspace_id: str, *, snapshot: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert snapshot == "baseline"
return {
"workspace_id": workspace_id,
"state": "started",
"environment": "debian:12",
"workspace_path": "/workspace",
"workspace_seed": {
"mode": "directory",
"seed_path": "/tmp/repo",
"destination": "/workspace",
"entry_count": 1,
"bytes_written": 4,
},
"execution_mode": "guest_vsock",
"command_count": 0,
"service_count": 0,
"running_service_count": 0,
"reset_count": 3,
"last_reset_at": 123.0,
"workspace_reset": {
"snapshot_name": "baseline",
"kind": "baseline",
"destination": "/workspace",
"entry_count": 1,
"bytes_written": 4,
},
}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="reset",
workspace_id="workspace-123",
snapshot="baseline",
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
output = capsys.readouterr().out
assert "Reset source: baseline (baseline)" in output
assert "Reset count: 3" in output
def test_cli_workspace_snapshot_prints_human_output(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def create_snapshot(self, workspace_id: str, snapshot_name: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert snapshot_name == "checkpoint"
return {
"workspace_id": workspace_id,
"snapshot": {
"snapshot_name": snapshot_name,
"kind": "named",
"entry_count": 3,
"bytes_written": 42,
},
}
def list_snapshots(self, workspace_id: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
return {
"workspace_id": workspace_id,
"count": 2,
"snapshots": [
{
"snapshot_name": "baseline",
"kind": "baseline",
"entry_count": 1,
"bytes_written": 10,
"deletable": False,
},
{
"snapshot_name": "checkpoint",
"kind": "named",
"entry_count": 3,
"bytes_written": 42,
"deletable": True,
},
],
}
def delete_snapshot(self, workspace_id: str, snapshot_name: str) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert snapshot_name == "checkpoint"
return {
"workspace_id": workspace_id,
"snapshot_name": snapshot_name,
"deleted": True,
}
class SnapshotCreateParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="create",
workspace_id="workspace-123",
snapshot_name="checkpoint",
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: SnapshotCreateParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
create_output = capsys.readouterr().out
assert "[workspace-snapshot-create] workspace_id=workspace-123" in create_output
assert "snapshot_name=checkpoint kind=named" in create_output
class SnapshotListParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="list",
workspace_id="workspace-123",
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: SnapshotListParser())
cli.main()
list_output = capsys.readouterr().out
assert "baseline [baseline]" in list_output
assert "checkpoint [named]" in list_output
class SnapshotDeleteParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="delete",
workspace_id="workspace-123",
snapshot_name="checkpoint",
json=False,
)
monkeypatch.setattr(cli, "_build_parser", lambda: SnapshotDeleteParser())
cli.main()
delete_output = capsys.readouterr().out
assert "Deleted workspace snapshot: checkpoint" in delete_output
def test_cli_workspace_snapshot_error_paths(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def create_snapshot(self, workspace_id: str, snapshot_name: str) -> dict[str, Any]:
del workspace_id, snapshot_name
raise RuntimeError("create boom")
def list_snapshots(self, workspace_id: str) -> dict[str, Any]:
del workspace_id
raise RuntimeError("list boom")
def delete_snapshot(self, workspace_id: str, snapshot_name: str) -> dict[str, Any]:
del workspace_id, snapshot_name
raise RuntimeError("delete boom")
def _run(args: argparse.Namespace) -> tuple[str, str]:
class StubParser:
def parse_args(self) -> argparse.Namespace:
return args
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
with pytest.raises(SystemExit, match="1"):
cli.main()
captured = capsys.readouterr()
return captured.out, captured.err
out, err = _run(
argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="create",
workspace_id="workspace-123",
snapshot_name="checkpoint",
json=True,
)
)
assert json.loads(out)["error"] == "create boom"
assert err == ""
out, err = _run(
argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="list",
workspace_id="workspace-123",
json=False,
)
)
assert out == ""
assert "[error] list boom" in err
out, err = _run(
argparse.Namespace(
command="workspace",
workspace_command="snapshot",
workspace_snapshot_command="delete",
workspace_id="workspace-123",
snapshot_name="checkpoint",
json=False,
)
)
assert out == ""
assert "[error] delete boom" in err
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,
secret_env: dict[str, str] | None = None,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert command == "false"
assert timeout_seconds == 30
assert secret_env is None
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,
secret_env=[],
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,
secret_env: dict[str, str] | None = None,
) -> dict[str, Any]:
del workspace_id, command, timeout_seconds
assert secret_env is None
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,
secret_env=[],
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,
secret_env: dict[str, str] | None = None,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert cwd == "/workspace"
assert cols == 120
assert rows == 30
assert secret_env is None
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,
plain: bool = False,
wait_for_idle_ms: int | None = None,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert shell_id == "shell-123"
assert cursor == 0
assert max_chars == 1024
assert plain is True
assert wait_for_idle_ms == 300
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,
"plain": plain,
"wait_for_idle_ms": wait_for_idle_ms,
}
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,
secret_env=[],
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,
plain=True,
wait_for_idle_ms=300,
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
assert "plain=True" in captured.err
assert "wait_for_idle_ms=300" in captured.err
def test_cli_workspace_shell_open_prints_id_only(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def open_shell(
self,
workspace_id: str,
*,
cwd: str,
cols: int,
rows: int,
secret_env: dict[str, str] | None = None,
) -> dict[str, Any]:
assert workspace_id == "workspace-123"
assert cwd == "/workspace"
assert cols == 120
assert rows == 30
assert secret_env is None
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",
}
class StubParser:
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,
secret_env=[],
json=False,
id_only=True,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
captured = capsys.readouterr()
assert captured.out == "shell-123\n"
assert captured.err == ""
def test_chat_host_docs_and_examples_recommend_workspace_core() -> None:
readme = Path("README.md").read_text(encoding="utf-8")
install = Path("docs/install.md").read_text(encoding="utf-8")
first_run = Path("docs/first-run.md").read_text(encoding="utf-8")
integrations = Path("docs/integrations.md").read_text(encoding="utf-8")
mcp_config = Path("examples/mcp_client_config.md").read_text(encoding="utf-8")
claude_code = Path("examples/claude_code_mcp.md").read_text(encoding="utf-8")
codex = Path("examples/codex_mcp.md").read_text(encoding="utf-8")
opencode = json.loads(Path("examples/opencode_mcp_config.json").read_text(encoding="utf-8"))
claude_cmd = (
"claude mcp add pyro -- uvx --from pyro-mcp pyro mcp serve --profile workspace-core"
)
codex_cmd = (
"codex mcp add pyro -- uvx --from pyro-mcp pyro mcp serve --profile workspace-core"
)
assert "## Chat Host Quickstart" in readme
assert "pyro mcp serve --profile workspace-core" in readme
assert claude_cmd in readme
assert codex_cmd in readme
assert "examples/opencode_mcp_config.json" in readme
assert "recommended first profile for normal persistent chat editing" in readme
assert "## Chat Host Quickstart" in install
assert "pyro mcp serve --profile workspace-core" in install
assert claude_cmd in install
assert codex_cmd in install
assert "advanced 3.x compatibility surface" in install
assert claude_cmd in first_run
assert codex_cmd in first_run
assert "Start most chat hosts with `workspace-core`." in integrations
assert "examples/claude_code_mcp.md" in integrations
assert "examples/codex_mcp.md" in integrations
assert "examples/opencode_mcp_config.json" in integrations
assert '`Pyro.create_server(profile="workspace-core")` for most chat hosts' in integrations
assert "Recommended default for most chat hosts: `workspace-core`." in mcp_config
assert "Use the host-specific examples first when they apply:" in mcp_config
assert "claude_code_mcp.md" in mcp_config
assert "codex_mcp.md" in mcp_config
assert "opencode_mcp_config.json" in mcp_config
assert claude_cmd in claude_code
assert "claude mcp list" in claude_code
assert "workspace-full" in claude_code
assert codex_cmd in codex
assert "codex mcp list" in codex
assert "workspace-full" in codex
assert opencode == {
"mcp": {
"pyro": {
"type": "local",
"enabled": True,
"command": [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
"--profile",
"workspace-core",
],
}
}
}
def test_content_only_read_docs_are_aligned() -> None:
readme = Path("README.md").read_text(encoding="utf-8")
install = Path("docs/install.md").read_text(encoding="utf-8")
first_run = Path("docs/first-run.md").read_text(encoding="utf-8")
assert 'workspace file read "$WORKSPACE_ID" note.txt --content-only' in readme
assert 'workspace file read "$WORKSPACE_ID" note.txt --content-only' in install
assert 'workspace file read "$WORKSPACE_ID" note.txt --content-only' in first_run
assert 'workspace disk read "$WORKSPACE_ID" note.txt --content-only' in first_run
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,
secret_env: dict[str, str] | None = None,
) -> dict[str, Any]:
assert secret_env is None
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,
plain: bool = False,
wait_for_idle_ms: int | None = None,
) -> dict[str, Any]:
assert plain is False
assert wait_for_idle_ms is None
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,
"plain": plain,
"wait_for_idle_ms": wait_for_idle_ms,
}
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,
secret_env=[],
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,
plain=False,
wait_for_idle_ms=None,
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,
"plain": False,
"wait_for_idle_ms": None,
},
),
("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"}
assert kwargs["published_ports"] == [{"host_port": 18080, "guest_port": 8080}]
return {
"workspace_id": workspace_id,
"service_name": service_name,
"state": "running",
"cwd": "/workspace",
"execution_mode": "guest_vsock",
"published_ports": [
{
"host": "127.0.0.1",
"host_port": 18080,
"guest_port": 8080,
"protocol": "tcp",
}
],
}
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,
publish=["18080:8080"],
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",
"published_ports": [
{
"host": "127.0.0.1",
"host_port": 18080,
"guest_port": 8080,
"protocol": "tcp",
}
],
"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 published=127.0.0.1:18080->8080/tcp" 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": "3.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: 3.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": "3.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, *, profile: str) -> Any:
observed["profile"] = profile
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", profile="workspace-core")
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
assert observed == {"profile": "workspace-core", "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
def test_cli_workspace_create_passes_secrets(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
) -> None:
secret_file = tmp_path / "token.txt"
secret_file.write_text("from-file\n", encoding="utf-8")
class StubPyro:
def create_workspace(self, **kwargs: Any) -> dict[str, Any]:
assert kwargs["environment"] == "debian:12"
assert kwargs["seed_path"] == "./repo"
assert kwargs["secrets"] == [
{"name": "API_TOKEN", "value": "expected"},
{"name": "FILE_TOKEN", "file_path": str(secret_file)},
]
assert kwargs["name"] is None
assert kwargs["labels"] is None
return {"workspace_id": "ws-123"}
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_policy="off",
allow_host_compat=False,
seed_path="./repo",
name=None,
label=[],
secret=["API_TOKEN=expected"],
secret_file=[f"FILE_TOKEN={secret_file}"],
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"] == "ws-123"
def test_cli_workspace_exec_passes_secret_env(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def exec_workspace(self, workspace_id: str, **kwargs: Any) -> dict[str, Any]:
assert workspace_id == "ws-123"
assert kwargs["command"] == "sh -lc 'test \"$API_TOKEN\" = \"expected\"'"
assert kwargs["secret_env"] == {"API_TOKEN": "API_TOKEN", "TOKEN": "PIP_TOKEN"}
return {"exit_code": 0, "stdout": "", "stderr": ""}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="exec",
workspace_id="ws-123",
timeout_seconds=30,
secret_env=["API_TOKEN", "TOKEN=PIP_TOKEN"],
json=True,
command_args=["--", "sh", "-lc", 'test "$API_TOKEN" = "expected"'],
)
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_workspace_shell_open_passes_secret_env(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class StubPyro:
def open_shell(self, workspace_id: str, **kwargs: Any) -> dict[str, Any]:
assert workspace_id == "ws-123"
assert kwargs["secret_env"] == {"TOKEN": "TOKEN", "API": "API_TOKEN"}
return {"workspace_id": workspace_id, "shell_id": "shell-1", "state": "running"}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="shell",
workspace_shell_command="open",
workspace_id="ws-123",
cwd="/workspace",
cols=120,
rows=30,
secret_env=["TOKEN", "API=API_TOKEN"],
json=True,
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
output = json.loads(capsys.readouterr().out)
assert output["shell_id"] == "shell-1"
def test_cli_workspace_service_start_passes_secret_env(
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 == "ws-123"
assert service_name == "app"
assert kwargs["secret_env"] == {"TOKEN": "TOKEN", "API": "API_TOKEN"}
assert kwargs["readiness"] == {"type": "file", "path": ".ready"}
assert kwargs["command"] == "sh -lc 'touch .ready && while true; do sleep 60; done'"
return {"workspace_id": workspace_id, "service_name": service_name, "state": "running"}
class StubParser:
def parse_args(self) -> argparse.Namespace:
return argparse.Namespace(
command="workspace",
workspace_command="service",
workspace_service_command="start",
workspace_id="ws-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,
secret_env=["TOKEN", "API=API_TOKEN"],
json=True,
command_args=["--", "sh", "-lc", "touch .ready && while true; do sleep 60; done"],
)
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
output = json.loads(capsys.readouterr().out)
assert output["state"] == "running"
def test_cli_workspace_secret_parsers_validate_syntax(tmp_path: Path) -> None:
secret_file = tmp_path / "token.txt"
secret_file.write_text("expected\n", encoding="utf-8")
assert cli._parse_workspace_secret_option("API_TOKEN=expected") == { # noqa: SLF001
"name": "API_TOKEN",
"value": "expected",
}
assert cli._parse_workspace_secret_file_option(f"FILE_TOKEN={secret_file}") == { # noqa: SLF001
"name": "FILE_TOKEN",
"file_path": str(secret_file),
}
assert cli._parse_workspace_secret_env_options(["TOKEN", "API=PIP_TOKEN"]) == { # noqa: SLF001
"TOKEN": "TOKEN",
"API": "PIP_TOKEN",
}
with pytest.raises(ValueError, match="NAME=VALUE"):
cli._parse_workspace_secret_option("API_TOKEN") # noqa: SLF001
with pytest.raises(ValueError, match="NAME=PATH"):
cli._parse_workspace_secret_file_option("FILE_TOKEN=") # noqa: SLF001
with pytest.raises(ValueError, match="must name a secret"):
cli._parse_workspace_secret_env_options(["=TOKEN"]) # noqa: SLF001
with pytest.raises(ValueError, match="must name an environment variable"):
cli._parse_workspace_secret_env_options(["TOKEN="]) # noqa: SLF001
with pytest.raises(ValueError, match="more than once"):
cli._parse_workspace_secret_env_options(["TOKEN", "TOKEN=API_TOKEN"]) # noqa: SLF001
def test_cli_workspace_publish_parser_validates_syntax() -> None:
assert cli._parse_workspace_publish_options(["8080"]) == [ # noqa: SLF001
{"host_port": None, "guest_port": 8080}
]
assert cli._parse_workspace_publish_options(["18080:8080"]) == [ # noqa: SLF001
{"host_port": 18080, "guest_port": 8080}
]
with pytest.raises(ValueError, match="must not be empty"):
cli._parse_workspace_publish_options([" "]) # noqa: SLF001
with pytest.raises(ValueError, match="must use GUEST_PORT or HOST_PORT:GUEST_PORT"):
cli._parse_workspace_publish_options(["bad"]) # noqa: SLF001
with pytest.raises(ValueError, match="must use GUEST_PORT or HOST_PORT:GUEST_PORT"):
cli._parse_workspace_publish_options(["bad:8080"]) # noqa: SLF001
def test_cli_workspace_service_start_rejects_multiple_readiness_flags_json(
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=None,
ready_http="http://127.0.0.1:8080/",
ready_command=None,
ready_timeout_seconds=30,
ready_interval_ms=500,
publish=[],
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)
with pytest.raises(SystemExit, match="1"):
cli.main()
payload = json.loads(capsys.readouterr().out)
assert "choose at most one" in payload["error"]
def test_cli_workspace_service_start_prints_human_with_ready_http(
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": "http", "url": "http://127.0.0.1:8080/ready"}
return {
"workspace_id": workspace_id,
"service_name": service_name,
"state": "running",
"cwd": "/workspace",
"execution_mode": "guest_vsock",
"readiness": kwargs["readiness"],
}
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="http://127.0.0.1:8080/ready",
ready_command=None,
ready_timeout_seconds=30,
ready_interval_ms=500,
publish=[],
secret_env=[],
json=False,
command_args=["--", "sh", "-lc", "while true; do sleep 60; done"],
)
monkeypatch.setattr(cli, "_build_parser", lambda: StartParser())
monkeypatch.setattr(cli, "Pyro", StubPyro)
cli.main()
captured = capsys.readouterr()
assert "workspace-service-start" in captured.err
assert "service_name=app" in captured.err
def test_print_workspace_summary_human_includes_secret_metadata(
capsys: pytest.CaptureFixture[str],
) -> None:
cli._print_workspace_summary_human(
{
"workspace_id": "ws-123",
"environment": "debian:12",
"state": "started",
"workspace_path": "/workspace",
"workspace_seed": {
"mode": "directory",
"seed_path": "/tmp/repo",
},
"secrets": [
{"name": "API_TOKEN", "source_kind": "literal"},
{"name": "FILE_TOKEN", "source_kind": "file"},
],
"execution_mode": "guest_vsock",
"vcpu_count": 1,
"mem_mib": 1024,
"command_count": 0,
},
action="Workspace",
)
output = capsys.readouterr().out
assert "Workspace ID: ws-123" in output
assert "Workspace seed: directory from /tmp/repo" in output
assert "Secrets: API_TOKEN (literal), FILE_TOKEN (file)" in output
View git blame Copy permalink