thaloco/pyro-mcp
1
Fork 0
Code Pull requests Releases Packages Activity Actions

Add host bootstrap and repair helpers

Browse source 
Add a dedicated pyro host surface for supported chat hosts so Claude Code, Codex, and OpenCode users can connect or repair the canonical MCP setup without hand-writing raw commands or config edits.

Implement the shared host helper layer and wire it through the CLI with connect, print-config, doctor, and repair, all generated from the same canonical pyro mcp serve command shape and project-source flags. Update the docs, public contract, examples, changelog, and roadmap so the helper flow becomes the primary onramp while raw host-specific commands remain as reference material.

Harden the verification path that this milestone exposed: temp git repos in tests now disable commit signing, socket-based port tests skip cleanly when the sandbox forbids those primitives, and make test still uses multiple cores by default but caps xdist workers to a stable value so make check stays fast and deterministic here.

Validation:
- uv lock
- UV_OFFLINE=1 UV_CACHE_DIR=.uv-cache make check
- UV_OFFLINE=1 UV_CACHE_DIR=.uv-cache make dist-check
This commit is contained in:
Thales Maciel 2026-03-13 16:46:10 -03:00
parent 535efc6919
commit 899a6760c4
25 changed files with 1658 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tests/test_api.py
View file

@ -18,7 +18,7 @@ from pyro_mcp.vm_network import TapNetworkManager
def _git(repo: Path, *args: str) -> str:
result = subprocess.run( # noqa: S603
["git", *args],
["git", "-c", "commit.gpgsign=false", *args],
cwd=repo,
check=True,
capture_output=True,

151
tests/test_cli.py
View file

@ -9,6 +9,7 @@ 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:
@ -31,10 +32,11 @@ def test_cli_help_guides_first_run() -> None:
assert "pyro env list" in help_text
assert "pyro env pull debian:12" in help_text
assert "pyro run debian:12 -- git --version" in help_text
assert "pyro mcp serve" in help_text
assert "pyro host connect claude-code" in help_text
assert "Connect a chat host after that:" in help_text
assert "claude mcp add pyro -- uvx --from pyro-mcp pyro mcp serve" in help_text
assert "codex mcp add pyro -- uvx --from pyro-mcp pyro mcp serve" 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 "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 snapshot create WORKSPACE_ID checkpoint" in help_text
@ -57,6 +59,31 @@ def test_cli_subcommand_help_includes_examples_and_guidance() -> None:
assert "pyro env pull debian:12" in env_help
assert "downloads from public Docker Hub" in env_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 "pyro doctor --json" in doctor_help
@ -316,6 +343,94 @@ def test_cli_subcommand_help_includes_examples_and_guidance() -> None:
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],
@ -2823,30 +2938,38 @@ def test_chat_host_docs_and_examples_recommend_workspace_core() -> None:
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"
codex_helper = "pyro host connect codex"
opencode_helper = "pyro host print-config opencode"
claude_cmd = "claude mcp add pyro -- uvx --from pyro-mcp pyro mcp serve"
codex_cmd = "codex mcp add pyro -- uvx --from pyro-mcp pyro mcp serve"
assert "## Chat Host Quickstart" in readme
assert "uvx --from pyro-mcp pyro mcp serve" in readme
assert claude_cmd in readme
assert codex_cmd in readme
assert claude_helper in readme
assert codex_helper in readme
assert opencode_helper in readme
assert "examples/opencode_mcp_config.json" in readme
assert "pyro host doctor" in readme
assert "bare `pyro mcp serve` starts `workspace-core`" in readme
assert "auto-detects\nthe current Git checkout" 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 "## 5. Connect a chat host" in install
assert "uvx --from pyro-mcp pyro mcp serve" in install
assert claude_cmd in install
assert codex_cmd in install
assert claude_helper in install
assert codex_helper in install
assert opencode_helper in install
assert "workspace-full" in install
assert "--project-path /abs/path/to/repo" in install
assert claude_cmd in first_run
assert codex_cmd in first_run
assert claude_helper in first_run
assert codex_helper in first_run
assert opencode_helper in first_run
assert "--project-path /abs/path/to/repo" in first_run
assert claude_helper in integrations
assert codex_helper in integrations
assert opencode_helper in integrations
assert "Bare `pyro mcp serve` starts `workspace-core`." in integrations
assert "auto-detects the current Git checkout" in integrations
assert "examples/claude_code_mcp.md" in integrations
@ -2862,13 +2985,17 @@ def test_chat_host_docs_and_examples_recommend_workspace_core() -> None:
assert "codex_mcp.md" in mcp_config
assert "opencode_mcp_config.json" 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" 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" in codex
assert "workspace-full" in codex
assert "--project-path /abs/path/to/repo" in codex

484
tests/test_host_helpers.py Normal file
View file

@ -0,0 +1,484 @@
from __future__ import annotations
import json
import shutil
import sys
from pathlib import Path
from subprocess import CompletedProcess
import pytest
import pyro_mcp.host_helpers as host_helpers
from pyro_mcp.host_helpers import (
DEFAULT_OPENCODE_CONFIG_PATH,
HostServerConfig,
_canonical_server_command,
_command_matches,
_repair_command,
connect_cli_host,
doctor_hosts,
print_or_write_opencode_config,
repair_host,
)
def _install_fake_mcp_cli(tmp_path: Path, name: str) -> tuple[Path, Path]:
bin_dir = tmp_path / "bin"
bin_dir.mkdir(parents=True)
state_path = tmp_path / f"{name}-state.json"
script_path = bin_dir / name
script_path.write_text(
"\n".join(
[
f"#!{sys.executable}",
"import json",
"import shlex",
"import sys",
f"STATE_PATH = {str(state_path)!r}",
"try:",
" with open(STATE_PATH, 'r', encoding='utf-8') as handle:",
" state = json.load(handle)",
"except FileNotFoundError:",
" state = {}",
"args = sys.argv[1:]",
"if args[:2] == ['mcp', 'add']:",
" name = args[2]",
" marker = args.index('--')",
" state[name] = args[marker + 1:]",
" with open(STATE_PATH, 'w', encoding='utf-8') as handle:",
" json.dump(state, handle)",
" print(f'added {name}')",
" raise SystemExit(0)",
"if args[:2] == ['mcp', 'remove']:",
" name = args[2]",
" if name in state:",
" del state[name]",
" with open(STATE_PATH, 'w', encoding='utf-8') as handle:",
" json.dump(state, handle)",
" print(f'removed {name}')",
" raise SystemExit(0)",
" print('not found', file=sys.stderr)",
" raise SystemExit(1)",
"if args[:2] == ['mcp', 'get']:",
" name = args[2]",
" if name not in state:",
" print('not found', file=sys.stderr)",
" raise SystemExit(1)",
" print(f'{name}: {shlex.join(state[name])}')",
" raise SystemExit(0)",
"if args[:2] == ['mcp', 'list']:",
" for item in sorted(state):",
" print(item)",
" raise SystemExit(0)",
"print('unsupported', file=sys.stderr)",
"raise SystemExit(2)",
]
),
encoding="utf-8",
)
script_path.chmod(0o755)
return bin_dir, state_path
def test_connect_cli_host_replaces_existing_entry(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
bin_dir, state_path = _install_fake_mcp_cli(tmp_path, "codex")
state_path.write_text(json.dumps({"pyro": ["old", "command"]}), encoding="utf-8")
monkeypatch.setenv("PATH", str(bin_dir))
payload = connect_cli_host("codex", config=HostServerConfig())
assert payload["host"] == "codex"
assert payload["server_command"] == ["uvx", "--from", "pyro-mcp", "pyro", "mcp", "serve"]
assert json.loads(state_path.read_text(encoding="utf-8")) == {
"pyro": ["uvx", "--from", "pyro-mcp", "pyro", "mcp", "serve"]
}
def test_canonical_server_command_validates_and_renders_variants() -> None:
assert _canonical_server_command(HostServerConfig(installed_package=True)) == [
"pyro",
"mcp",
"serve",
]
assert _canonical_server_command(
HostServerConfig(profile="workspace-full", project_path="/repo")
) == [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
"--profile",
"workspace-full",
"--project-path",
"/repo",
]
assert _canonical_server_command(
HostServerConfig(repo_url="https://example.com/repo.git", repo_ref="main")
) == [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
"--repo-url",
"https://example.com/repo.git",
"--repo-ref",
"main",
]
assert _canonical_server_command(HostServerConfig(no_project_source=True)) == [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
"--no-project-source",
]
with pytest.raises(ValueError, match="mutually exclusive"):
_canonical_server_command(
HostServerConfig(project_path="/repo", repo_url="https://example.com/repo.git")
)
with pytest.raises(ValueError, match="cannot be combined"):
_canonical_server_command(HostServerConfig(project_path="/repo", no_project_source=True))
with pytest.raises(ValueError, match="requires --repo-url"):
_canonical_server_command(HostServerConfig(repo_ref="main"))
def test_repair_command_and_command_matches_cover_edge_cases() -> None:
assert _repair_command("codex", HostServerConfig()) == "pyro host repair codex"
assert _repair_command("codex", HostServerConfig(project_path="/repo")) == (
"pyro host repair codex --project-path /repo"
)
assert _repair_command(
"opencode",
HostServerConfig(installed_package=True, profile="workspace-full", repo_url="file:///repo"),
config_path=Path("/tmp/opencode.json"),
) == (
"pyro host repair opencode --installed-package --profile workspace-full "
"--repo-url file:///repo --config-path /tmp/opencode.json"
)
assert _repair_command("codex", HostServerConfig(no_project_source=True)) == (
"pyro host repair codex --no-project-source"
)
assert _command_matches(
"pyro: uvx --from pyro-mcp pyro mcp serve",
["uvx", "--from", "pyro-mcp", "pyro", "mcp", "serve"],
)
assert _command_matches(
'"uvx --from pyro-mcp pyro mcp serve',
['"uvx', "--from", "pyro-mcp", "pyro", "mcp", "serve"],
)
assert not _command_matches("pyro: uvx --from pyro-mcp pyro mcp serve --profile vm-run", [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
])
def test_connect_cli_host_reports_missing_cli_and_add_failure(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
with pytest.raises(ValueError, match="unsupported CLI host"):
connect_cli_host("unsupported", config=HostServerConfig())
monkeypatch.setenv("PATH", "")
with pytest.raises(RuntimeError, match="codex CLI is not installed"):
connect_cli_host("codex", config=HostServerConfig())
bin_dir = tmp_path / "bin"
bin_dir.mkdir()
script_path = bin_dir / "codex"
script_path.write_text(
"\n".join(
[
f"#!{sys.executable}",
"import sys",
"raise SystemExit(1 if sys.argv[1:3] == ['mcp', 'add'] else 0)",
]
),
encoding="utf-8",
)
script_path.chmod(0o755)
monkeypatch.setenv("PATH", str(bin_dir))
with pytest.raises(RuntimeError, match="codex mcp add failed"):
connect_cli_host("codex", config=HostServerConfig())
def test_doctor_hosts_reports_ok_and_drifted(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
codex_bin, codex_state = _install_fake_mcp_cli(tmp_path / "codex", "codex")
claude_bin, claude_state = _install_fake_mcp_cli(tmp_path / "claude", "claude")
combined_path = str(codex_bin) + ":" + str(claude_bin)
monkeypatch.setenv("PATH", combined_path)
codex_state.write_text(
json.dumps({"pyro": ["uvx", "--from", "pyro-mcp", "pyro", "mcp", "serve"]}),
encoding="utf-8",
)
claude_state.write_text(
json.dumps(
{
"pyro": [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
"--profile",
"workspace-full",
]
}
),
encoding="utf-8",
)
opencode_config = tmp_path / "opencode.json"
opencode_config.write_text(
json.dumps(
{
"mcp": {
"pyro": {
"type": "local",
"enabled": True,
"command": ["uvx", "--from", "pyro-mcp", "pyro", "mcp", "serve"],
}
}
}
),
encoding="utf-8",
)
entries = doctor_hosts(config=HostServerConfig(), config_path=opencode_config)
by_host = {entry.host: entry for entry in entries}
assert by_host["codex"].status == "ok"
assert by_host["codex"].configured is True
assert by_host["claude-code"].status == "drifted"
assert by_host["claude-code"].configured is True
assert by_host["opencode"].status == "ok"
assert by_host["opencode"].configured is True
def test_doctor_hosts_reports_missing_and_drifted_opencode_shapes(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setenv("PATH", "")
config_path = tmp_path / "opencode.json"
config_path.write_text("[]", encoding="utf-8")
entries = doctor_hosts(config=HostServerConfig(), config_path=config_path)
by_host = {entry.host: entry for entry in entries}
assert by_host["opencode"].status == "unavailable"
assert "JSON object" in by_host["opencode"].details
config_path.write_text(json.dumps({"mcp": {}}), encoding="utf-8")
entries = doctor_hosts(config=HostServerConfig(), config_path=config_path)
by_host = {entry.host: entry for entry in entries}
assert by_host["opencode"].status == "unavailable"
assert "missing mcp.pyro" in by_host["opencode"].details
config_path.write_text(
json.dumps({"mcp": {"pyro": {"type": "local", "enabled": False, "command": ["wrong"]}}}),
encoding="utf-8",
)
entries = doctor_hosts(config=HostServerConfig(), config_path=config_path)
by_host = {entry.host: entry for entry in entries}
assert by_host["opencode"].status == "drifted"
assert by_host["opencode"].configured is True
def test_doctor_hosts_reports_invalid_json_for_installed_opencode(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
config_path = tmp_path / "opencode.json"
config_path.write_text("{invalid", encoding="utf-8")
monkeypatch.setattr(
shutil,
"which",
lambda name: "/usr/bin/opencode" if name == "opencode" else None,
)
entries = doctor_hosts(config=HostServerConfig(), config_path=config_path)
by_host = {entry.host: entry for entry in entries}
assert by_host["opencode"].status == "drifted"
assert "invalid JSON" in by_host["opencode"].details
def test_repair_opencode_preserves_unrelated_keys(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setenv("PATH", "")
config_path = tmp_path / "opencode.json"
config_path.write_text(
json.dumps({"theme": "light", "mcp": {"other": {"type": "local"}}}),
encoding="utf-8",
)
payload = repair_host("opencode", config=HostServerConfig(), config_path=config_path)
assert payload["config_path"] == str(config_path.resolve())
repaired = json.loads(config_path.read_text(encoding="utf-8"))
assert repaired["theme"] == "light"
assert repaired["mcp"]["other"] == {"type": "local"}
assert repaired["mcp"]["pyro"] == {
"type": "local",
"enabled": True,
"command": ["uvx", "--from", "pyro-mcp", "pyro", "mcp", "serve"],
}
def test_repair_opencode_backs_up_non_object_json(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setenv("PATH", "")
config_path = tmp_path / "opencode.json"
config_path.write_text("[]", encoding="utf-8")
payload = repair_host("opencode", config=HostServerConfig(), config_path=config_path)
backup_path = Path(str(payload["backup_path"]))
assert backup_path.exists()
assert backup_path.read_text(encoding="utf-8") == "[]"
def test_repair_opencode_backs_up_invalid_json(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setenv("PATH", "")
config_path = tmp_path / "opencode.json"
config_path.write_text("{invalid", encoding="utf-8")
payload = repair_host("opencode", config=HostServerConfig(), config_path=config_path)
backup_path = Path(str(payload["backup_path"]))
assert backup_path.exists()
assert backup_path.read_text(encoding="utf-8") == "{invalid"
repaired = json.loads(config_path.read_text(encoding="utf-8"))
assert repaired["mcp"]["pyro"]["command"] == [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
]
def test_print_or_write_opencode_config_writes_json(tmp_path: Path) -> None:
output_path = tmp_path / "opencode.json"
payload = print_or_write_opencode_config(
config=HostServerConfig(project_path="/repo"),
output_path=output_path,
)
assert payload["output_path"] == str(output_path)
rendered = json.loads(output_path.read_text(encoding="utf-8"))
assert rendered == {
"mcp": {
"pyro": {
"type": "local",
"enabled": True,
"command": [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
"--project-path",
"/repo",
],
}
}
}
def test_print_or_write_opencode_config_returns_rendered_text() -> None:
payload = print_or_write_opencode_config(config=HostServerConfig(profile="vm-run"))
assert payload["host"] == "opencode"
assert payload["server_command"] == [
"uvx",
"--from",
"pyro-mcp",
"pyro",
"mcp",
"serve",
"--profile",
"vm-run",
]
rendered = str(payload["rendered_config"])
assert '"type": "local"' in rendered
assert '"command": [' in rendered
def test_doctor_reports_opencode_missing_when_config_absent(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setenv("PATH", "")
entries = doctor_hosts(
config=HostServerConfig(),
config_path=tmp_path / "missing-opencode.json",
)
by_host = {entry.host: entry for entry in entries}
assert by_host["opencode"].status == "unavailable"
assert str(DEFAULT_OPENCODE_CONFIG_PATH) not in by_host["opencode"].details
def test_repair_host_delegates_non_opencode_and_doctor_handles_list_only_configured(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
host_helpers,
"connect_cli_host",
lambda host, *, config: {"host": host, "profile": config.profile},
)
assert repair_host("codex", config=HostServerConfig(profile="vm-run")) == {
"host": "codex",
"profile": "vm-run",
}
commands: list[list[str]] = []
def _fake_run_command(command: list[str]) -> CompletedProcess[str]:
commands.append(command)
if command[:3] == ["codex", "mcp", "get"]:
return CompletedProcess(command, 1, "", "not found")
if command[:3] == ["codex", "mcp", "list"]:
return CompletedProcess(command, 0, "pyro\n", "")
raise AssertionError(command)
monkeypatch.setattr(
shutil,
"which",
lambda name: "/usr/bin/codex" if name == "codex" else None,
)
monkeypatch.setattr(host_helpers, "_run_command", _fake_run_command)
entry = host_helpers._doctor_cli_host("codex", config=HostServerConfig())
assert entry.status == "drifted"
assert entry.configured is True
assert commands == [["codex", "mcp", "get", "pyro"], ["codex", "mcp", "list"]]

123
tests/test_project_startup.py
View file

@ -5,6 +5,7 @@ from pathlib import Path
import pytest
import pyro_mcp.project_startup as project_startup
from pyro_mcp.project_startup import (
ProjectStartupSource,
describe_project_startup_source,
@ -15,7 +16,7 @@ from pyro_mcp.project_startup import (
def _git(repo: Path, *args: str) -> str:
result = subprocess.run( # noqa: S603
["git", *args],
["git", "-c", "commit.gpgsign=false", *args],
cwd=repo,
check=True,
capture_output=True,
@ -76,6 +77,52 @@ def test_resolve_project_startup_source_validates_flag_combinations(tmp_path: Pa
resolve_project_startup_source(project_path=repo, no_project_source=True)
def test_resolve_project_startup_source_handles_explicit_none_and_empty_values(
tmp_path: Path,
) -> None:
repo = _make_repo(tmp_path / "repo")
outside = tmp_path / "outside"
outside.mkdir()
assert resolve_project_startup_source(no_project_source=True, cwd=tmp_path) is None
assert resolve_project_startup_source(cwd=outside) is None
with pytest.raises(ValueError, match="must not be empty"):
resolve_project_startup_source(repo_url=" ", cwd=repo)
with pytest.raises(ValueError, match="must not be empty"):
resolve_project_startup_source(repo_url="https://example.com/repo.git", repo_ref=" ")
def test_resolve_project_startup_source_rejects_missing_or_non_directory_project_path(
tmp_path: Path,
) -> None:
missing = tmp_path / "missing"
file_path = tmp_path / "note.txt"
file_path.write_text("hello\n", encoding="utf-8")
with pytest.raises(ValueError, match="does not exist"):
resolve_project_startup_source(project_path=missing, cwd=tmp_path)
with pytest.raises(ValueError, match="must be a directory"):
resolve_project_startup_source(project_path=file_path, cwd=tmp_path)
def test_resolve_project_startup_source_keeps_plain_relative_directory_when_not_a_repo(
tmp_path: Path,
) -> None:
plain = tmp_path / "plain"
plain.mkdir()
resolved = resolve_project_startup_source(project_path="plain", cwd=tmp_path)
assert resolved == ProjectStartupSource(
kind="project_path",
origin_ref=str(plain.resolve()),
resolved_path=plain.resolve(),
)
def test_materialize_project_startup_source_clones_local_repo_url_at_ref(tmp_path: Path) -> None:
repo = _make_repo(tmp_path / "repo", content="one\n")
first_commit = _git(repo, "rev-parse", "HEAD")
@ -93,6 +140,60 @@ def test_materialize_project_startup_source_clones_local_repo_url_at_ref(tmp_pat
assert (clone_dir / "note.txt").read_text(encoding="utf-8") == "one\n"
def test_materialize_project_startup_source_validates_project_source_and_clone_failures(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
with pytest.raises(RuntimeError, match="missing a resolved path"):
with materialize_project_startup_source(
ProjectStartupSource(kind="project_path", origin_ref="/repo", resolved_path=None)
):
pass
source = ProjectStartupSource(kind="repo_url", origin_ref="https://example.com/repo.git")
def _clone_failure(
command: list[str],
*,
cwd: Path | None = None,
) -> subprocess.CompletedProcess[str]:
del cwd
return subprocess.CompletedProcess(command, 1, "", "clone failed")
monkeypatch.setattr("pyro_mcp.project_startup._run_git", _clone_failure)
with pytest.raises(RuntimeError, match="failed to clone repo_url"):
with materialize_project_startup_source(source):
pass
def test_materialize_project_startup_source_reports_checkout_failure(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
repo = _make_repo(tmp_path / "repo", content="one\n")
source = ProjectStartupSource(
kind="repo_url",
origin_ref=str(repo.resolve()),
repo_ref="missing-ref",
)
original_run_git = project_startup._run_git
def _checkout_failure(
command: list[str],
*,
cwd: Path | None = None,
) -> subprocess.CompletedProcess[str]:
if command[:2] == ["git", "checkout"]:
return subprocess.CompletedProcess(command, 1, "", "checkout failed")
return original_run_git(command, cwd=cwd)
monkeypatch.setattr("pyro_mcp.project_startup._run_git", _checkout_failure)
with pytest.raises(RuntimeError, match="failed to checkout repo_ref"):
with materialize_project_startup_source(source):
pass
def test_describe_project_startup_source_formats_project_and_repo_sources(tmp_path: Path) -> None:
repo = _make_repo(tmp_path / "repo")
@ -113,3 +214,23 @@ def test_describe_project_startup_source_formats_project_and_repo_sources(tmp_pa
assert project_description == f"the current project at {repo.resolve()}"
assert repo_description == "the clean clone source https://example.com/repo.git at ref main"
def test_describe_project_startup_source_handles_none_and_repo_without_ref() -> None:
assert describe_project_startup_source(None) is None
assert (
describe_project_startup_source(
ProjectStartupSource(kind="repo_url", origin_ref="https://example.com/repo.git")
)
== "the clean clone source https://example.com/repo.git"
)
def test_detect_git_root_returns_none_for_empty_stdout(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
project_startup,
"_run_git",
lambda command, *, cwd=None: subprocess.CompletedProcess(command, 0, "\n", ""),
)
assert project_startup._detect_git_root(Path.cwd()) is None

28
tests/test_public_contract.py
View file

@ -16,6 +16,11 @@ from pyro_mcp.contract import (
PUBLIC_CLI_COMMANDS,
PUBLIC_CLI_DEMO_SUBCOMMANDS,
PUBLIC_CLI_ENV_SUBCOMMANDS,
PUBLIC_CLI_HOST_CONNECT_FLAGS,
PUBLIC_CLI_HOST_DOCTOR_FLAGS,
PUBLIC_CLI_HOST_PRINT_CONFIG_FLAGS,
PUBLIC_CLI_HOST_REPAIR_FLAGS,
PUBLIC_CLI_HOST_SUBCOMMANDS,
PUBLIC_CLI_MCP_SERVE_FLAGS,
PUBLIC_CLI_MCP_SUBCOMMANDS,
PUBLIC_CLI_RUN_FLAGS,
@ -102,6 +107,29 @@ def test_public_cli_help_lists_commands_and_run_flags() -> None:
env_help_text = _subparser_choice(parser, "env").format_help()
for subcommand_name in PUBLIC_CLI_ENV_SUBCOMMANDS:
assert subcommand_name in env_help_text
host_help_text = _subparser_choice(parser, "host").format_help()
for subcommand_name in PUBLIC_CLI_HOST_SUBCOMMANDS:
assert subcommand_name in host_help_text
host_connect_help_text = _subparser_choice(
_subparser_choice(parser, "host"), "connect"
).format_help()
for flag in PUBLIC_CLI_HOST_CONNECT_FLAGS:
assert flag in host_connect_help_text
host_doctor_help_text = _subparser_choice(
_subparser_choice(parser, "host"), "doctor"
).format_help()
for flag in PUBLIC_CLI_HOST_DOCTOR_FLAGS:
assert flag in host_doctor_help_text
host_print_config_help_text = _subparser_choice(
_subparser_choice(parser, "host"), "print-config"
).format_help()
for flag in PUBLIC_CLI_HOST_PRINT_CONFIG_FLAGS:
assert flag in host_print_config_help_text
host_repair_help_text = _subparser_choice(
_subparser_choice(parser, "host"), "repair"
).format_help()
for flag in PUBLIC_CLI_HOST_REPAIR_FLAGS:
assert flag in host_repair_help_text
mcp_help_text = _subparser_choice(parser, "mcp").format_help()
for subcommand_name in PUBLIC_CLI_MCP_SUBCOMMANDS:
assert subcommand_name in mcp_help_text

2
tests/test_server.py
View file

@ -19,7 +19,7 @@ from pyro_mcp.vm_network import TapNetworkManager
def _git(repo: Path, *args: str) -> str:
result = subprocess.run( # noqa: S603
["git", *args],
["git", "-c", "commit.gpgsign=false", *args],
cwd=repo,
check=True,
capture_output=True,

48
tests/test_workspace_ports.py
View file

@ -15,6 +15,13 @@ import pytest
from pyro_mcp import workspace_ports
def _socketpair_or_skip() -> tuple[socket.socket, socket.socket]:
try:
return socket.socketpair()
except PermissionError as exc:
pytest.skip(f"socketpair unavailable in this environment: {exc}")
class _EchoHandler(socketserver.BaseRequestHandler):
def handle(self) -> None:
data = self.request.recv(65536)
@ -50,18 +57,26 @@ def test_workspace_port_proxy_handler_ignores_upstream_connect_failure(
def test_workspace_port_proxy_forwards_tcp_traffic() -> None:
upstream = socketserver.ThreadingTCPServer(
(workspace_ports.DEFAULT_PUBLISHED_PORT_HOST, 0),
_EchoHandler,
)
try:
upstream = socketserver.ThreadingTCPServer(
(workspace_ports.DEFAULT_PUBLISHED_PORT_HOST, 0),
_EchoHandler,
)
except PermissionError as exc:
pytest.skip(f"TCP bind unavailable in this environment: {exc}")
upstream_thread = threading.Thread(target=upstream.serve_forever, daemon=True)
upstream_thread.start()
upstream_host = str(upstream.server_address[0])
upstream_port = int(upstream.server_address[1])
proxy = workspace_ports._ProxyServer( # noqa: SLF001
(workspace_ports.DEFAULT_PUBLISHED_PORT_HOST, 0),
(upstream_host, upstream_port),
)
try:
proxy = workspace_ports._ProxyServer( # noqa: SLF001
(workspace_ports.DEFAULT_PUBLISHED_PORT_HOST, 0),
(upstream_host, upstream_port),
)
except PermissionError as exc:
upstream.shutdown()
upstream.server_close()
pytest.skip(f"proxy TCP bind unavailable in this environment: {exc}")
proxy_thread = threading.Thread(target=proxy.serve_forever, daemon=True)
proxy_thread.start()
try:
@ -202,8 +217,8 @@ def test_workspace_ports_main_shutdown_handler_stops_server(
def test_workspace_port_proxy_handler_handles_empty_and_invalid_selector_events(
monkeypatch: Any,
) -> None:
source, source_peer = socket.socketpair()
upstream, upstream_peer = socket.socketpair()
source, source_peer = _socketpair_or_skip()
upstream, upstream_peer = _socketpair_or_skip()
source_peer.close()
class FakeSelector:
@ -246,10 +261,17 @@ def test_workspace_port_proxy_handler_handles_recv_and_send_errors(
monkeypatch: Any,
) -> None:
def _run_once(*, close_source: bool) -> None:
source, source_peer = socket.socketpair()
upstream, upstream_peer = socket.socketpair()
source, source_peer = _socketpair_or_skip()
upstream, upstream_peer = _socketpair_or_skip()
if not close_source:
source_peer.sendall(b"hello")
try:
source_peer.sendall(b"hello")
except PermissionError as exc:
source.close()
source_peer.close()
upstream.close()
upstream_peer.close()
pytest.skip(f"socket send unavailable in this environment: {exc}")
class FakeSelector:
def register(self, *_args: Any, **_kwargs: Any) -> None: