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_rvctl_compatibility_wrapper_is_silent(self) -> None: wrapper = Path(__file__).resolve().parents[1] / "rvctl" completed = subprocess.run( [str(wrapper), "--help"], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) self.assertEqual(completed.returncode, 0, completed.stderr) self.assertNotIn("deprecated", completed.stderr.lower()) self.assertNotIn("warning", completed.stderr.lower()) 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()