from __future__ import annotations import json import io import os import socket import subprocess import tempfile import unittest from contextlib import redirect_stderr, 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", "session"] 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") def test_mcp_select_accepts_instance_or_container_id(self) -> None: parser = rvctl.build_parser() by_instance = parser.parse_args(["mcp", "select", "rp2350-pointers-final"]) by_container = parser.parse_args(["mcp", "select", "3aca2c1c4c7a"]) self.assertEqual(by_instance.mcp_kind, "select") self.assertEqual(by_instance.selector, "rp2350-pointers-final") self.assertEqual(by_container.selector, "3aca2c1c4c7a") def test_env_sources_selects_a_card_and_supports_read_only_check(self) -> None: parser = rvctl.build_parser() args = parser.parse_args(["env", "sources", "inf", "pointers", "--check"]) self.assertEqual(args.env_command, "sources") self.assertEqual(args.series, "inf") self.assertEqual(args.card, "pointers") self.assertTrue(args.check) def test_mcp_select_switches_both_sockets_with_one_current_symlink(self) -> None: with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) socket_dir = root / "source" socket_dir.mkdir() nvim_server = socket.socket(socket.AF_UNIX) tmux_server = socket.socket(socket.AF_UNIX) try: nvim_server.bind(str(socket_dir / "n.sock")) tmux_server.bind(str(socket_dir / "t.sock")) config = SimpleNamespace(socket_root=root) record = { "runtime": "podman", "container_name": "stem-rp2350-example", "container_id": "3aca2c1c4c7a" + "0" * 52, "profile": "rp2350", "target": "rp2350-rv", "instance": "rp2350-example", "instance_key": "example000001", "thread_key": "thread000001", "socket_dir": str(socket_dir), } with mock.patch.object(rvctl, "resolve_mcp_record", return_value=record), mock.patch.object( rvctl, "validated_mcp_record", return_value=(record, socket_dir / "n.sock", socket_dir / "t.sock"), ), redirect_stdout(io.StringIO()): rvctl.select_mcp_instance(config, "rp2350-example") current = root / "mcp-selected" / "current" self.assertEqual(os.readlink(current), "targets/3aca2c1c4c7a") self.assertEqual((current / "n.sock").resolve(), socket_dir / "n.sock") self.assertEqual((current / "t.sock").resolve(), socket_dir / "t.sock") finally: nvim_server.close() tmux_server.close() def test_session_parser_exposes_numbered_target_and_stage_controls(self) -> None: parser = rvctl.build_parser() args = parser.parse_args( [ "session", "reset", "hazard3-sim", "--series", "1", "--card", "7", "--task", "4", "--phase", "2", "--step", "3", "--pane", "current:0", ] ) self.assertEqual(args.session_command, "reset") self.assertEqual(args.card, "7") self.assertEqual(args.task, "4") self.assertEqual(args.phase, "2") self.assertEqual(args.step, "3") self.assertEqual(args.pane, "current:0") def test_session_phase_local_step_falls_back_after_global_number(self) -> None: entries = [] for phase_index, numbers in enumerate(((1, 2), (6, 7, 8))): for step_index, number in enumerate(numbers): entries.append( { "task": {"id": "task04", "label": "Task04", "index": 0}, "block": {"id": "flow", "label": "Flow", "index": 0}, "phase": { "id": f"phase-{phase_index + 1}", "label": f"Phase {phase_index + 1}", "index": phase_index, }, "step": { "id": f"step-{number}", "label": f"Step {number}", "number": number, "index": step_index, "global_index": len(entries), }, "snapshot": {"ref": f"task04.step-{number}", "index": step_index}, } ) args = SimpleNamespace( block=None, phase="2", step="3", snapshot=None, stage=None, ) selected = rvctl.resolve_navigation_entry({"entries": entries}, "task04_address_arithmetic_alloc", args) self.assertEqual(selected["phase"]["id"], "phase-2") self.assertEqual(selected["step"]["number"], 8) def test_session_material_status_has_three_polish_values(self) -> None: entry = { "step": {"id": "one"}, "snapshot": {"ref": "task04.one"}, } self.assertEqual( rvctl.navigation_entry_status( entry, { "task04.one": { "stop": {"symbol": "main", "offset": 0}, "verify": {"expressions": []}, } }, ), "opracowane", ) self.assertEqual(rvctl.navigation_entry_status(entry, {}), "robocze") self.assertEqual(rvctl.development_status([]), "brak") def test_session_card_number_uses_manifest_order(self) -> None: cards = [ rvctl.CardInfo("inf", "first", "first", "main", "", "a", "b", "r1", "r1a", "main"), rvctl.CardInfo("inf", "second", "second", "main", "", "a", "b", "r1", "r1a", "main"), ] config = SimpleNamespace(defaults={}) with mock.patch.object(rvctl, "source_card_infos", return_value=cards): self.assertEqual(rvctl.resolve_session_card(config, "inf", "2"), "second") def test_session_rejects_destructive_online_reset_without_pane_in_preflight(self) -> None: parser = rvctl.build_parser() args = parser.parse_args( [ "session", "reset", "hazard3-sim", "--series", "inf", "--card", "7", "--task", "4", "--pane", "none", "--step", "12", ] ) with self.assertRaisesRegex(SystemExit, "requires a debugger pane"): rvctl.validate_session_command_args(args) def test_session_rejects_profile_target_pair_before_mutation(self) -> None: args = SimpleNamespace( series="inf", card="7", task="4", profile="hazard3-sim", target="rp2350-rv", instance=None, ) with ( mock.patch.object(rvctl, "resolve_session_series", return_value="inf"), mock.patch.object(rvctl, "resolve_session_card", return_value="pointers"), mock.patch.object(rvctl, "resolve_workspace_card_path", return_value=Path("/tmp/card")), mock.patch.object(rvctl, "resolve_submission_task", return_value="task04"), self.assertRaisesRegex(SystemExit, "does not belong to profile hazard3-sim"), ): rvctl.resolve_session_target(SimpleNamespace(), args) def test_session_validates_pane_before_container_commands(self) -> None: parser = rvctl.build_parser() args = parser.parse_args( ["session", "reset", "hazard3-sim", "--series", "inf", "--card", "7", "--task", "4"] ) target = rvctl.SessionTarget( "inf", "pointers", Path("/tmp/card"), "task04_example", "hazard3-sim", "hazard3-baremetal", "hazard3-sim-inf-pointers-t4" ) with ( mock.patch.object(rvctl, "resolve_session_target", return_value=target), mock.patch.object(rvctl, "resolve_tmux_session_pane", side_effect=SystemExit("protected")), mock.patch.object(subprocess, "run") as run, self.assertRaisesRegex(SystemExit, "protected"), ): rvctl.run_session(SimpleNamespace(), args) run.assert_not_called() def test_session_start_pane_none_applies_offline_stage(self) -> None: parser = rvctl.build_parser() args = parser.parse_args( [ "session", "start", "hazard3-sim", "--series", "inf", "--card", "7", "--task", "4", "--pane", "none", "--step", "12", "--offline" ] ) target = rvctl.SessionTarget( "inf", "pointers", Path("/tmp/card"), "task04_example", "hazard3-sim", "hazard3-baremetal", "hazard3-sim-inf-pointers-t4" ) prepared = { "task": {"id": "task04"}, "block": {"id": "flow"}, "phase": {"id": "alloc-5"}, "step": {"id": "commit", "number": 12}, "snapshot": {"ref": "task04.alloc5.commit"}, } with ( mock.patch.object(rvctl, "resolve_session_target", return_value=target), mock.patch.object(rvctl, "prepare_session_stage", return_value=prepared), mock.patch.object(rvctl, "session_env_command", return_value=("true", {"command": "true"})), mock.patch.object(rvctl, "select_session_stage") as select_stage, mock.patch.object(subprocess, "run"), redirect_stdout(io.StringIO()), ): rvctl.run_session(SimpleNamespace(), args) select_stage.assert_called_once_with( mock.ANY, target, args, online=False, prepared_entry=prepared ) def test_session_stage_requires_explicit_selector_and_finite_timeout(self) -> None: parser = rvctl.build_parser() args = parser.parse_args( ["session", "stage", "hazard3-sim", "--series", "inf", "--card", "7", "--task", "4"] ) with self.assertRaisesRegex(SystemExit, "requires a UML selector"): rvctl.validate_session_command_args(args) with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): parser.parse_args( [ "session", "stage", "hazard3-sim", "--series", "inf", "--card", "7", "--task", "4", "--first", "--timeout", "nan" ] ) def test_session_protects_codex_owned_pane_by_pid_marker(self) -> None: completed = subprocess.CompletedProcess( [], 0, stdout="1:0.0\t%3\t/tmp\t999\t999\tthread-1\n", stderr="" ) with ( mock.patch.object(rvctl, "mark_invoking_codex_pane"), mock.patch.object(subprocess, "run", return_value=completed), mock.patch.dict(os.environ, {"TMUX_PANE": "%33"}), ): with self.assertRaisesRegex(SystemExit, "owned by Codex"): rvctl.resolve_tmux_session_pane("%3") self.assertEqual(rvctl.resolve_tmux_session_pane("%3", force=True)[1], "%3") def test_session_identity_is_fail_closed_on_uuid_or_hash_mismatch(self) -> None: with tempfile.TemporaryDirectory() as temporary: card_path = Path(temporary) (card_path / "json").mkdir() source = { "card": { "id": "card-1", "uuid": "uuid-1", "version": "v1", } } (card_path / "json/card_source.json").write_text(json.dumps(source), encoding="utf-8") target = rvctl.SessionTarget( "inf", "card", card_path, "task04", "hazard3-sim", "hazard3-baremetal", "instance" ) served = { "id": "card-1", "uuid": "different", "version": "v1", "source_sha256": "different", } with mock.patch.object(rvctl, "card_api_json", return_value=served): with self.assertRaisesRegex(SystemExit, "uuid, source_sha256"): rvctl.verify_card_api_identity("http://127.0.0.1:8080", target, 1) def test_session_cancelled_checkpoint_is_not_published(self) -> None: target = rvctl.SessionTarget( "inf", "pointers", Path("/tmp/card"), "task04", "hazard3-sim", "hazard3-baremetal", "hazard3-sim-inf-pointers-t4" ) args = SimpleNamespace( card_url="http://127.0.0.1:8080", timeout=1, focus="step", control=False, ) entry = { "task": {"id": "task04"}, "block": {"id": "allocator-flow"}, "phase": {"id": "alloc-5"}, "step": {"id": "alloc5-commit", "number": 12}, "snapshot": {"ref": "task04.alloc5.commit"}, } record = { "container_id": "abc123", "instance": target.instance, "profile": target.profile, "target": target.target, } api = mock.Mock(return_value={"status": "cancelled", "message": "newer request"}) with ( mock.patch.object(rvctl, "resolve_mcp_record", return_value=record), mock.patch.object(rvctl, "validated_mcp_record"), mock.patch.object(rvctl, "validate_mcp_card_mount"), mock.patch.object(rvctl, "select_mcp_instance"), mock.patch.object( rvctl, "verify_card_api_identity", return_value={ "id": "card-1", "uuid": "uuid-1", "version": "v1", "source_sha256": "abc123", }, ), mock.patch.object(rvctl, "card_api_json", api), self.assertRaisesRegex(SystemExit, "did not reach ready: cancelled"), ): rvctl.select_session_stage(SimpleNamespace(), target, args, True, entry) self.assertEqual(api.call_count, 1) payload = api.call_args.kwargs["payload"] self.assertEqual(payload["expected_identity"]["source_sha256"], "abc123") def test_mcp_registry_profile_and_target_must_match_container_labels(self) -> None: record = { "runtime": "podman", "container_name": "container", "container_id": "abc123", "instance_key": "instance-key", "profile": "hazard3-sim", "target": "hazard3-baremetal", "socket_dir": "/tmp/not-used", } inspection = [{ "Id": "abc123", "Config": {"Labels": { "edu.stem.instance-key": "instance-key", "edu.stem.profile": "hazard3-sim", "edu.stem.target": "rp2350-rv", }}, }] completed = subprocess.CompletedProcess([], 0, stdout=json.dumps(inspection), stderr="") with ( mock.patch.object(rvctl.shutil, "which", return_value="/usr/bin/podman"), mock.patch.object(subprocess, "run", return_value=completed), self.assertRaisesRegex(SystemExit, "Stale MCP registry"), ): rvctl.validated_mcp_record(record) def test_session_task_matching_is_anchored(self) -> None: self.assertTrue(rvctl.same_task("task04", "task04_address_arithmetic_alloc")) self.assertFalse(rvctl.same_task("notask04", "task04_address_arithmetic_alloc")) self.assertFalse(rvctl.same_task("figure2026", "task04_address_arithmetic_alloc")) if __name__ == "__main__": unittest.main()