173 lines
7.3 KiB
Python
173 lines
7.3 KiB
Python
"""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()
|