Fix shutdown probe tests

Signed-off-by: Loren Eteval <loren.eteval@proton.me>

tests/fixtures/shutdown_probe.py

@@ -134,7 +134,9 @@ class _Application(DesktopApplication):
             self.connectionController._state = ConnectionState.Connected
 
     def _initializeUI(self):
-        if self.mode == 'startup-failure':
+        if self.mode.startswith('startup-failure'):
+            _mark(self.path, 'startup failed')
+
             raise RuntimeError('fixture startup failure')
 
         if self.mode in ('subscription', 'statistics'):
@@ -161,6 +163,14 @@ class _Application(DesktopApplication):
 
         QtCore.QTimer.singleShot(self.idleMilliseconds, self._requestQuit)
 
+    def _cleanupControllers(self):
+        if self.mode == 'startup-failure-early-rollback':
+            # Exercise deadline enforcement before plugin cleanup can write
+            # its first event, without relying on machine-dependent delays.
+            threading.Event().wait()
+
+        super()._cleanupControllers()
+
     def _requestQuit(self):
         _mark(self.path, 'quit requested')
 
@@ -191,6 +201,10 @@ def runShutdownProbe(mode, *, timeout=0.5, exitcode=0, idleMilliseconds=0):
     """Return actual child reaping and exit classification, with a test failsafe."""
     with tempfile.TemporaryDirectory(prefix='furious-shutdown-') as directory:
         path = str(Path(directory) / 'events.txt')
+        # The child may be killed before publishing any events. Preserve an
+        # empty log so the result still exposes its exit/deadline diagnostics.
+        Path(path).touch()
+
         process = AppMainProcess(
             functools.partial(_Application, mode, path, exitcode, idleMilliseconds)
         )

tests/test_application_process.py

@@ -119,6 +119,19 @@ def _exitingApplication(stage, crashDirectory):
 class AppMainProcessTest(unittest.TestCase):
     """Verify exact child ownership, crash mapping, and signal safety."""
 
+    def testShutdownProbeReportsChildExitWithoutEvents(self):
+        """An early child exit retains diagnostics even without fixture events."""
+        with mock.patch(
+            'tests.fixtures.shutdown_probe.AppMainProcess',
+            side_effect=lambda _factory: AppMainProcess(_successfulApplication),
+        ):
+            result = runShutdownProbe('normal')
+
+        self.assertEqual(result['exitcode'], 0, result)
+        self.assertFalse(result['forced'], result)
+        self.assertFalse(result['alive'], result)
+        self.assertEqual(result['events'], [], result)
+
     def testStuckWorkersAndCleanupCannotKeepTheApplicationAlive(self):
         for mode in ('subscription', 'statistics', 'cleanup', 'gil', 'runtime-stuck'):
             with self.subTest(mode=mode):
@@ -181,15 +194,23 @@ class AppMainProcessTest(unittest.TestCase):
             process.close()
 
     def testStartupFailureArmsDeadlineBeforeBlockedRollback(self):
-        result = runShutdownProbe('startup-failure')
+        for mode in ('startup-failure', 'startup-failure-early-rollback'):
+            with self.subTest(mode=mode):
+                result = runShutdownProbe(mode)
 
-        self.assertTrue(result['forced'], result)
-        self.assertEqual(
-            result['exitcode'],
-            ApplicationRunner.ExitCode.UnknownException.value,
-            result,
-        )
-        self.assertIn('cleanup entered', result['events'])
+                self.assertTrue(result['forced'], result)
+                self.assertEqual(
+                    result['exitcode'],
+                    ApplicationRunner.ExitCode.UnknownException.value,
+                    result,
+                )
+                self.assertFalse(result['alive'], result)
+                self.assertLess(result['deadlineOverrun'], 3, result)
+                self.assertIn('startup failed', result['events'], result)
+                self.assertNotIn('cleanup finished', result['events'], result)
+
+                if mode == 'startup-failure-early-rollback':
+                    self.assertNotIn('cleanup entered', result['events'], result)
 
     def testRepeatedShutdownNotificationsDoNotExtendDeadline(self):
         process = AppMainProcess(_successfulApplication)