66 lines
2.7 KiB
Python
66 lines
2.7 KiB
Python
"""Exercise real Windows process cleanup without touching PostgreSQL."""
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
@unittest.skipUnless(os.name == 'nt', 'Windows launcher')
|
|
class LauncherLifecycle(unittest.TestCase):
|
|
def test_forced_owner_exit_stops_descendants(self):
|
|
kernel = ctypes.WinDLL('kernel32', use_last_error=True)
|
|
kernel.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
|
|
kernel.OpenProcess.restype = wintypes.HANDLE
|
|
kernel.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD]
|
|
kernel.WaitForSingleObject.restype = wintypes.DWORD
|
|
kernel.CloseHandle.argtypes = [wintypes.HANDLE]
|
|
code = '''from start_dashboard import TerminalJob
|
|
import subprocess, sys, time
|
|
job = TerminalJob()
|
|
child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
print(child.pid, flush=True)
|
|
time.sleep(60)
|
|
'''
|
|
owner = subprocess.Popen([sys.executable, '-u', '-c', code], cwd=ROOT,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
|
child_handle = None
|
|
try:
|
|
line = owner.stdout.readline().strip()
|
|
if not line:
|
|
self.fail(owner.communicate(timeout=5)[1])
|
|
child_handle = kernel.OpenProcess(0x100000, False, int(line)) # SYNCHRONIZE
|
|
self.assertTrue(child_handle)
|
|
owner.kill()
|
|
owner.wait(timeout=5)
|
|
self.assertEqual(kernel.WaitForSingleObject(child_handle, 5000), 0,
|
|
'A child survived the owner terminal process exiting.')
|
|
finally:
|
|
if owner.poll() is None:
|
|
owner.kill()
|
|
owner.communicate(timeout=5)
|
|
if child_handle:
|
|
kernel.CloseHandle(child_handle)
|
|
|
|
def test_busy_port_leaves_existing_service_alone(self):
|
|
with socket.socket() as listener:
|
|
listener.bind(('127.0.0.1', 0))
|
|
listener.listen()
|
|
env = dict(os.environ, MOCKUP_PORT=str(listener.getsockname()[1]))
|
|
result = subprocess.run([sys.executable, 'start_dashboard.py', '--no-browser'],
|
|
cwd=ROOT, env=env, capture_output=True, text=True, timeout=10)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertTrue('already in use' in result.stderr or 'already running' in result.stderr,result.stderr)
|
|
with socket.create_connection(listener.getsockname(), timeout=1):
|
|
pass
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|