Probe installed user VM and audio continuity across logout

tests/test_iso_smoke.py

@@ -297,3 +297,132 @@ else: print('fixture tool')
 
     def test_tcp_probe_rejects_truncated_marker(self):
         self.assertNotEqual(self.tcp_probe([b'HYDRA_LAN_TO_']).returncode, 0)
+
+    def lifecycle_fixture(self):
+        import json
+        import os
+        import uuid
+        name = 'logout-test'
+        runtime = self.root / 'private'
+        run = runtime / name
+        run.mkdir(parents=True, mode=0o700)
+        runtime.chmod(0o700)
+        audio = self.root / 'user-runtime'
+        (audio / 'pulse').mkdir(parents=True)
+        context = {'name': name, 'uuid': str(uuid.uuid4()), 'runtime': str(runtime),
+                   'user_runtime': str(audio), 'uid': os.getuid()}
+        process = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)',
+                                    '-name', name, '-uuid', context['uuid']])
+        self.addCleanup(lambda: process.wait(timeout=3))
+        self.addCleanup(lambda: process.poll() is None and process.kill())
+        (run / 'qemu.pid').write_text(str(process.pid))
+        (run / 'identity.json').write_text(json.dumps({'name': name, 'uuid': context['uuid']}))
+        listeners = []
+        for path in (run / 'qmp.sock', run / 'spice.sock', audio / 'pipewire-0', audio / 'pulse/native'):
+            listener = socket.socket(socket.AF_UNIX)
+            listener.bind(str(path))
+            listener.listen(5)
+            listener.settimeout(.1)
+            listeners.append(listener)
+            self.addCleanup(listener.close)
+        stopped = threading.Event()
+        def qmp_server():
+            while not stopped.is_set():
+                try:
+                    connection, _ = listeners[0].accept()
+                except socket.timeout:
+                    continue
+                except OSError:
+                    break
+                with connection, connection.makefile('rwb', buffering=0) as stream:
+                    stream.write(b'{"QMP": {}}\n')
+                    for line in stream:
+                        request = json.loads(line)
+                        replies = {'query-name': {'name': context.get('qmp_name', name)},
+                                   'query-uuid': {'UUID': context['uuid']},
+                                   'query-status': {'status': 'running'},
+                                   'query-kvm': {'enabled': True, 'present': True}}
+                        stream.write(json.dumps({'return': replies.get(request['execute'], {}),
+                                                 'id': request['id']}).encode() + b'\n')
+        thread = threading.Thread(target=qmp_server)
+        thread.start()
+        self.addCleanup(lambda: thread.join(timeout=3))
+        self.addCleanup(stopped.set)
+        return context, process, run
+
+    def test_logout_probe_detects_lost_private_socket(self):
+        context, process, run = self.lifecycle_fixture()
+        before = self.mod.lifecycle_snapshot(context)
+        self.assertEqual(self.mod.lifecycle_snapshot(context), before)
+        (run / 'spice.sock').unlink()
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.lifecycle_snapshot(context)
+
+    def test_logout_probe_rejects_stale_metadata_and_foreign_qmp_identity(self):
+        import json
+        context, process, run = self.lifecycle_fixture()
+        self.mod.lifecycle_snapshot(context)
+        (run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': 'stale'}))
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.lifecycle_snapshot(context)
+        (run / 'identity.json').write_text(json.dumps({'name': context['name'], 'uuid': context['uuid']}))
+        context['qmp_name'] = 'foreign-process'
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.lifecycle_snapshot(context)
+
+    def test_logout_probe_rejects_dead_process_and_changed_runtime_socket(self):
+        context, process, run = self.lifecycle_fixture()
+        before = self.mod.lifecycle_snapshot(context)
+        audio = Path(context['user_runtime']) / 'pipewire-0'
+        audio.unlink()
+        replacement = socket.socket(socket.AF_UNIX)
+        self.addCleanup(replacement.close)
+        replacement.bind(str(audio))
+        replacement.listen(1)
+        after = self.mod.lifecycle_snapshot(context)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.verify_lifecycle_survival(before, after)
+        process.terminate()
+        process.wait(timeout=3)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.lifecycle_snapshot(context)
+
+    def test_logout_probe_accepts_verified_same_runtime_and_rejects_process_restart(self):
+        context, process, run = self.lifecycle_fixture()
+        before = self.mod.lifecycle_snapshot(context)
+        self.mod.verify_lifecycle_survival(before, self.mod.lifecycle_snapshot(context))
+        process.terminate()
+        process.wait(timeout=3)
+        replacement = subprocess.Popen(['python3', '-c', 'import time; time.sleep(30)',
+                                        '-name', context['name'], '-uuid', context['uuid']])
+        self.addCleanup(lambda: replacement.wait(timeout=3))
+        self.addCleanup(lambda: replacement.poll() is None and replacement.kill())
+        (run / 'qemu.pid').write_text(str(replacement.pid))
+        changed = self.mod.lifecycle_snapshot(context)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.verify_lifecycle_survival(before, changed)
+
+    def test_large_guest_probe_survives_serial_tty_canonical_line_limit(self):
+        import os
+        import pty
+        import termios
+        master, slave = pty.openpty()
+        self.addCleanup(os.close, master)
+        self.addCleanup(os.close, slave)
+        options = termios.tcgetattr(slave)
+        options[3] &= ~termios.ECHO
+        termios.tcsetattr(slave, termios.TCSANOW, options)
+        process = subprocess.Popen(['/bin/sh'], stdin=slave, stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE)
+        self.addCleanup(process.stdout.close)
+        self.addCleanup(process.stderr.close)
+        self.addCleanup(lambda: process.wait(timeout=3))
+        self.addCleanup(lambda: process.poll() is None and process.kill())
+        payload = 'guest-proof-' * 1200
+        command = self.mod.frame_command('printf %s ' + payload, 'longprobe')
+        os.write(master, (command + '\nexit\n').encode())
+        output, error = process.communicate(timeout=3)
+        self.assertEqual(self.mod.parse_frame(output, 'longprobe'), (0, payload.encode()), error)
+
+    def test_generated_guest_lifecycle_probe_compiles(self):
+        compile(self.mod.lifecycle_guest_script(), '<guest-lifecycle-probe>', 'exec')