105 lines
4.7 KiB
Python
105 lines
4.7 KiB
Python
"""Real boot consumer, inert Python children; Linux subreaper owns red cleanup."""
|
|
import ctypes
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
SOURCE = Path(sys.argv.pop(1)).resolve()
|
|
STUB = '''import json, os, signal, time
|
|
from pathlib import Path
|
|
name = Path(__file__).parent.name if Path(__file__).parent.name == 'business' else 'adapter'
|
|
root = Path.cwd()
|
|
def stop(signum, frame):
|
|
if (root/'ignore-term').exists() and name == 'business': return
|
|
(root / (name + '.stopped')).write_text(str(signum))
|
|
raise SystemExit(0)
|
|
signal.signal(signal.SIGTERM, stop)
|
|
signal.signal(signal.SIGINT, stop)
|
|
ready = root / (name + '.ready')
|
|
ready_tmp = root / (name + '.ready.tmp-' + str(os.getpid()))
|
|
ready_tmp.write_text(json.dumps({'pid':os.getpid(), 'env':{k:v for k,v in os.environ.items() if k.startswith('JWB_')}}))
|
|
os.replace(ready_tmp, ready)
|
|
while True:
|
|
if name == 'adapter' and (root/'exit-adapter').exists(): raise SystemExit(7)
|
|
time.sleep(.02)
|
|
'''
|
|
|
|
class BootTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
# Reap only our own deliberately orphaned fixture descendants on RED.
|
|
if ctypes.CDLL(None, use_errno=True).prctl(36, 1, 0, 0, 0) != 0:
|
|
raise OSError(ctypes.get_errno(), 'subreaper unavailable')
|
|
|
|
def run_case(self, signum, stubborn=False):
|
|
with tempfile.TemporaryDirectory(prefix='justworks-boot-') as tmp:
|
|
root = Path(tmp)
|
|
if stubborn: (root/'ignore-term').touch()
|
|
(root/'business').mkdir()
|
|
(root/'business/server.py').write_text(STUB)
|
|
(root/'server.py').write_text(STUB)
|
|
shutil.copyfile(SOURCE, root/'boot.py')
|
|
env = {'PATH': str(Path(sys.executable).parent) + ':/usr/bin:/bin',
|
|
'HOME': str(root), 'TMPDIR': str(root), 'PYTHONDONTWRITEBYTECODE': '1'}
|
|
proc = subprocess.Popen([sys.executable, '-I', '-S', '-B', 'boot.py'], cwd=root, env=env,
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
pids = []
|
|
try:
|
|
deadline = time.monotonic()+3
|
|
while not all((root/(n+'.ready')).exists() for n in ['business','adapter']):
|
|
if proc.poll() is not None or time.monotonic()>deadline:
|
|
self.fail('fixture children failed readiness')
|
|
time.sleep(.01)
|
|
ready = {n:json.loads((root/(n+'.ready')).read_text()) for n in ['business','adapter']}
|
|
pids = [r['pid'] for r in ready.values()]
|
|
self.assertEqual(ready['business']['env'], {
|
|
'JWB_HOST':'127.0.0.1', 'JWB_PORT':'4310', 'JWB_BASE_PATH':'/business-demo',
|
|
'JWB_DATA_DIR':'/data/business', 'JWB_AUTH_MODE':'enforce',
|
|
'JWB_COOKIE_SECURE':'0', 'JWB_ARCHIPELAGO':'1'})
|
|
self.assertEqual(ready['adapter']['env'], {})
|
|
if signum is None:
|
|
(root/'exit-adapter').touch()
|
|
else:
|
|
proc.send_signal(signum)
|
|
proc.wait(timeout=8)
|
|
self.assertEqual(proc.returncode, 1 if signum is None else 0)
|
|
for name in ready:
|
|
if (stubborn and name == 'business') or (signum is None and name == 'adapter'):
|
|
self.assertFalse((root/(name+'.stopped')).exists())
|
|
else:
|
|
self.assertTrue((root/(name+'.stopped')).exists(), name+' missed graceful termination')
|
|
for pid in pids:
|
|
with self.assertRaises(ProcessLookupError): os.kill(pid, 0)
|
|
finally:
|
|
if proc.poll() is None: proc.kill()
|
|
# Do not signal process groups or discover any unrelated process.
|
|
for file in [root/'business.ready',root/'adapter.ready']:
|
|
if file.exists():
|
|
pid=json.loads(file.read_text())['pid']
|
|
try: os.kill(pid, signal.SIGKILL)
|
|
except ProcessLookupError: pass
|
|
try: os.waitpid(pid, 0)
|
|
except ChildProcessError: pass
|
|
proc.communicate(timeout=3)
|
|
|
|
def test_sigterm_reaches_both_children_and_reaps(self):
|
|
self.run_case(signal.SIGTERM)
|
|
|
|
def test_resistant_child_is_killed_and_reaped(self):
|
|
self.run_case(signal.SIGTERM, stubborn=True)
|
|
|
|
def test_sigint_reaches_both_children_and_reaps(self):
|
|
self.run_case(signal.SIGINT)
|
|
|
|
def test_child_failure_stops_sibling_and_remains_failure(self):
|
|
self.run_case(None)
|
|
|
|
if __name__ == '__main__': unittest.main(verbosity=2)
|