test_logging_concurrency.py #1

#!/usr/bin/env python3
"""Real-process Maintenance log leases; files stay inside disposable directories."""
import json
import os
from pathlib import Path
import signal
import subprocess
import sys
import tempfile
import time
import unittest

SDP = Path(__file__).resolve().parents[1]
CHILD = r'''
import json, os, signal, sys, time
from pathlib import Path
sys.path[:0] = [sys.argv[1], sys.argv[2]]
import sdputils
import sdp_structured_log as logger
root = Path(os.environ['LOGS'])
name, mode = sys.argv[3:5]
(root / (name + '.requested')).touch()
while not (root / 'start').exists(): time.sleep(.01)
if mode == 'handled-term':
    def terminate(signum, frame):
        print(name + '-terminated', flush=True)
        raise SystemExit(128 + signum)
    signal.signal(signal.SIGTERM, terminate)
with logger.run('concurrent.py', instance='fixture'), sdputils.maintenance_run('concurrent.py', 'fixture'):
    print(name + '-start', flush=True)
    fd = sys.stdout.logfile.fileno()
    (root / (name + '.ready')).write_text(json.dumps({'inode': os.fstat(fd).st_ino, 'links': os.fstat(fd).st_nlink}))
    while not (root / (name + '.go')).exists(): time.sleep(.01)
    print(name + '-finish', flush=True)
    (root / (name + '.end')).write_text(json.dumps({'inode': os.fstat(fd).st_ino, 'links': os.fstat(fd).st_nlink}))
    if mode == 'failure': raise RuntimeError(name + '-specific-failure')
'''


class MaintenanceLogConcurrency(unittest.TestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory(prefix='sdp-log-lease-test-')
        self.addCleanup(self.temporary.cleanup)
        self.root = Path(self.temporary.name)
        self.children = []
        self.addCleanup(self.reap)

    def reap(self):
        for child in self.children:
            if child.poll() is None:
                child.kill()
            child.communicate(timeout=5)

    def wait_file(self, path):
        deadline = time.monotonic() + 10
        while not path.exists():
            if time.monotonic() >= deadline:
                self.fail('Timed out waiting for ' + str(path))
            time.sleep(.01)

    def start(self, name, mode='success', keep='1', simultaneous=False):
        env = dict(os.environ, LOGS=str(self.root), SDP_RUN_LOG_MAX_BYTES='0', SDP_LOG_FORMAT='jsonl')
        if keep is None:
            env.pop('SDP_RUN_LOG_KEEP', None)
        else:
            env['SDP_RUN_LOG_KEEP'] = keep
        child = subprocess.Popen([sys.executable, '-c', CHILD, str(SDP / 'Maintenance'),
                                  str(SDP / 'Server/Unix/p4/common/bin'), name, mode],
                                 env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        self.children.append(child)
        self.wait_file(self.root / (name + '.requested'))
        if not simultaneous:
            (self.root / 'start').touch()
            self.wait_file(self.root / (name + '.ready'))
        return child

    def finish(self, name, child, code=0):
        (self.root / (name + '.go')).touch()
        out, err = child.communicate(timeout=10)
        self.assertEqual(child.returncode, code, out + err)
        return out, err

    def logs(self):
        return {p.name: p.read_text() for p in self.root.iterdir()
                if p.name == 'concurrent.log' or p.name.startswith('concurrent.log.run-')}

    def test_overlapping_runs_keep_active_inode_and_all_diagnostics_at_one_and_default_retention(self):
        for keep in ('1', None):
            with self.subTest(keep=keep):
                for path in self.root.iterdir():
                    path.unlink()
                owner = self.start('owner', keep=keep)
                inode = json.loads((self.root / 'owner.ready').read_text())['inode']
                # Exceeds the default seven archives while the first run stays active.
                for index in range(9):
                    name = 'peer' + str(index)
                    peer = self.start(name, keep=keep)
                    self.finish(name, peer)
                    self.assertEqual((self.root / 'concurrent.log').stat().st_ino, inode)
                    self.assertEqual(len(self.logs()), 1)
                self.finish('owner', owner)
                self.assertEqual(json.loads((self.root / 'owner.end').read_text())['links'], 1)
                after = self.start('after', keep=keep)
                self.finish('after', after)
                archived, = (content for name, content in self.logs().items() if '.run-' in name)
                for name in ['owner'] + ['peer' + str(i) for i in range(9)]:
                    self.assertIn(name + '-start', archived)
                    self.assertIn(name + '-finish', archived)
                events = [json.loads(line) for line in (self.root / 'concurrent.jsonl').read_text().splitlines()]
                self.assertEqual(len(events), 11)
                self.assertTrue(all(row['status'] == 'success' for row in events))
                self.assertEqual(len({row['run_id'] for row in events}), 11)
                if keep == '1':
                    latest = self.start('latest', keep=keep)
                    self.finish('latest', latest)
                    self.assertEqual(len(self.logs()), 2)
                    self.assertNotIn('owner-start', '\n'.join(self.logs().values()))

    def test_simultaneous_preparation_opens_one_shared_inode(self):
        peers = [(str(index), self.start(str(index), simultaneous=True)) for index in range(8)]
        (self.root / 'start').touch()
        for name, _ in peers:
            self.wait_file(self.root / (name + '.ready'))
        inodes = {json.loads((self.root / (name + '.ready')).read_text())['inode'] for name, _ in peers}
        self.assertEqual(len(inodes), 1)
        for name, child in peers:
            self.finish(name, child)
        self.assertEqual(len(self.logs()), 1)
        text = (self.root / 'concurrent.log').read_text()
        for name, _ in peers:
            self.assertIn(name + '-finish', text)

    def test_failure_and_handled_or_abrupt_term_release_leases_and_preserve_evidence(self):
        for mode, code in [('failure', 1), ('handled-term', 143), ('abrupt-term', -signal.SIGTERM)]:
            with self.subTest(mode=mode):
                for path in self.root.iterdir():
                    path.unlink()
                first = self.start('first', mode)
                peer = self.start('peer')
                if mode == 'failure':
                    self.finish('first', first, code)
                else:
                    first.send_signal(signal.SIGTERM)
                    first.communicate(timeout=10)
                    self.assertEqual(first.returncode, code)
                self.finish('peer', peer)
                after = self.start('after')
                self.finish('after', after)
                archived, = (content for name, content in self.logs().items() if '.run-' in name)
                self.assertIn('first-start', archived)
                self.assertIn('peer-finish', archived)
                if mode == 'failure':
                    self.assertIn('first-specific-failure', archived)
                if mode == 'handled-term':
                    self.assertIn('first-terminated', archived)
                    self.assertIn('failed (exit 143)', archived)


if __name__ == '__main__':
    unittest.main()