feat: add EgressManager — per-service egress enforcement via host iptables
Unit Tests / test (push) Successful in 11m20s
Unit Tests / test (push) Successful in 11m20s
Routes outbound traffic from installed service containers through alternate exits (wireguard_ext, openvpn, tor) using host-side iptables fwmark policy-routing in a dedicated PIC_EGRESS chain. Marks 0x110/0x120/0x130 are distinct from ConnectivityManager's 0x10/0x20/0x30. Container IPs discovered at runtime via docker inspect. Wired into ServiceStoreManager install/remove lifecycle and managers.py singleton. 22 new tests. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,586 @@
|
||||
"""
|
||||
Tests for EgressManager — per-service egress enforcement via host iptables.
|
||||
|
||||
All subprocess calls (iptables, iptables-save, iptables-restore, ip rule,
|
||||
docker inspect) and config_manager state are mocked so these tests run
|
||||
without any live infrastructure or root privileges.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch, call
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'api'))
|
||||
|
||||
import egress_manager as em_module
|
||||
from egress_manager import EgressManager, MARKS, TABLES, EXIT_TYPES, EGRESS_CHAIN
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_manager(installed=None, overrides=None):
|
||||
"""Build an EgressManager backed by a mock config_manager."""
|
||||
cm = MagicMock()
|
||||
cm.get_installed_services.return_value = installed or {}
|
||||
# Wire up configs dict so _get_egress_overrides / _set_egress_overrides work
|
||||
cm.configs = {'egress_overrides': overrides or {}}
|
||||
cm._save_all_configs = MagicMock()
|
||||
return EgressManager(config_manager=cm), cm
|
||||
|
||||
|
||||
def _subprocess_ok(stdout=''):
|
||||
"""Return a MagicMock simulating a successful subprocess.run result."""
|
||||
return MagicMock(returncode=0, stdout=stdout, stderr='')
|
||||
|
||||
|
||||
def _subprocess_fail(stderr='error', stdout=''):
|
||||
"""Return a MagicMock simulating a failed subprocess.run result."""
|
||||
return MagicMock(returncode=1, stdout=stdout, stderr=stderr)
|
||||
|
||||
|
||||
def _make_manifest(has_egress=True, egress_default='wireguard_ext',
|
||||
allowed=None, container_name='cell-myapp'):
|
||||
"""Return a minimal manifest dict with optional egress configuration."""
|
||||
m = {
|
||||
'id': 'myapp',
|
||||
'name': 'My App',
|
||||
'container_name': container_name,
|
||||
}
|
||||
if has_egress:
|
||||
m['has_egress'] = True
|
||||
m['egress'] = {
|
||||
'default': egress_default,
|
||||
'allowed': allowed if allowed is not None else list(EXIT_TYPES),
|
||||
}
|
||||
else:
|
||||
m['has_egress'] = False
|
||||
return m
|
||||
|
||||
|
||||
def _installed_with_manifest(manifest, service_id='myapp'):
|
||||
"""Return an installed-services dict containing one service record."""
|
||||
return {service_id: {'id': service_id, 'manifest': manifest}}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. test_apply_service_default_exit_no_iptables_calls
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyServiceDefaultExit(unittest.TestCase):
|
||||
|
||||
def test_apply_service_default_exit_no_iptables_calls(self):
|
||||
"""When egress.default is 'default', apply_service must not touch iptables."""
|
||||
manifest = _make_manifest(egress_default='default')
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
with patch('subprocess.run') as mock_run:
|
||||
# docker inspect must return an IP so we don't fail earlier
|
||||
mock_run.return_value = _subprocess_ok(stdout='172.20.0.50\n')
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertTrue(result['ok'])
|
||||
self.assertEqual(result.get('exit_via'), 'default')
|
||||
|
||||
# No iptables rule-insertion or mark call should have been made.
|
||||
# iptables-save from clear_service is allowed; we only check that
|
||||
# no iptables -A / -I (rule-adding) calls were made.
|
||||
rule_add_calls = [
|
||||
c for c in mock_run.call_args_list
|
||||
if c.args and c.args[0][:1] == ['iptables']
|
||||
and any(a in c.args[0] for a in ('-A', '-I', 'MARK', 'REDIRECT'))
|
||||
]
|
||||
self.assertEqual(rule_add_calls, [])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. test_apply_service_wireguard_ext_adds_mark_rule
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyServiceWireguardExt(unittest.TestCase):
|
||||
|
||||
def test_apply_service_wireguard_ext_adds_mark_rule(self):
|
||||
"""wireguard_ext exit must add a mangle MARK rule with 0x110 and the correct comment."""
|
||||
manifest = _make_manifest(egress_default='wireguard_ext')
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
calls_made = []
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
calls_made.append(cmd)
|
||||
# docker inspect → return IP
|
||||
if 'docker' in cmd and 'inspect' in cmd:
|
||||
return _subprocess_ok(stdout='172.20.0.50\n')
|
||||
# iptables-save → empty ruleset
|
||||
if 'iptables-save' in cmd:
|
||||
return _subprocess_ok(stdout='')
|
||||
# iptables-restore → success
|
||||
if 'iptables-restore' in cmd:
|
||||
return _subprocess_ok()
|
||||
# ip rule del → fail (none to delete)
|
||||
if cmd[:3] == ['ip', 'rule', 'del']:
|
||||
return _subprocess_fail()
|
||||
return _subprocess_ok()
|
||||
|
||||
with patch('subprocess.run', side_effect=fake_run):
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertTrue(result['ok'], result)
|
||||
self.assertEqual(result['exit_via'], 'wireguard_ext')
|
||||
|
||||
# Find the mangle MARK -A call
|
||||
mark_calls = [
|
||||
c for c in calls_made
|
||||
if 'iptables' in str(c) and 'MARK' in c and '--set-mark' in c
|
||||
]
|
||||
self.assertGreater(len(mark_calls), 0, 'No MARK rule was added')
|
||||
mark_cmd = ' '.join(mark_calls[0])
|
||||
self.assertIn('0x110', mark_cmd)
|
||||
self.assertIn('pic-egr-myapp', mark_cmd)
|
||||
self.assertIn('mangle', mark_cmd)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. test_apply_service_openvpn_adds_mark_rule
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyServiceOpenVPN(unittest.TestCase):
|
||||
|
||||
def test_apply_service_openvpn_adds_mark_rule(self):
|
||||
"""openvpn exit must add a mangle MARK rule with 0x120."""
|
||||
manifest = _make_manifest(egress_default='openvpn')
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
calls_made = []
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
calls_made.append(cmd)
|
||||
if 'docker' in cmd and 'inspect' in cmd:
|
||||
return _subprocess_ok(stdout='172.20.0.51\n')
|
||||
if 'iptables-save' in cmd:
|
||||
return _subprocess_ok(stdout='')
|
||||
if 'iptables-restore' in cmd:
|
||||
return _subprocess_ok()
|
||||
if cmd[:3] == ['ip', 'rule', 'del']:
|
||||
return _subprocess_fail()
|
||||
return _subprocess_ok()
|
||||
|
||||
with patch('subprocess.run', side_effect=fake_run):
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertTrue(result['ok'], result)
|
||||
self.assertEqual(result['exit_via'], 'openvpn')
|
||||
|
||||
mark_calls = [
|
||||
c for c in calls_made
|
||||
if 'iptables' in str(c) and 'MARK' in c and '--set-mark' in c
|
||||
]
|
||||
self.assertGreater(len(mark_calls), 0)
|
||||
self.assertIn('0x120', ' '.join(mark_calls[0]))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. test_apply_service_tor_adds_mark_and_redirect
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyServiceTor(unittest.TestCase):
|
||||
|
||||
def test_apply_service_tor_adds_mark_and_redirect(self):
|
||||
"""tor exit must add a mangle MARK 0x130 AND a nat REDIRECT to port 9040."""
|
||||
manifest = _make_manifest(egress_default='tor')
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
calls_made = []
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
calls_made.append(cmd)
|
||||
if 'docker' in cmd and 'inspect' in cmd:
|
||||
return _subprocess_ok(stdout='172.20.0.52\n')
|
||||
if 'iptables-save' in cmd:
|
||||
return _subprocess_ok(stdout='')
|
||||
if 'iptables-restore' in cmd:
|
||||
return _subprocess_ok()
|
||||
if cmd[:3] == ['ip', 'rule', 'del']:
|
||||
return _subprocess_fail()
|
||||
return _subprocess_ok()
|
||||
|
||||
with patch('subprocess.run', side_effect=fake_run):
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertTrue(result['ok'], result)
|
||||
self.assertEqual(result['exit_via'], 'tor')
|
||||
|
||||
mark_calls = [
|
||||
c for c in calls_made
|
||||
if 'iptables' in str(c) and 'MARK' in c and '--set-mark' in c
|
||||
]
|
||||
self.assertGreater(len(mark_calls), 0, 'No MARK rule found')
|
||||
self.assertIn('0x130', ' '.join(mark_calls[0]))
|
||||
|
||||
redirect_calls = [
|
||||
c for c in calls_made
|
||||
if 'iptables' in str(c) and 'REDIRECT' in c
|
||||
]
|
||||
self.assertGreater(len(redirect_calls), 0, 'No REDIRECT rule found')
|
||||
redirect_cmd = ' '.join(redirect_calls[0])
|
||||
self.assertIn('9040', redirect_cmd)
|
||||
self.assertIn('nat', redirect_cmd)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. test_apply_service_no_container_ip_returns_error
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyServiceNoContainerIP(unittest.TestCase):
|
||||
|
||||
def test_apply_service_no_container_ip_returns_error(self):
|
||||
"""When docker inspect returns an empty IP, apply_service must return ok=False."""
|
||||
manifest = _make_manifest(egress_default='wireguard_ext')
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
if 'docker' in cmd and 'inspect' in cmd:
|
||||
return _subprocess_ok(stdout='\n') # empty IP
|
||||
if 'iptables-save' in cmd:
|
||||
return _subprocess_ok(stdout='')
|
||||
if 'iptables-restore' in cmd:
|
||||
return _subprocess_ok()
|
||||
return _subprocess_ok()
|
||||
|
||||
with patch('subprocess.run', side_effect=fake_run):
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertFalse(result['ok'])
|
||||
self.assertIn('container IP not discoverable', result.get('error', ''))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6. test_apply_service_container_ip_retries
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyServiceRetries(unittest.TestCase):
|
||||
|
||||
def test_apply_service_container_ip_retries(self):
|
||||
"""First docker inspect attempt fails; second succeeds — result must be ok=True."""
|
||||
manifest = _make_manifest(egress_default='wireguard_ext')
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
inspect_count = [0]
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
if 'docker' in cmd and 'inspect' in cmd:
|
||||
inspect_count[0] += 1
|
||||
if inspect_count[0] == 1:
|
||||
return _subprocess_ok(stdout='\n') # first attempt: empty
|
||||
return _subprocess_ok(stdout='172.20.0.50\n') # second: success
|
||||
if 'iptables-save' in cmd:
|
||||
return _subprocess_ok(stdout='')
|
||||
if 'iptables-restore' in cmd:
|
||||
return _subprocess_ok()
|
||||
if cmd[:3] == ['ip', 'rule', 'del']:
|
||||
return _subprocess_fail()
|
||||
return _subprocess_ok()
|
||||
|
||||
with patch('subprocess.run', side_effect=fake_run):
|
||||
with patch('time.sleep'): # skip actual delays
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertTrue(result['ok'], result)
|
||||
self.assertGreaterEqual(inspect_count[0], 2)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 7. test_has_egress_false_skips_rules
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestHasEgressFalse(unittest.TestCase):
|
||||
|
||||
def test_has_egress_false_skips_rules(self):
|
||||
"""A manifest with has_egress=False must skip rules and return skipped=True."""
|
||||
manifest = _make_manifest(has_egress=False)
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
with patch('subprocess.run') as mock_run:
|
||||
mock_run.return_value = _subprocess_ok(stdout='')
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertTrue(result['ok'])
|
||||
self.assertTrue(result.get('skipped'))
|
||||
|
||||
# No iptables rule-insertion call should have been made.
|
||||
# iptables-save from clear_service is permitted; only check no -A/-I.
|
||||
rule_add_calls = [
|
||||
c for c in mock_run.call_args_list
|
||||
if c.args and c.args[0][:1] == ['iptables']
|
||||
and any(a in c.args[0] for a in ('-A', '-I', 'MARK', 'REDIRECT'))
|
||||
]
|
||||
self.assertEqual(rule_add_calls, [])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 8. test_has_egress_missing_egress_block_skips
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestHasEgressMissingBlock(unittest.TestCase):
|
||||
|
||||
def test_has_egress_missing_egress_block_skips(self):
|
||||
"""has_egress=True but no 'egress' dict → must skip (skipped=True)."""
|
||||
manifest = {
|
||||
'id': 'myapp',
|
||||
'container_name': 'cell-myapp',
|
||||
'has_egress': True,
|
||||
# 'egress' key intentionally absent
|
||||
}
|
||||
mgr, _ = _make_manager(
|
||||
installed=_installed_with_manifest(manifest)
|
||||
)
|
||||
|
||||
with patch('subprocess.run') as mock_run:
|
||||
mock_run.return_value = _subprocess_ok(stdout='')
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertTrue(result['ok'])
|
||||
self.assertTrue(result.get('skipped'))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 9. test_clear_service_removes_tagged_rules
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestClearService(unittest.TestCase):
|
||||
|
||||
def test_clear_service_removes_tagged_rules(self):
|
||||
"""iptables-restore is called with the tagged lines removed."""
|
||||
mgr, _ = _make_manager()
|
||||
|
||||
mangle_rules = (
|
||||
'-A PIC_EGRESS -s 172.20.0.50 -j MARK --set-mark 0x110 '
|
||||
'-m comment --comment "pic-egr-myapp"\n'
|
||||
'-A PIC_EGRESS -s 172.20.0.99 -j MARK --set-mark 0x110 '
|
||||
'-m comment --comment "pic-egr-otherapp"\n'
|
||||
)
|
||||
nat_rules = ''
|
||||
|
||||
restore_inputs = {}
|
||||
|
||||
def fake_run(cmd, input=None, **kwargs):
|
||||
if cmd == ['iptables-save', '-t', 'mangle']:
|
||||
return _subprocess_ok(stdout=mangle_rules)
|
||||
if cmd == ['iptables-save', '-t', 'nat']:
|
||||
return _subprocess_ok(stdout=nat_rules)
|
||||
if cmd == ['iptables-restore', '-T', 'mangle']:
|
||||
restore_inputs['mangle'] = input
|
||||
return _subprocess_ok()
|
||||
if cmd == ['iptables-restore', '-T', 'nat']:
|
||||
restore_inputs['nat'] = input
|
||||
return _subprocess_ok()
|
||||
return _subprocess_ok()
|
||||
|
||||
with patch('subprocess.run', side_effect=fake_run):
|
||||
result = mgr.clear_service('myapp')
|
||||
|
||||
self.assertTrue(result['ok'])
|
||||
# The restored mangle rules must not contain myapp's tag
|
||||
restored = restore_inputs.get('mangle', '')
|
||||
self.assertNotIn('pic-egr-myapp', restored)
|
||||
# But the other service's rules must be preserved
|
||||
self.assertIn('pic-egr-otherapp', restored)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 10. test_set_service_exit_rejects_not_in_allowed
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSetServiceExitRejectNotAllowed(unittest.TestCase):
|
||||
|
||||
def test_set_service_exit_rejects_not_in_allowed(self):
|
||||
"""Exit type not in manifest's allowed list must return ok=False."""
|
||||
manifest = _make_manifest(
|
||||
egress_default='default',
|
||||
allowed=['default', 'tor'], # wireguard_ext not in allowed
|
||||
)
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
result = mgr.set_service_exit('myapp', 'wireguard_ext')
|
||||
|
||||
self.assertFalse(result['ok'])
|
||||
self.assertIn('error', result)
|
||||
self.assertIn('allowed', result['error'])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 11. test_set_service_exit_persists_and_applies
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSetServiceExitPersistsAndApplies(unittest.TestCase):
|
||||
|
||||
def test_set_service_exit_persists_and_applies(self):
|
||||
"""Valid override must be persisted to config_manager and apply_service called."""
|
||||
manifest = _make_manifest(egress_default='default', allowed=list(EXIT_TYPES))
|
||||
mgr, cm = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
apply_calls = []
|
||||
original_apply = mgr.apply_service
|
||||
|
||||
def fake_apply(sid):
|
||||
apply_calls.append(sid)
|
||||
return {'ok': True, 'exit_via': 'tor'}
|
||||
|
||||
mgr.apply_service = fake_apply
|
||||
|
||||
result = mgr.set_service_exit('myapp', 'tor')
|
||||
|
||||
self.assertTrue(result['ok'], result)
|
||||
# apply_service was called
|
||||
self.assertIn('myapp', apply_calls)
|
||||
# override was persisted
|
||||
cm._save_all_configs.assert_called()
|
||||
self.assertEqual(cm.configs['egress_overrides'].get('myapp'), 'tor')
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 12. test_apply_all_iterates_installed_services
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyAll(unittest.TestCase):
|
||||
|
||||
def test_apply_all_iterates_installed_services(self):
|
||||
"""apply_all must call apply_service for every service with a manifest."""
|
||||
manifests = {
|
||||
'svc1': _make_manifest(egress_default='wireguard_ext'),
|
||||
'svc2': _make_manifest(egress_default='openvpn'),
|
||||
'svc3': _make_manifest(egress_default='tor'),
|
||||
}
|
||||
installed = {
|
||||
sid: {'id': sid, 'manifest': m}
|
||||
for sid, m in manifests.items()
|
||||
}
|
||||
mgr, _ = _make_manager(installed=installed)
|
||||
|
||||
applied = []
|
||||
mgr.apply_service = lambda sid: applied.append(sid) or {'ok': True}
|
||||
|
||||
result = mgr.apply_all()
|
||||
|
||||
self.assertTrue(result['ok'])
|
||||
self.assertEqual(sorted(applied), ['svc1', 'svc2', 'svc3'])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 13. test_marks_do_not_collide_with_connectivity_manager
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestMarksNoCollision(unittest.TestCase):
|
||||
|
||||
def test_marks_do_not_collide_with_connectivity_manager(self):
|
||||
"""EgressManager marks must be disjoint from ConnectivityManager marks."""
|
||||
connectivity_marks = {0x10, 0x20, 0x30}
|
||||
egress_mark_values = set(MARKS.values())
|
||||
collision = connectivity_marks & egress_mark_values
|
||||
self.assertEqual(
|
||||
collision, set(),
|
||||
f'Mark collision with ConnectivityManager: {collision}',
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 14. test_apply_service_unknown_exit_in_allowed_rejected
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyServiceUnknownExit(unittest.TestCase):
|
||||
|
||||
def test_apply_service_unknown_exit_in_allowed_rejected(self):
|
||||
"""An egress.default value that is not a known EXIT_TYPE must return ok=False."""
|
||||
manifest = {
|
||||
'id': 'myapp',
|
||||
'container_name': 'cell-myapp',
|
||||
'has_egress': True,
|
||||
'egress': {
|
||||
'default': 'internet_fast_lane', # unknown exit
|
||||
'allowed': ['internet_fast_lane'],
|
||||
},
|
||||
}
|
||||
mgr, _ = _make_manager(installed=_installed_with_manifest(manifest))
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
if 'docker' in cmd and 'inspect' in cmd:
|
||||
return _subprocess_ok(stdout='172.20.0.50\n')
|
||||
if 'iptables-save' in cmd:
|
||||
return _subprocess_ok(stdout='')
|
||||
if 'iptables-restore' in cmd:
|
||||
return _subprocess_ok()
|
||||
return _subprocess_ok()
|
||||
|
||||
with patch('subprocess.run', side_effect=fake_run):
|
||||
result = mgr.apply_service('myapp')
|
||||
|
||||
self.assertFalse(result['ok'])
|
||||
self.assertIn('error', result)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Additional coverage: _has_egress edge cases
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestHasEgressLogic(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.mgr, _ = _make_manager()
|
||||
|
||||
def test_has_egress_both_required(self):
|
||||
"""Both has_egress=True and non-empty egress dict required."""
|
||||
m = {'has_egress': True, 'egress': {'default': 'tor', 'allowed': ['tor']}}
|
||||
self.assertTrue(self.mgr._has_egress(m))
|
||||
|
||||
def test_has_egress_false_field(self):
|
||||
m = {'has_egress': False, 'egress': {'default': 'tor', 'allowed': ['tor']}}
|
||||
self.assertFalse(self.mgr._has_egress(m))
|
||||
|
||||
def test_has_egress_missing_has_egress_key(self):
|
||||
m = {'egress': {'default': 'tor', 'allowed': ['tor']}}
|
||||
self.assertFalse(self.mgr._has_egress(m))
|
||||
|
||||
def test_has_egress_empty_egress_dict(self):
|
||||
m = {'has_egress': True, 'egress': {}}
|
||||
self.assertFalse(self.mgr._has_egress(m))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Additional coverage: _resolve_exit
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestResolveExit(unittest.TestCase):
|
||||
|
||||
def test_override_takes_precedence(self):
|
||||
mgr, _ = _make_manager(overrides={'myapp': 'openvpn'})
|
||||
manifest = _make_manifest(egress_default='wireguard_ext')
|
||||
self.assertEqual(mgr._resolve_exit('myapp', manifest), 'openvpn')
|
||||
|
||||
def test_manifest_default_used_when_no_override(self):
|
||||
mgr, _ = _make_manager(overrides={})
|
||||
manifest = _make_manifest(egress_default='tor')
|
||||
self.assertEqual(mgr._resolve_exit('myapp', manifest), 'tor')
|
||||
|
||||
def test_fallback_to_default_when_no_egress_block(self):
|
||||
mgr, _ = _make_manager(overrides={})
|
||||
manifest = {'id': 'myapp'}
|
||||
self.assertEqual(mgr._resolve_exit('myapp', manifest), 'default')
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Additional: apply_service with missing manifest
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApplyServiceMissingManifest(unittest.TestCase):
|
||||
|
||||
def test_apply_service_missing_manifest_returns_error(self):
|
||||
mgr, _ = _make_manager(installed={})
|
||||
result = mgr.apply_service('ghost')
|
||||
self.assertFalse(result['ok'])
|
||||
self.assertIn('error', result)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user