"""Exercise the actual router main loop with bounded inert adapters.""" import importlib.util import json import os from pathlib import Path import subprocess import tempfile import unittest spec = importlib.util.spec_from_file_location( 'runtime_router', Path(__file__).resolve().parents[2] / 'docker/public-web-router/router.py') router = importlib.util.module_from_spec(spec) spec.loader.exec_module(router) GENERATED = ('frpc.json', 'Caddyfile', 'gateway.crt') def config(name='site-a', domain='site.example', routes=True): return {'schema': 1, 'certificate_mode': 'test', 'gateway': { 'host': '192.0.2.1', 'port': 7400, 'node_id': 'node-a', 'transport_token': 'a' * 64, 'enrollment_token': 'b' * 64, 'ca_pem': '-----BEGIN CERTIFICATE-----\ntest-only fixture\n-----END CERTIFICATE-----', 'tls_server_name': 'gateway.example', 'domains': ['site.example', 'next.example']}, 'routes': ([{'id': name, 'domain': domain, 'fips_address': 'fd00::1', 'port': 32000}] if routes else [])} class InertChild: def __init__(self, command): self.command = tuple(command) self.returncode = None self.terminations = 0 def poll(self): return self.returncode def terminate(self): self.terminations += 1 self.returncode = 0 def wait(self, timeout=None): return self.returncode def kill(self): self.returncode = -9 class RuntimeTests(unittest.TestCase): def exercise(self, states, *, fail_validate=None, interrupt_at=None): with tempfile.TemporaryDirectory(prefix='router-runtime-', dir=os.environ['TMPDIR']) as directory: base = Path(directory) root = base / 'router' source = base / 'config' / 'router.json' source.parent.mkdir() root.mkdir() source.write_bytes(states[0]) unrelated = root / 'unrelated.state' unrelated.write_bytes(b'preserve me') children = [] runs = [] checkpoints = 0 snapshots = [] def run(command, **kwargs): runs.append(tuple(command)) if command[:3] == router.FRPC_VERIFY: document = json.loads(Path(command[3]).read_text()) staged_ca = Path(document['transport']['tls']['trustedCaFile']) self.assertIn('.next-', staged_ca.name) self.assertTrue(staged_ca.is_file()) if fail_validate and fail_validate(Path(command[3])): raise subprocess.CalledProcessError(1, command) def popen(command): child = InertChild(command) children.append(child) return child def checkpoint(label): nonlocal checkpoints checkpoints += 1 if interrupt_at == checkpoints: raise InterruptedError(f'test-only interruption at {label}') def pause(seconds): self.assertEqual(seconds, 2) snapshots.append({ 'files': {path.name: path.read_bytes() for path in root.iterdir() if path.is_file()}, 'status': json.loads((root / 'status.json').read_bytes()), 'runs': list(runs), 'children': list(children), }) index = len(snapshots) if index < len(states): source.write_bytes(states[index]) router.supervise(source=source, root=root, run=run, popen=popen, pause=pause, max_cycles=len(states), checkpoint=checkpoint) self.assertEqual(unrelated.read_bytes(), b'preserve me') self.assertEqual(len(snapshots), len(states)) return snapshots @staticmethod def expected_files(state): caddy, frpc, pem = router.render(json.loads(state)) return {'Caddyfile': caddy.encode(), 'frpc.json': json.dumps(frpc).encode(), 'gateway.crt': pem.encode()} def test_nonempty_update_promotes_validated_managed_configuration(self): first = json.dumps(config()).encode() second = json.dumps(config('site-b', 'next.example')).encode() initial, updated = self.exercise([first, second]) for name, value in self.expected_files(first).items(): self.assertEqual(initial['files'][name], value) for name, value in self.expected_files(second).items(): self.assertEqual(updated['files'][name], value) self.assertEqual(updated['status']['routes'], 1) self.assertFalse(updated['status']['update_failed']) self.assertTrue(any('.next-' in part for command in updated['runs'] for part in command)) self.assertTrue(all(child.returncode == 0 for child in initial['children'])) def test_validation_failure_preserves_last_nonempty_configuration(self): first = json.dumps(config()).encode() second = json.dumps(config('site-b', 'next.example')).encode() snapshots = self.exercise( [first, second], fail_validate=lambda path: path.name.startswith('Caddyfile.next-') and b'next.example' in path.read_bytes()) failed = snapshots[-1] for name, value in self.expected_files(first).items(): self.assertEqual(failed['files'][name], value) self.assertTrue(failed['status']['configured']) self.assertTrue(failed['status']['update_failed']) self.assertEqual(failed['status']['routes'], 1) self.assertGreaterEqual(len(failed['children']), 2) def test_interrupted_promotion_rolls_back_all_managed_files(self): first = json.dumps(config()).encode() second = json.dumps(config('site-b', 'next.example')).encode() failed = self.exercise([first, second], interrupt_at=5)[-1] for name, value in self.expected_files(first).items(): self.assertEqual(failed['files'][name], value) self.assertTrue(failed['status']['configured']) self.assertTrue(failed['status']['update_failed']) self.assertFalse(any('.next-' in name or '.rollback-' in name for name in failed['files'])) def test_invalid_configuration_deactivates_without_touching_unrelated_state(self): first = json.dumps(config()).encode() active, rejected = self.exercise([first, b'{invalid json']) self.assertTrue(set(GENERATED).issubset(active['files'])) self.assertTrue(set(GENERATED).isdisjoint(rejected['files'])) self.assertFalse(rejected['status']['configured']) self.assertTrue(rejected['status']['update_failed']) self.assertIn('unrelated.state', rejected['files']) def test_zero_route_transition_removes_only_managed_files(self): first = json.dumps(config()).encode() empty = json.dumps(config(routes=False)).encode() active, removed = self.exercise([first, empty]) self.assertTrue(set(GENERATED).issubset(active['files'])) self.assertTrue(set(GENERATED).isdisjoint(removed['files'])) self.assertEqual(removed['status']['routes'], 0) self.assertFalse(removed['status']['update_failed']) self.assertIn('unrelated.state', removed['files']) if __name__ == '__main__': unittest.main()