Files
stem-launcher/tests/test_stemctl.py
T

283 lines
12 KiB
Python

from __future__ import annotations
import json
import io
import os
import subprocess
import tempfile
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import rvctl # noqa: E402
class StemctlContractTests(unittest.TestCase):
def test_new_command_help_never_crashes(self) -> None:
script = Path(__file__).resolve().parents[1] / "rvctl.py"
commands = ["env", "build", "test", "run", "debug", "deploy", "shell", "start", "status", "attach", "stop", "rm", "probe", "mcp"]
for command in commands:
with self.subTest(command=command):
completed = subprocess.run(
[sys.executable, str(script), command, "--help"],
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
self.assertEqual(completed.returncode, 0, completed.stderr)
def test_profile_aliases_are_canonical(self) -> None:
self.assertEqual(rvctl.normalize_env_profile("host"), "native-amd64")
self.assertEqual(rvctl.normalize_env_profile("rv32i"), "hazard3-sim")
self.assertEqual(rvctl.normalize_env_profile("rp2350"), "rp2350")
def test_instance_is_stable_across_actions(self) -> None:
debug = rvctl.default_env_instance("hazard3-sim", "inf", "bss", "task1", "debug")
stop = rvctl.default_env_instance("hazard3-sim", "inf", "bss", "task1", "stop")
self.assertEqual(debug, stop)
self.assertEqual(debug, "hazard3-sim-inf-bss-t1")
def test_lifecycle_with_explicit_instance_needs_no_card_or_defaults(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
workspace = root / "workspace"
workspace.mkdir()
tools = root / "tools"
tools.mkdir()
(tools / "stem").write_text("#!/bin/sh\n", encoding="utf-8")
config_path = root / "workspace.json"
config_path.write_text(
json.dumps(
{
"workspace_root": str(workspace),
"original_root": str(root / "original"),
"tools_root_candidates": [str(tools)],
}
),
encoding="utf-8",
)
config = rvctl.load_config(config_path)
for action in ("status", "attach", "stop", "rm"):
with self.subTest(action=action):
args = SimpleNamespace(
profile="hazard3-sim",
selector=[],
pane=None,
task=None,
instance="lesson-1",
editor=None,
target=None,
device=None,
backend=None,
allow_existing_firmware=False,
dry_run=True,
)
output = io.StringIO()
with (
mock.patch.object(rvctl, "resolve_env_target") as resolve,
redirect_stdout(output),
):
rvctl.run_profile_action(config, args, action)
resolve.assert_not_called()
rendered = output.getvalue()
self.assertIn("instance\tlesson-1", rendered)
self.assertIn(f"./stem {action} --target hazard3-baremetal", rendered)
def test_tmux_container_delegates_to_canonical_rootless_runtime(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
workspace = root / "workspace"
card = workspace / "series" / "inf" / "card"
(card / "src/tasks").mkdir(parents=True)
(card / "src/tasks/task1.c").write_text("int main(void) { return 0; }\n", encoding="utf-8")
tools = root / "tools"
tools.mkdir()
(tools / "stem").write_text("#!/bin/sh\n", encoding="utf-8")
config_path = root / "workspace.json"
config_path.write_text(
json.dumps(
{
"workspace_root": str(workspace),
"original_root": str(root / "original"),
"tools_root_candidates": [str(tools)],
"defaults": {"series": "inf", "card": "card", "task": "task1"},
}
),
encoding="utf-8",
)
config = rvctl.load_config(config_path)
args = SimpleNamespace(
series="inf",
card="card",
profile="hazard3-sim",
session="test-shell",
window="shell",
instance=None,
attach=False,
dry_run=True,
)
output = io.StringIO()
with (
mock.patch.object(rvctl, "tmux_target_exists", return_value=False),
mock.patch.object(rvctl, "resolve_workspace_card_path", return_value=card),
redirect_stdout(output),
):
rvctl.run_tmux_container(config, args)
rendered = output.getvalue()
self.assertIn("./stem shell", rendered)
self.assertIn(f"STEM_REPO_PATH={card}", rendered)
self.assertIn("STEM_SOCKET_ROOT=", rendered)
self.assertNotIn("docker compose", rendered)
def test_workspace_doctor_rejects_docker_runtime(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
workspace = root / "workspace"
(workspace / "series").mkdir(parents=True)
tools = root / "tools"
tools.mkdir()
(tools / "stem").write_text("#!/bin/sh\n", encoding="utf-8")
config_path = root / "workspace.json"
config_path.write_text(
json.dumps(
{
"workspace_root": str(workspace),
"original_root": str(root / "original"),
"tools_root_candidates": [str(tools)],
}
),
encoding="utf-8",
)
config = rvctl.load_config(config_path)
output = io.StringIO()
with mock.patch.dict(os.environ, {"STEM_RUNTIME": "docker"}), redirect_stdout(output):
with self.assertRaisesRegex(SystemExit, "1"):
rvctl.run_workspace(config, SimpleNamespace(workspace_command="doctor"))
rendered = output.getvalue()
self.assertIn("runtime\tFAIL\tdocker is not supported", rendered)
self.assertIn("rootless\tFAIL\trootless Podman is required", rendered)
def test_existing_legacy_workspace_is_detected(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
legacy = root / "rv"
legacy.mkdir()
config_path = root / "workspace.json"
config_path.write_text(
json.dumps(
{
"workspace_root": str(root / "stem"),
"legacy_workspace_root": str(legacy),
"original_root": str(root / "original"),
}
),
encoding="utf-8",
)
with mock.patch.dict(os.environ, {}, clear=False):
os.environ.pop("STEM_WORKSPACE_ROOT", None)
config = rvctl.load_config(config_path)
self.assertEqual(config.workspace_root, legacy.resolve())
self.assertEqual(config.series_root, (legacy / "series").resolve())
def test_xdg_socket_path_stays_inside_budget(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
with mock.patch.dict(os.environ, {"XDG_RUNTIME_DIR": str(root)}):
socket_root = rvctl.default_runtime_socket_root(root / "workspace")
path = socket_root / ("t" * 12) / ("i" * 12) / ("c" * 12) / "n.sock"
self.assertLessEqual(len(os.fsencode(path)), 100)
def test_action_exports_new_and_legacy_contract(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
workspace = root / "workspace"
card = workspace / "series" / "inf" / "card"
card.mkdir(parents=True)
tools = root / "tools"
tools.mkdir()
(tools / "stem").write_text("#!/bin/sh\n", encoding="utf-8")
config_path = root / "workspace.json"
config_path.write_text(
json.dumps(
{
"workspace_root": str(workspace),
"original_root": str(root / "original"),
"tools_root_candidates": [str(tools)],
}
),
encoding="utf-8",
)
config = rvctl.load_config(config_path)
with mock.patch.object(rvctl, "resolve_workspace_card_path", return_value=card):
command, details = rvctl.env_action_command(
config,
"rv32i",
"test",
"inf",
"card",
"task1",
target="hazard3-baremetal",
)
self.assertEqual(details["profile"], "hazard3-sim")
self.assertIn("STEM_PROFILE=hazard3-sim", command)
self.assertIn("RV_PROFILE=hazard3-sim", command)
self.assertIn("./stem test --target hazard3-baremetal task1", command)
def test_deploy_passes_explicit_device_and_backend(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
workspace = root / "workspace"
card = workspace / "series" / "inf" / "card"
card.mkdir(parents=True)
tools = root / "tools"
tools.mkdir()
(tools / "stem").write_text("#!/bin/sh\n", encoding="utf-8")
config_path = root / "workspace.json"
config_path.write_text(
json.dumps(
{
"workspace_root": str(workspace),
"original_root": str(root / "original"),
"tools_root_candidates": [str(tools)],
}
),
encoding="utf-8",
)
config = rvctl.load_config(config_path)
with mock.patch.object(rvctl, "resolve_workspace_card_path", return_value=card):
command, details = rvctl.env_action_command(
config,
"rp2350",
"deploy",
"inf",
"card",
"task1",
target="rp2350-rv",
device="/dev/bus/usb/001/006",
deploy_backend="probe",
)
self.assertEqual(details["device"], "/dev/bus/usb/001/006")
self.assertEqual(details["backend"], "probe")
self.assertIn("--device /dev/bus/usb/001/006", command)
self.assertIn("--backend probe", command)
def test_env_build_uses_source_profile_command(self) -> None:
source = (Path(__file__).resolve().parents[1] / "rvctl.py").read_text(encoding="utf-8")
self.assertIn('"profile-build"', source)
self.assertNotIn('"image-build",', source)
def test_mcp_requires_explicit_instance(self) -> None:
parser = rvctl.build_parser()
args = parser.parse_args(["mcp", "nvim", "hazard3-inf-bss-task1"])
self.assertEqual(args.mcp_kind, "nvim")
self.assertEqual(args.instance, "hazard3-inf-bss-task1")
if __name__ == "__main__":
unittest.main()