"""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()