Capture shutdown probe tracebacks

Signed-off-by: Loren Eteval <loren.eteval@proton.me>
This commit is contained in:
Loren Eteval
2026-10-13 19:36:09 +08:00
parent 1c52fc306a
commit e4069d04da
3 changed files with 82 additions and 10 deletions
+5 -2
View File
@@ -311,8 +311,11 @@ python -m tests.fixtures.shutdown_probe normal
```
JSON reports include actual child liveness, forced-exit classification, semantic
result, deadline overrun and cleanup events. The unittest assertions check these
outcomes; standalone invocations provide the same evidence for compiled checks.
result, deadline overrun, cleanup events and captured application-run stderr.
Expected startup-failure tracebacks are retained in the `stderr` field, including
after forced termination, instead of being printed as CI error annotations.
The unittest assertions include the result on failure so those diagnostics remain
available. Standalone invocations provide the same evidence for compiled checks.
A test-only failsafe reaps the exact child if production supervision regresses.
For a Windows build, use COFF constants objects, as `Deploy.py` does for the
+21 -3
View File
@@ -36,6 +36,7 @@ from Furious.Utility.AppMainProcess import AppMainProcess
from PySide6 import QtCore
from contextlib import redirect_stderr
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
@@ -86,11 +87,12 @@ class _Runtime(CoreRuntime):
class _Application(DesktopApplication):
def __init__(self, mode, path, exitcode, idleMilliseconds):
def __init__(self, mode, path, exitcode, idleMilliseconds, diagnosticPath):
super().__init__(['shutdown-probe'])
self.mode, self.path = mode, path
self.requestedCode, self.idleMilliseconds = exitcode, idleMilliseconds
self.diagnosticPath = diagnosticPath
self._shutdownRetryTimer.setInterval(25)
self.setQuitOnLastWindowClosed(False)
@@ -193,7 +195,13 @@ class _Application(DesktopApplication):
_mark(self.path, 'cleanup finished')
def run(self):
with mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'):
# Forced termination never exits this context. Flush each traceback
# line as it is written so the parent can still report the failure.
with (
open(self.diagnosticPath, 'w', encoding='utf-8', buffering=1) as diagnostic,
redirect_stderr(diagnostic),
mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'),
):
return super().run()
@@ -204,9 +212,18 @@ def runShutdownProbe(mode, *, timeout=0.5, exitcode=0, idleMilliseconds=0):
# 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()
diagnosticPath = Path(directory) / 'stderr.txt'
diagnosticPath.touch()
process = AppMainProcess(
functools.partial(_Application, mode, path, exitcode, idleMilliseconds)
functools.partial(
_Application,
mode,
path,
exitcode,
idleMilliseconds,
str(diagnosticPath),
)
)
process.ShutdownTimeout = timeout
@@ -231,6 +248,7 @@ def runShutdownProbe(mode, *, timeout=0.5, exitcode=0, idleMilliseconds=0):
'alive': process.is_alive(),
'deadlineOverrun': time.monotonic() - process._shutdownDeadline.value,
'events': Path(path).read_text(encoding='utf-8').splitlines(),
'stderr': diagnosticPath.read_text(encoding='utf-8'),
}
finally:
safety.cancel()
+56 -5
View File
@@ -32,8 +32,10 @@ from unittest import mock
import importlib
import functools
import json
import multiprocessing
import signal
import subprocess
import sys
import tempfile
import unittest
@@ -131,6 +133,7 @@ class AppMainProcessTest(unittest.TestCase):
self.assertFalse(result['forced'], result)
self.assertFalse(result['alive'], result)
self.assertEqual(result['events'], [], result)
self.assertEqual(result['stderr'], '', result)
def testStuckWorkersAndCleanupCannotKeepTheApplicationAlive(self):
for mode in ('subscription', 'statistics', 'cleanup', 'gil', 'runtime-stuck'):
@@ -141,12 +144,14 @@ class AppMainProcessTest(unittest.TestCase):
self.assertEqual(result['exitcode'], 0, result)
self.assertFalse(result['alive'], result)
self.assertLess(result['deadlineOverrun'], 3, result)
self.assertIn('quit requested', result['events'])
self.assertIn('quit requested', result['events'], result)
if mode == 'statistics':
self.assertIn('cleanup finished', result['events'])
self.assertIn('cleanup finished', result['events'], result)
elif mode == 'runtime-stuck':
self.assertGreater(result['events'].count('runtime stop'), 1)
self.assertGreater(
result['events'].count('runtime stop'), 1, result
)
def testNormalAndRecoveredShutdownDoNotForceExit(self):
for mode in ('normal', 'runtime-retry'):
@@ -155,10 +160,12 @@ class AppMainProcessTest(unittest.TestCase):
self.assertFalse(result['forced'], result)
self.assertEqual(result['exitcode'], 23, result)
self.assertIn('cleanup finished', result['events'])
self.assertIn('cleanup finished', result['events'], result)
if mode == 'runtime-retry':
self.assertGreaterEqual(result['events'].count('runtime stop'), 3)
self.assertGreaterEqual(
result['events'].count('runtime stop'), 3, result
)
def testDeadlineDoesNotApplyUntilShutdownStarts(self):
"""Long ordinary operation never arms the supervisor's shutdown deadline."""
@@ -212,6 +219,50 @@ class AppMainProcessTest(unittest.TestCase):
if mode == 'startup-failure-early-rollback':
self.assertNotIn('cleanup entered', result['events'], result)
def testStartupFailureTracebacksStayInProbeDiagnostics(self):
"""Forced exits retain tracebacks without leaking CI annotation lines."""
for mode in ('startup-failure', 'startup-failure-early-rollback'):
with self.subTest(mode=mode):
completed = subprocess.run(
[sys.executable, '-m', 'tests.fixtures.shutdown_probe', mode],
capture_output=True,
text=True,
encoding='utf-8',
timeout=30,
)
output = (completed.stdout, completed.stderr)
self.assertEqual(completed.returncode, 0, repr(output))
self.assertNotIn('Traceback (most recent call last):', completed.stderr)
self.assertNotIn('fixture startup failure', completed.stderr)
self.assertEqual(len(completed.stdout.splitlines()), 1, repr(output))
result = json.loads(completed.stdout)
diagnostic = result['stderr']
self.assertTrue(result['forced'], result)
self.assertFalse(result['alive'], result)
self.assertEqual(
result['exitcode'],
ApplicationRunner.ExitCode.UnknownException.value,
result,
)
self.assertEqual(
diagnostic.count('Traceback (most recent call last):'), 1, result
)
self.assertIn('shutdown_probe.py', diagnostic, result)
self.assertIn('_initializeUI', diagnostic, result)
self.assertIn(
'RuntimeError: fixture startup failure', diagnostic, result
)
# Ordinary assertions include the result, preserving the full
# traceback as escaped text when a real expectation fails.
with self.assertRaises(AssertionError) as failure:
self.assertEqual(result['exitcode'], 0, result)
self.assertIn(repr(diagnostic), str(failure.exception))
def testRepeatedShutdownNotificationsDoNotExtendDeadline(self):
process = AppMainProcess(_successfulApplication)