98 lines
2.7 KiB
Python
98 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
import pyro_mcp.demo as demo_module
|
|
|
|
|
|
def test_run_demo_happy_path(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
class StubPyro:
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
def run_in_vm(
|
|
self,
|
|
*,
|
|
environment: str,
|
|
command: str,
|
|
vcpu_count: int,
|
|
mem_mib: int,
|
|
timeout_seconds: int,
|
|
ttl_seconds: int,
|
|
network: bool,
|
|
) -> dict[str, Any]:
|
|
calls.append(
|
|
(
|
|
"run_in_vm",
|
|
{
|
|
"environment": environment,
|
|
"command": command,
|
|
"vcpu_count": vcpu_count,
|
|
"mem_mib": mem_mib,
|
|
"timeout_seconds": timeout_seconds,
|
|
"ttl_seconds": ttl_seconds,
|
|
"network": network,
|
|
},
|
|
)
|
|
)
|
|
return {"vm_id": "vm-1", "stdout": "git version 2.x", "exit_code": 0}
|
|
|
|
monkeypatch.setattr(demo_module, "Pyro", StubPyro)
|
|
result = demo_module.run_demo()
|
|
|
|
assert result["exit_code"] == 0
|
|
assert calls == [
|
|
(
|
|
"run_in_vm",
|
|
{
|
|
"environment": "debian:12",
|
|
"command": "git --version",
|
|
"vcpu_count": 1,
|
|
"mem_mib": 1024,
|
|
"timeout_seconds": 30,
|
|
"ttl_seconds": 600,
|
|
"network": False,
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
def test_demo_command_prefers_network_probe_for_guest_vsock() -> None:
|
|
status = {"network_enabled": True, "execution_mode": "guest_vsock"}
|
|
assert "https://example.com" in demo_module._demo_command(status)
|
|
|
|
|
|
def test_main_prints_json(
|
|
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
demo_module,
|
|
"run_demo",
|
|
lambda: {"stdout": "git version 2.x", "exit_code": 0},
|
|
)
|
|
demo_module.main()
|
|
rendered = json.loads(capsys.readouterr().out)
|
|
assert rendered["exit_code"] == 0
|
|
|
|
|
|
def test_run_demo_network_uses_probe(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured: dict[str, Any] = {}
|
|
|
|
class StubPyro:
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
def run_in_vm(self, **kwargs: Any) -> dict[str, Any]:
|
|
captured.update(kwargs)
|
|
return {"exit_code": 0}
|
|
|
|
monkeypatch.setattr(demo_module, "Pyro", StubPyro)
|
|
demo_module.run_demo(network=True)
|
|
assert "https://example.com" in str(captured["command"])
|
|
assert captured["network"] is True
|
|
assert captured["mem_mib"] == 1024
|