user
662 lines
28 KiB
Python
662 lines
28 KiB
Python
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()
|