Distinguish closed login scopes during lifecycle acceptance

tests/test_iso_smoke.py

@@ -342,6 +342,8 @@ else: print('fixture tool')
                                    'query-uuid': {'UUID': context['uuid']},
                                    'query-status': {'status': 'running'},
                                    'query-kvm': {'enabled': True, 'present': True}}
+                        if request['execute'] == 'query-kvm' and 'on_snapshot' in context:
+                            context.pop('on_snapshot')()
                         stream.write(json.dumps({'return': replies.get(request['execute'], {}),
                                                  'id': request['id']}).encode() + b'\n')
         thread = threading.Thread(target=qmp_server)
@@ -426,3 +428,75 @@ else: print('fixture tool')
 
     def test_generated_guest_lifecycle_probe_compiles(self):
         compile(self.mod.lifecycle_guest_script(), '<guest-lifecycle-probe>', 'exec')
+
+    def session_fixture(self, records):
+        import json
+        import os
+        from unittest.mock import patch
+        source = self.root / 'session-records.json'
+        source.write_text(json.dumps(records))
+        commands = self.root / 'session-fixture-bin'
+        commands.mkdir()
+        command = commands / 'loginctl'
+        command.write_text('''#!/usr/bin/env python3
+import json, os, sys
+records = json.load(open(os.environ['SMOKE_SESSION_RECORDS']))
+args = sys.argv[1:]
+if args[0] == 'show-user':
+    print(' '.join(record['Id'] for record in records if str(record['User']) == args[1]))
+elif args[0] == 'show-session':
+    record = next((record for record in records if record['Id'] == args[1]), None)
+    if record is None: sys.exit(1)
+    for key, value in record.items(): print(key + '=' + str(value))
+else: sys.exit(1)
+''')
+        command.chmod(0o755)
+        environment = patch.dict(os.environ, {'PATH': str(commands) + ':' + os.environ['PATH'],
+                                              'SMOKE_SESSION_RECORDS': str(source)})
+        environment.start()
+        self.addCleanup(environment.stop)
+        return source
+
+    def session_record(self, ident='serial1', state='active', cls='user', leader=1234):
+        import os
+        return {'Id': ident, 'User': os.getuid(), 'State': state, 'Class': cls,
+                'Leader': leader, 'TTY': 'ttyS0' if cls == 'user' else ''}
+
+    def test_logout_sessions_accept_closing_scope_and_manager_with_surviving_vm(self):
+        context, process, run = self.lifecycle_fixture()
+        original = self.session_record()
+        context['login_session'] = original
+        context['before'] = self.mod.lifecycle_snapshot(context)
+        self.session_fixture([dict(original, State='closing'),
+                              self.session_record('manager1', 'active', 'manager')])
+        self.assertTrue(self.mod.no_open_logins(context))
+        self.assertEqual(self.mod.capture_logout_snapshot(context), context['before'])
+
+    def test_logout_sessions_initial_login_ignores_manager_but_rejects_open_replacement(self):
+        import json
+        import os
+        original = self.session_record()
+        context = {'uid': os.getuid(), 'login_session': original}
+        source = self.session_fixture([original, self.session_record('manager1', 'active', 'manager')])
+        self.assertEqual(self.mod.require_initial_login(os.getuid(), 'serial1'), original)
+        for state in ('online', 'active'):
+            with self.subTest(state=state):
+                source.write_text(json.dumps([dict(original, State='closing'),
+                                             self.session_record('new-login', state)]))
+                self.assertFalse(self.mod.no_open_logins(context))
+        source.write_text(json.dumps([dict(original, State='closing', Leader=9999)]))
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.no_open_logins(context)
+
+    def test_logout_sessions_reject_login_appearing_during_snapshot(self):
+        import json
+        context, process, run = self.lifecycle_fixture()
+        original = self.session_record()
+        context['login_session'] = original
+        context['before'] = self.mod.lifecycle_snapshot(context)
+        source = self.session_fixture([dict(original, State='closing')])
+        context['on_snapshot'] = lambda: source.write_text(json.dumps([
+            dict(original, State='closing'), self.session_record('relogin')]))
+        self.assertTrue(self.mod.no_open_logins(context))
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.capture_logout_snapshot(context)