188 lines
7.8 KiB
Python
188 lines
7.8 KiB
Python
"""Offline production-main tests: simulated privileged I/O, real temp files.
|
|
|
|
No shell command or socket is allowed to reach the host. These fixtures are not
|
|
installed nft/systemd/Quadlet acceptance.
|
|
"""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('existing', Path(__file__).with_name('test_repair_npm_tunnel.py'))
|
|
existing = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(existing)
|
|
m = existing.m
|
|
DROP, RULES = existing.DROP, existing.RULES
|
|
|
|
|
|
class EffectiveState(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.home = Path(self.temp.name)
|
|
self.drop = self.home / '.config/containers/systemd/nginx-proxy-manager.container.d/web-tunnel.conf'
|
|
self.drop.parent.mkdir(parents=True)
|
|
self.drop.write_text(DROP)
|
|
self.journal = self.home / '.local/state/archipelago/npm-tunnel-migration/pending.json'
|
|
self.disk_rules = RULES
|
|
self.live_rules = RULES
|
|
self.active = 'active'
|
|
self.calls = []
|
|
self.ignore_new_live = False
|
|
self.ignore_new_disk = False
|
|
self.ignore_restore_live = False
|
|
self.restart_state = 'active'
|
|
self.stop_fails_on_rollback = False
|
|
self.stops = 0
|
|
|
|
def command(self, *args, input=None):
|
|
self.calls.append((args, input))
|
|
if 'cat' in args:
|
|
return self.disk_rules
|
|
if 'list' in args:
|
|
return self.live_rules
|
|
if 'grep' in args:
|
|
return 'PreUp = nft -f /etc/wireguard/wg-web.nft\nPostDown = nft delete table inet web_tunnel'
|
|
if 'show' in args:
|
|
return self.active
|
|
if 'stop' in args:
|
|
self.stops += 1
|
|
if self.stop_fails_on_rollback and self.stops == 2:
|
|
raise RuntimeError('fixture rollback stop failure')
|
|
self.active = 'inactive'
|
|
if 'restart' in args:
|
|
self.active = self.restart_state
|
|
if 'nft' in args and '-f' in args and '--check' not in args:
|
|
rules = input.removeprefix('delete table inet web_tunnel\n')
|
|
new = ':18081' in rules
|
|
if not ((new and self.ignore_new_live) or (not new and self.ignore_restore_live)):
|
|
self.live_rules = rules
|
|
return ''
|
|
|
|
def root_write(self, path, content):
|
|
if not (':18081' in content and self.ignore_new_disk):
|
|
self.disk_rules = content
|
|
|
|
def run_main(self):
|
|
with patch.object(Path, 'home', return_value=self.home), \
|
|
patch.object(m, 'command', side_effect=self.command), \
|
|
patch.object(m, 'root_write', side_effect=self.root_write), \
|
|
patch.object(m.socket, 'socket'), contextlib.redirect_stdout(io.StringIO()) as stdout:
|
|
m.main()
|
|
return stdout.getvalue()
|
|
|
|
def test_success_reads_back_before_restart_and_before_commit(self):
|
|
self.assertIn('NPM tunnel port repaired', self.run_main())
|
|
expected = m.plan(DROP, RULES)[1]
|
|
self.assertEqual(m.normalized_rules(self.live_rules), m.normalized_rules(expected))
|
|
self.assertFalse(self.journal.exists())
|
|
restart = next(i for i, (args, _) in enumerate(self.calls) if 'restart' in args)
|
|
self.assertTrue(any('list' in args for args, _ in self.calls[:restart]))
|
|
self.assertTrue(any('list' in args for args, _ in self.calls[restart + 1:]))
|
|
self.assertEqual(self.active, 'active')
|
|
|
|
def test_list_canonical_comments_counters_priority_are_accepted(self):
|
|
original = self.command
|
|
|
|
def canonical(*args, input=None):
|
|
result = original(*args, input=input)
|
|
if 'list' in args:
|
|
result = '\n'.join(line for line in result.splitlines() if not line.lstrip().startswith('#'))
|
|
result = result.replace('priority -10', 'priority filter - 10')
|
|
result = result.replace('counter drop', 'counter packets 7 bytes 42 drop')
|
|
return result
|
|
|
|
self.command = canonical
|
|
self.run_main()
|
|
self.assertFalse(self.journal.exists())
|
|
|
|
def test_post_apply_read_failure_and_rollback_read_failure_keeps_journal(self):
|
|
original = self.command
|
|
|
|
def unreadable(*args, input=None):
|
|
if 'list' in args and self.stops:
|
|
raise RuntimeError('fixture unavailable observation')
|
|
return original(*args, input=input)
|
|
|
|
self.command = unreadable
|
|
with self.assertRaisesRegex(RuntimeError, 'unavailable observation'):
|
|
self.run_main()
|
|
self.assertTrue(self.journal.exists())
|
|
self.assertEqual(self.active, 'inactive')
|
|
|
|
def test_rollback_stop_failure_preserves_candidate_and_journal(self):
|
|
self.ignore_new_live = True
|
|
self.stop_fails_on_rollback = True
|
|
with self.assertRaisesRegex(RuntimeError, 'rollback stop failure'):
|
|
self.run_main()
|
|
self.assertTrue(self.journal.exists())
|
|
self.assertEqual(self.drop.read_text(), m.plan(DROP, RULES)[0])
|
|
|
|
def test_restart_drift_is_rechecked_and_stopped_before_restore(self):
|
|
original = self.command
|
|
|
|
def drift_after_restart(*args, input=None):
|
|
result = original(*args, input=input)
|
|
if 'restart' in args:
|
|
self.live_rules = RULES
|
|
return result
|
|
|
|
self.command = drift_after_restart
|
|
with self.assertRaisesRegex(RuntimeError, 'effective tunnel rules'):
|
|
self.run_main()
|
|
self.assertEqual(self.active, 'inactive')
|
|
self.assertEqual(self.drop.read_text(), DROP)
|
|
self.assertFalse(self.journal.exists())
|
|
restart = next(i for i, (args, _) in enumerate(self.calls) if 'restart' in args)
|
|
restore = next(i for i, (args, data) in enumerate(self.calls)
|
|
if i > restart and 'nft' in args and data == 'delete table inet web_tunnel\n' + RULES)
|
|
self.assertTrue(any('stop' in args for args, _ in self.calls[restart + 1:restore]))
|
|
|
|
def test_crash_recovery_does_not_discard_unobserved_restore(self):
|
|
import json
|
|
self.journal.parent.mkdir(parents=True)
|
|
self.journal.write_text(json.dumps({'drop': DROP, 'rules': RULES, 'was_active': 'active'}))
|
|
new_drop, new_rules, _ = m.plan(DROP, RULES)
|
|
self.drop.write_text(new_drop)
|
|
self.disk_rules = new_rules
|
|
self.live_rules = new_rules
|
|
self.ignore_restore_live = True
|
|
with self.assertRaises((RuntimeError, ValueError)):
|
|
self.run_main()
|
|
self.assertTrue(self.journal.exists(), 'crash recovery discarded unverified intent')
|
|
self.assertFalse(any('restart' in args for args, _ in self.calls))
|
|
|
|
def test_unobserved_restore_keeps_recovery_intent(self):
|
|
self.ignore_restore_live = True
|
|
original = self.command
|
|
failed = False
|
|
|
|
def fail_reload(*args, input=None):
|
|
nonlocal failed
|
|
if 'daemon-reload' in args and not failed:
|
|
failed = True
|
|
raise RuntimeError('fixture reload failure')
|
|
return original(*args, input=input)
|
|
|
|
self.command = fail_reload
|
|
with self.assertRaises(RuntimeError):
|
|
self.run_main()
|
|
self.assertTrue(self.journal.exists(), 'unverified rollback discarded recovery intent')
|
|
self.assertEqual(self.drop.read_text(), DROP)
|
|
|
|
def test_zero_exit_without_effective_firewall_is_not_success(self):
|
|
self.ignore_new_live = True
|
|
with self.assertRaisesRegex(RuntimeError, 'effective tunnel rules'):
|
|
self.run_main()
|
|
self.assertEqual(self.drop.read_text(), DROP)
|
|
self.assertEqual(self.disk_rules, RULES)
|
|
self.assertEqual(self.live_rules, RULES)
|
|
self.assertFalse(self.journal.exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|