fix: avoid VM name collisions in recovery commands

tests/test_session.py

@@ -5,6 +5,7 @@ import json
 import os
 import shlex
 import socket
+import subprocess
 import tempfile
 import unittest
 from pathlib import Path
@@ -192,6 +193,64 @@ else:
             with patch.object(session, 'view', side_effect=AssertionError('must not duplicate')), contextlib.redirect_stdout(io.StringIO()):
                 self.assertEqual(session.session_helper('reconnect:alpha', manifest), 0)
 
+    def _recovery_fixture(self):
+        for name in ('s', 'q'):
+            self.manager.add(Profile(name, self.iso))
+            paths = self.manager.paths(name)
+            paths.spice.parent.mkdir(parents=True)
+            fixture = socket.socket(socket.AF_UNIX)
+            self.addCleanup(fixture.close)
+            fixture.bind(str(paths.spice))
+        viewer = self.root / 'viewer'
+        viewer.write_text('#!' + os.sys.executable + '\n' + '''import os, sys
+from pathlib import Path
+from urllib.parse import unquote, urlsplit
+if '--help' in sys.argv:
+    print('--hide-toolbar')
+else:
+    name = Path(unquote(urlsplit(sys.argv[-1]).path)).parent.name
+    Path(os.environ['RECOVERY_CAPTURE'], name + '.connected').write_text(name)
+''')
+        viewer.chmod(0o700)
+        shell = self.root / 'shell'
+        shell.write_text('#!' + os.sys.executable + '\nimport os\nfrom pathlib import Path\nPath(os.environ["RECOVERY_CAPTURE"], "shell.opened").touch()\n')
+        shell.chmod(0o700)
+        manifest = self.root / 'session.json'
+        manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['s', 'q'], 'terminal': '/bin/sh'}))
+        return manifest, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': str(viewer), 'RECOVERY_CAPTURE': str(self.root), 'SHELL': str(shell)}
+
+    def _check_recovery_vm_name(self, name):
+        manifest, env = self._recovery_fixture()
+        processes = []
+        original_popen = subprocess.Popen
+        def launch(*args, **kwargs):
+            process = original_popen(*args, **kwargs)
+            processes.append(process)
+            return process
+        with patch.dict(os.environ, env), patch('builtins.input', side_effect=[name, ':quit', EOFError()]), patch.object(subprocess, 'Popen', launch), contextlib.redirect_stdout(io.StringIO()):
+            self.assertEqual(session.session_helper('recover', manifest), 0)
+            for process in processes:
+                self.assertEqual(process.wait(timeout=5), 0)
+        connected = self.root / (name + '.connected')
+        self.assertTrue(connected.exists(), f'VM {name} was not reconnected')
+        self.assertEqual(connected.read_text(), name)
+        self.assertFalse((self.root / 'shell.opened').exists())
+        self.assertFalse(self.manager.paths(name).disk.exists())
+
+    def test_recovery_reconnects_vm_named_s(self):
+        self._check_recovery_vm_name('s')
+
+    def test_recovery_reconnects_vm_named_q(self):
+        self._check_recovery_vm_name('q')
+
+    def test_recovery_colon_commands_open_shell_and_close_terminal(self):
+        manifest, env = self._recovery_fixture()
+        with patch.dict(os.environ, env), patch('builtins.input', side_effect=[':shell', ':quit', EOFError()]), contextlib.redirect_stdout(io.StringIO()):
+            self.assertEqual(session.session_helper('recover', manifest), 0)
+        self.assertTrue((self.root / 'shell.opened').exists())
+        self.assertFalse((self.root / 's.connected').exists())
+        self.assertFalse((self.root / 'q.connected').exists())
+
 
 if __name__ == '__main__':
     unittest.main()