feat: switch dynamic MCP container targets

This commit is contained in:
user
2026-07-17 08:24:37 +02:00
parent 1a98c14654
commit 1e1c124c9f
2 changed files with 259 additions and 2 deletions
+45
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import json
import io
import os
import socket
import subprocess
import tempfile
import unittest
@@ -289,6 +290,50 @@ class StemctlContractTests(unittest.TestCase):
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_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()
if __name__ == "__main__":
unittest.main()