Thales Maciel
663241d5d2
Make the local chat-host loop explicit and cheap so users can warm the machine once instead of rediscovering environment and guest setup on every session. Add cache-backed daily-loop manifests plus the new `pyro prepare` flow, extend `pyro doctor --environment` with warm/cold/stale readiness reporting, and add `make smoke-daily-loop` to prove the warmed repro-fix reset path end to end. Also fix `python -m pyro_mcp.cli` to invoke `main()` so the new smoke and `dist-check` actually exercise the CLI module, and update the docs/roadmap to present `doctor -> prepare -> connect host -> reset` as the recommended daily path. Validation: `uv lock`, `UV_OFFLINE=1 UV_CACHE_DIR=.uv-cache make check`, `UV_OFFLINE=1 UV_CACHE_DIR=.uv-cache make dist-check`, and `UV_OFFLINE=1 UV_CACHE_DIR=.uv-cache make smoke-daily-loop`.
4947 lines
171 KiB
Python
4947 lines
171 KiB
Python
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
|
|
from pyro_mcp.host_helpers import HostDoctorEntry
|
|
|
|
|
|
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 zero-to-hero path:" in help_text
|
|
assert "pyro doctor" in help_text
|
|
assert "pyro prepare debian:12" in help_text
|
|
assert "pyro run debian:12 -- git --version" in help_text
|
|
assert "pyro host connect claude-code" in help_text
|
|
assert "Connect a chat host after that:" in help_text
|
|
assert "pyro host connect claude-code" in help_text
|
|
assert "pyro host connect codex" in help_text
|
|
assert "pyro host print-config opencode" in help_text
|
|
assert "Daily local loop after the first warmup:" in help_text
|
|
assert "pyro doctor --environment debian:12" in help_text
|
|
assert "pyro workspace reset WORKSPACE_ID" in help_text
|
|
assert "If you want terminal-level visibility into the workspace model:" in help_text
|
|
assert "pyro workspace exec WORKSPACE_ID -- cat note.txt" in help_text
|
|
assert "pyro workspace summary WORKSPACE_ID" 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
|
|
|
|
|
|
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
|
|
|
|
prepare_help = _subparser_choice(parser, "prepare").format_help()
|
|
assert "Warm the recommended guest-backed daily loop" in prepare_help
|
|
assert "pyro prepare debian:12 --network" in prepare_help
|
|
assert "--network" in prepare_help
|
|
assert "--force" in prepare_help
|
|
|
|
host_help = _subparser_choice(parser, "host").format_help()
|
|
assert "Connect or repair the supported Claude Code, Codex, and OpenCode" in host_help
|
|
assert "pyro host connect claude-code" in host_help
|
|
assert "pyro host repair opencode" in host_help
|
|
|
|
host_connect_help = _subparser_choice(
|
|
_subparser_choice(parser, "host"), "connect"
|
|
).format_help()
|
|
assert "--installed-package" in host_connect_help
|
|
assert "--project-path" in host_connect_help
|
|
assert "--repo-url" in host_connect_help
|
|
assert "--repo-ref" in host_connect_help
|
|
assert "--no-project-source" in host_connect_help
|
|
|
|
host_print_config_help = _subparser_choice(
|
|
_subparser_choice(parser, "host"), "print-config"
|
|
).format_help()
|
|
assert "--output" in host_print_config_help
|
|
|
|
host_doctor_help = _subparser_choice(_subparser_choice(parser, "host"), "doctor").format_help()
|
|
assert "--config-path" in host_doctor_help
|
|
|
|
host_repair_help = _subparser_choice(_subparser_choice(parser, "host"), "repair").format_help()
|
|
assert "--config-path" in host_repair_help
|
|
|
|
doctor_help = _subparser_choice(parser, "doctor").format_help()
|
|
assert "Check host prerequisites and embedded runtime health" in doctor_help
|
|
assert "--environment" in doctor_help
|
|
assert "pyro doctor --environment debian:12" 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 "workspace-core: default for normal persistent chat editing" in mcp_help
|
|
assert "workspace-full: larger opt-in surface" in mcp_help
|
|
assert "--project-path" in mcp_help
|
|
assert "--repo-url" in mcp_help
|
|
assert "--repo-ref" in mcp_help
|
|
assert "--no-project-source" in mcp_help
|
|
assert "pyro mcp serve --project-path ." in mcp_help
|
|
assert "pyro mcp serve --repo-url https://github.com/example/project.git" in mcp_help
|
|
|
|
workspace_help = _subparser_choice(parser, "workspace").format_help()
|
|
assert "Use the workspace model when you need one sandbox to stay alive" 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 summary WORKSPACE_ID" 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_summary_help = _subparser_choice(
|
|
_subparser_choice(parser, "workspace"), "summary"
|
|
).format_help()
|
|
assert "Summarize the current workspace session since the last reset" in workspace_summary_help
|
|
assert "pyro workspace summary WORKSPACE_ID" in workspace_summary_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_host_connect_dispatch(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
class StubPyro:
|
|
pass
|
|
|
|
class StubParser:
|
|
def parse_args(self) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
command="host",
|
|
host_command="connect",
|
|
host="codex",
|
|
installed_package=False,
|
|
profile="workspace-core",
|
|
project_path=None,
|
|
repo_url=None,
|
|
repo_ref=None,
|
|
no_project_source=False,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
|
|
monkeypatch.setattr(cli, "Pyro", StubPyro)
|
|
monkeypatch.setattr(
|
|
cli,
|
|
"connect_cli_host",
|
|
lambda host, *, config: {
|
|
"host": host,
|
|
"server_command": ["uvx", "--from", "pyro-mcp", "pyro", "mcp", "serve"],
|
|
"verification_command": ["codex", "mcp", "list"],
|
|
},
|
|
)
|
|
|
|
cli.main()
|
|
captured = capsys.readouterr()
|
|
assert captured.out == (
|
|
"Connected pyro to codex.\n"
|
|
"Server command: uvx --from pyro-mcp pyro mcp serve\n"
|
|
"Verify with: codex mcp list\n"
|
|
)
|
|
assert captured.err == ""
|
|
|
|
|
|
def test_cli_host_doctor_prints_human(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
class StubPyro:
|
|
pass
|
|
|
|
class StubParser:
|
|
def parse_args(self) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
command="host",
|
|
host_command="doctor",
|
|
installed_package=False,
|
|
profile="workspace-core",
|
|
project_path=None,
|
|
repo_url=None,
|
|
repo_ref=None,
|
|
no_project_source=False,
|
|
config_path=None,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
|
|
monkeypatch.setattr(cli, "Pyro", StubPyro)
|
|
monkeypatch.setattr(
|
|
cli,
|
|
"doctor_hosts",
|
|
lambda **_: [
|
|
HostDoctorEntry(
|
|
host="codex",
|
|
installed=True,
|
|
configured=False,
|
|
status="missing",
|
|
details="codex entry missing",
|
|
repair_command="pyro host repair codex",
|
|
)
|
|
],
|
|
)
|
|
|
|
cli.main()
|
|
captured = capsys.readouterr()
|
|
assert "codex: missing installed=yes configured=no" in captured.out
|
|
assert "repair: pyro host repair codex" in captured.out
|
|
assert captured.err == ""
|
|
|
|
|
|
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",
|
|
environment="debian:12",
|
|
json=True,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
|
|
monkeypatch.setattr(
|
|
cli,
|
|
"doctor_report",
|
|
lambda *, platform, environment: {
|
|
"platform": platform,
|
|
"environment": environment,
|
|
"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_summary_prints_json(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
class StubPyro:
|
|
def summarize_workspace(self, workspace_id: str) -> dict[str, Any]:
|
|
assert workspace_id == "workspace-123"
|
|
return {
|
|
"workspace_id": workspace_id,
|
|
"name": "review-eval",
|
|
"labels": {"suite": "smoke"},
|
|
"environment": "debian:12",
|
|
"state": "started",
|
|
"last_activity_at": 2.0,
|
|
"session_started_at": 1.0,
|
|
"outcome": {
|
|
"command_count": 1,
|
|
"last_command": {"command": "cat note.txt", "exit_code": 0},
|
|
"service_count": 0,
|
|
"running_service_count": 0,
|
|
"export_count": 1,
|
|
"snapshot_count": 1,
|
|
"reset_count": 0,
|
|
},
|
|
"commands": {"total": 1, "recent": []},
|
|
"edits": {"recent": []},
|
|
"changes": {"available": True, "changed": False, "summary": None, "entries": []},
|
|
"services": {"current": [], "recent": []},
|
|
"artifacts": {"exports": []},
|
|
"snapshots": {"named_count": 1, "recent": []},
|
|
}
|
|
|
|
class SummaryParser:
|
|
def parse_args(self) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
command="workspace",
|
|
workspace_command="summary",
|
|
workspace_id="workspace-123",
|
|
json=True,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: SummaryParser())
|
|
monkeypatch.setattr(cli, "Pyro", StubPyro)
|
|
cli.main()
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["workspace_id"] == "workspace-123"
|
|
assert payload["outcome"]["export_count"] == 1
|
|
|
|
|
|
def test_cli_workspace_summary_prints_human(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
class StubPyro:
|
|
def summarize_workspace(self, workspace_id: str) -> dict[str, Any]:
|
|
assert workspace_id == "workspace-123"
|
|
return {
|
|
"workspace_id": workspace_id,
|
|
"name": "review-eval",
|
|
"labels": {"suite": "smoke", "use_case": "review-eval"},
|
|
"environment": "debian:12",
|
|
"state": "started",
|
|
"last_activity_at": 3.0,
|
|
"session_started_at": 1.0,
|
|
"outcome": {
|
|
"command_count": 2,
|
|
"last_command": {"command": "sh review.sh", "exit_code": 0},
|
|
"service_count": 1,
|
|
"running_service_count": 0,
|
|
"export_count": 1,
|
|
"snapshot_count": 1,
|
|
"reset_count": 0,
|
|
},
|
|
"commands": {
|
|
"total": 2,
|
|
"recent": [
|
|
{
|
|
"sequence": 2,
|
|
"command": "sh review.sh",
|
|
"cwd": "/workspace",
|
|
"exit_code": 0,
|
|
"duration_ms": 12,
|
|
"execution_mode": "guest_vsock",
|
|
"recorded_at": 3.0,
|
|
}
|
|
],
|
|
},
|
|
"edits": {
|
|
"recent": [
|
|
{
|
|
"event_kind": "patch_apply",
|
|
"recorded_at": 2.0,
|
|
"path": "/workspace/note.txt",
|
|
}
|
|
]
|
|
},
|
|
"changes": {
|
|
"available": True,
|
|
"changed": True,
|
|
"summary": {
|
|
"total": 1,
|
|
"added": 0,
|
|
"modified": 1,
|
|
"deleted": 0,
|
|
"type_changed": 0,
|
|
"text_patched": 1,
|
|
"non_text": 0,
|
|
},
|
|
"entries": [
|
|
{
|
|
"path": "/workspace/note.txt",
|
|
"status": "modified",
|
|
"artifact_type": "file",
|
|
}
|
|
],
|
|
},
|
|
"services": {
|
|
"current": [{"service_name": "app", "state": "stopped"}],
|
|
"recent": [
|
|
{
|
|
"event_kind": "service_stop",
|
|
"service_name": "app",
|
|
"state": "stopped",
|
|
}
|
|
],
|
|
},
|
|
"artifacts": {
|
|
"exports": [
|
|
{
|
|
"workspace_path": "review-report.txt",
|
|
"output_path": "/tmp/review-report.txt",
|
|
}
|
|
]
|
|
},
|
|
"snapshots": {
|
|
"named_count": 1,
|
|
"recent": [{"event_kind": "snapshot_create", "snapshot_name": "checkpoint"}],
|
|
},
|
|
}
|
|
|
|
class SummaryParser:
|
|
def parse_args(self) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
command="workspace",
|
|
workspace_command="summary",
|
|
workspace_id="workspace-123",
|
|
json=False,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: SummaryParser())
|
|
monkeypatch.setattr(cli, "Pyro", StubPyro)
|
|
cli.main()
|
|
output = capsys.readouterr().out
|
|
assert "Workspace review: workspace-123" in output
|
|
assert "Outcome: commands=2 services=0/1 exports=1 snapshots=1 resets=0" in output
|
|
assert "Recent commands:" in output
|
|
assert "Recent edits:" in output
|
|
assert "Changes: total=1 added=0 modified=1 deleted=0 type_changed=0 non_text=0" in output
|
|
assert "Recent exports:" in output
|
|
assert "Recent snapshot events:" in output
|
|
|
|
|
|
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_modes_first() -> 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_helper = "pyro host connect claude-code --mode cold-start"
|
|
codex_helper = "pyro host connect codex --mode repro-fix"
|
|
inspect_helper = "pyro host connect codex --mode inspect"
|
|
review_helper = "pyro host connect claude-code --mode review-eval"
|
|
opencode_helper = "pyro host print-config opencode --mode repro-fix"
|
|
claude_cmd = "claude mcp add pyro -- uvx --from pyro-mcp pyro mcp serve --mode cold-start"
|
|
codex_cmd = "codex mcp add pyro -- uvx --from pyro-mcp pyro mcp serve --mode repro-fix"
|
|
|
|
assert "## Chat Host Quickstart" in readme
|
|
assert claude_helper in readme
|
|
assert codex_helper in readme
|
|
assert inspect_helper in readme
|
|
assert review_helper in readme
|
|
assert opencode_helper in readme
|
|
assert "examples/opencode_mcp_config.json" in readme
|
|
assert "pyro host doctor" in readme
|
|
assert "pyro mcp serve --mode repro-fix" in readme
|
|
assert "generic no-mode path" in readme
|
|
assert "auto-detects the current Git checkout" in readme.replace("\n", " ")
|
|
assert "--project-path /abs/path/to/repo" in readme
|
|
assert "--repo-url https://github.com/example/project.git" in readme
|
|
|
|
assert "## 6. Connect a chat host" in install
|
|
assert claude_helper in install
|
|
assert codex_helper in install
|
|
assert inspect_helper in install
|
|
assert review_helper in install
|
|
assert opencode_helper in install
|
|
assert "workspace-full" in install
|
|
assert "--project-path /abs/path/to/repo" in install
|
|
assert "pyro mcp serve --mode cold-start" in install
|
|
|
|
assert claude_helper in first_run
|
|
assert codex_helper in first_run
|
|
assert inspect_helper in first_run
|
|
assert review_helper in first_run
|
|
assert opencode_helper in first_run
|
|
assert "--project-path /abs/path/to/repo" in first_run
|
|
assert "pyro mcp serve --mode review-eval" in first_run
|
|
|
|
assert claude_helper in integrations
|
|
assert codex_helper in integrations
|
|
assert inspect_helper in integrations
|
|
assert review_helper in integrations
|
|
assert opencode_helper in integrations
|
|
assert "## Recommended Modes" in integrations
|
|
assert "pyro mcp serve --mode inspect" in integrations
|
|
assert "auto-detects the current Git checkout" 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 "generic no-mode path" in integrations
|
|
assert "--project-path /abs/path/to/repo" in integrations
|
|
assert "--repo-url https://github.com/example/project.git" in integrations
|
|
|
|
assert "Recommended named modes for most chat hosts in `4.x`:" 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 '"serve", "--mode", "repro-fix"' in mcp_config
|
|
|
|
assert claude_helper in claude_code
|
|
assert claude_cmd in claude_code
|
|
assert "claude mcp list" in claude_code
|
|
assert "pyro host repair claude-code --mode cold-start" in claude_code
|
|
assert "workspace-full" in claude_code
|
|
assert "--project-path /abs/path/to/repo" in claude_code
|
|
|
|
assert codex_helper in codex
|
|
assert codex_cmd in codex
|
|
assert "codex mcp list" in codex
|
|
assert "pyro host repair codex --mode repro-fix" in codex
|
|
assert "workspace-full" in codex
|
|
assert "--project-path /abs/path/to/repo" in codex
|
|
|
|
assert opencode == {
|
|
"mcp": {
|
|
"pyro": {
|
|
"type": "local",
|
|
"enabled": True,
|
|
"command": [
|
|
"uvx",
|
|
"--from",
|
|
"pyro-mcp",
|
|
"pyro",
|
|
"mcp",
|
|
"serve",
|
|
"--mode",
|
|
"repro-fix",
|
|
],
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
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 patch apply "$WORKSPACE_ID" --patch-file fix.patch' in first_run
|
|
|
|
|
|
def test_daily_loop_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")
|
|
integrations = Path("docs/integrations.md").read_text(encoding="utf-8")
|
|
|
|
assert "pyro prepare debian:12" in readme
|
|
assert "pyro prepare debian:12" in install
|
|
assert "pyro prepare debian:12" in first_run
|
|
assert "pyro prepare debian:12" in integrations
|
|
assert "pyro doctor --environment debian:12" in readme
|
|
assert "pyro doctor --environment debian:12" in install
|
|
assert "pyro doctor --environment debian:12" in first_run
|
|
|
|
|
|
def test_workspace_summary_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 summary "$WORKSPACE_ID"' in readme
|
|
assert 'workspace summary "$WORKSPACE_ID"' in install
|
|
assert 'workspace summary "$WORKSPACE_ID"' 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",
|
|
environment="debian:12",
|
|
json=False,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
|
|
monkeypatch.setattr(
|
|
cli,
|
|
"doctor_report",
|
|
lambda *, platform, environment: {
|
|
"platform": platform,
|
|
"runtime_ok": True,
|
|
"issues": [],
|
|
"kvm": {"exists": True, "readable": True, "writable": True},
|
|
"runtime": {"catalog_version": "4.5.0", "cache_dir": "/cache"},
|
|
"daily_loop": {
|
|
"environment": environment,
|
|
"status": "cold",
|
|
"installed": False,
|
|
"network_prepared": False,
|
|
"prepared_at": None,
|
|
"manifest_path": "/cache/.prepare/linux-x86_64/debian_12.json",
|
|
"reason": "daily loop has not been prepared yet",
|
|
"cache_dir": "/cache",
|
|
},
|
|
},
|
|
)
|
|
cli.main()
|
|
output = capsys.readouterr().out
|
|
assert "Runtime: PASS" in output
|
|
assert "Daily loop: COLD (debian:12)" in output
|
|
assert "Run: pyro prepare debian:12" in output
|
|
|
|
|
|
def test_cli_prepare_prints_human(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
class StubManager:
|
|
def prepare_daily_loop(
|
|
self,
|
|
environment: str,
|
|
*,
|
|
network: bool,
|
|
force: bool,
|
|
) -> dict[str, object]:
|
|
assert environment == "debian:12"
|
|
assert network is True
|
|
assert force is False
|
|
return {
|
|
"environment": environment,
|
|
"status": "warm",
|
|
"prepared": True,
|
|
"reused": False,
|
|
"executed": True,
|
|
"network_prepared": True,
|
|
"prepared_at": 123.0,
|
|
"manifest_path": "/cache/.prepare/linux-x86_64/debian_12.json",
|
|
"cache_dir": "/cache",
|
|
"last_prepare_duration_ms": 456,
|
|
"reason": None,
|
|
}
|
|
|
|
class StubPyro:
|
|
def __init__(self) -> None:
|
|
self.manager = StubManager()
|
|
|
|
class StubParser:
|
|
def parse_args(self) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
command="prepare",
|
|
environment="debian:12",
|
|
network=True,
|
|
force=False,
|
|
json=False,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
|
|
monkeypatch.setattr(cli, "Pyro", StubPyro)
|
|
cli.main()
|
|
output = capsys.readouterr().out
|
|
assert "Prepare: debian:12" in output
|
|
assert "Daily loop: WARM" in output
|
|
assert "Result: prepared network_prepared=yes" in output
|
|
|
|
|
|
def test_cli_prepare_prints_json_and_errors(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
class SuccessManager:
|
|
def prepare_daily_loop(
|
|
self,
|
|
environment: str,
|
|
*,
|
|
network: bool,
|
|
force: bool,
|
|
) -> dict[str, object]:
|
|
assert environment == "debian:12"
|
|
assert network is False
|
|
assert force is True
|
|
return {"environment": environment, "reused": True}
|
|
|
|
class SuccessPyro:
|
|
def __init__(self) -> None:
|
|
self.manager = SuccessManager()
|
|
|
|
class SuccessParser:
|
|
def parse_args(self) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
command="prepare",
|
|
environment="debian:12",
|
|
network=False,
|
|
force=True,
|
|
json=True,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: SuccessParser())
|
|
monkeypatch.setattr(cli, "Pyro", SuccessPyro)
|
|
cli.main()
|
|
payload = json.loads(capsys.readouterr().out)
|
|
assert payload["reused"] is True
|
|
|
|
class ErrorManager:
|
|
def prepare_daily_loop(
|
|
self,
|
|
environment: str,
|
|
*,
|
|
network: bool,
|
|
force: bool,
|
|
) -> dict[str, object]:
|
|
del environment, network, force
|
|
raise RuntimeError("prepare failed")
|
|
|
|
class ErrorPyro:
|
|
def __init__(self) -> None:
|
|
self.manager = ErrorManager()
|
|
|
|
class ErrorParser:
|
|
def parse_args(self) -> argparse.Namespace:
|
|
return argparse.Namespace(
|
|
command="prepare",
|
|
environment="debian:12",
|
|
network=False,
|
|
force=False,
|
|
json=True,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: ErrorParser())
|
|
monkeypatch.setattr(cli, "Pyro", ErrorPyro)
|
|
with pytest.raises(SystemExit, match="1"):
|
|
cli.main()
|
|
error_payload = json.loads(capsys.readouterr().out)
|
|
assert error_payload["ok"] is False
|
|
assert error_payload["error"] == "prepare failed"
|
|
|
|
|
|
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, Any] = {}
|
|
|
|
class StubPyro:
|
|
def create_server(
|
|
self,
|
|
*,
|
|
profile: str,
|
|
mode: str | None,
|
|
project_path: str | None,
|
|
repo_url: str | None,
|
|
repo_ref: str | None,
|
|
no_project_source: bool,
|
|
) -> Any:
|
|
observed["profile"] = profile
|
|
observed["mode"] = mode
|
|
observed["project_path"] = project_path
|
|
observed["repo_url"] = repo_url
|
|
observed["repo_ref"] = repo_ref
|
|
observed["no_project_source"] = no_project_source
|
|
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",
|
|
mode=None,
|
|
project_path="/repo",
|
|
repo_url=None,
|
|
repo_ref=None,
|
|
no_project_source=False,
|
|
)
|
|
|
|
monkeypatch.setattr(cli, "_build_parser", lambda: StubParser())
|
|
monkeypatch.setattr(cli, "Pyro", StubPyro)
|
|
cli.main()
|
|
assert observed == {
|
|
"profile": "workspace-core",
|
|
"mode": None,
|
|
"project_path": "/repo",
|
|
"repo_url": None,
|
|
"repo_ref": None,
|
|
"no_project_source": False,
|
|
"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
|