roof/pic
1
0
Fork 0
Code Issues Pull Requests Actions Wiki Activity
Files
fa00c90328719a80ff12bc94f6af292dbd7a4b7b
pic/tests/test_connectivity_connections.py
T
roof 603225694c
Unit Tests / test (push) Successful in 13m5s
Details
feat: connectivity redesign phase 5 — one container per connection instance 
instanceable rendering, per-instance up/down on create/delete,
store-service-installed gate, per-instance health

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-06-10 22:56:31 -04:00

636 lines
26 KiB
Python
Raw Blame History

"""
Tests for the connectivity v2 data model — named connection instances.
Covers the resource allocator (no mark/table/port collisions, correct iface
naming, iface vs redirect-port split), the v1→v2 migration (legacy exits →
one connection each, secret repointing, idempotency, empty case), connection
CRUD (validation, single-tor, duplicate name, secret refs not values, delete
frees resources + vault secret, delete blocked when referenced), and computed
status.state (added vs configured).
ConfigManager runs against a tmp dir; the vault is an in-memory fake. No real
Docker/iptables/subprocess is invoked (apply_routes is never called here).
"""
import os
import sys
import shutil
import tempfile
import unittest
from unittest.mock import MagicMock, patch
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'api'))
from config_manager import ConfigManager
from connectivity_manager import ConnectivityManager
VALID_KEY = (
'-----BEGIN OPENSSH PRIVATE KEY-----\n'
'b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAAB\n'
'-----END OPENSSH PRIVATE KEY-----\n'
)
VALID_KNOWN_HOSTS = (
'ssh.example.com,203.0.113.5 ssh-ed25519 '
'AAAAC3NzaC1lZDI1NTE5AAAAIB5d0o0Yw1xP1Yw1xP1Yw1xP1Yw1xP1Yw1xP1Yw1xP1Y'
)
class FakeVault:
"""In-memory stand-in for VaultManager's secret store."""
def __init__(self):
self.store = {}
def store_secret(self, name, value):
self.store[name] = value
return True
def get_secret(self, name):
return self.store.get(name)
def delete_secret(self, name):
if name in self.store:
del self.store[name]
return True
return False
def list_secrets(self):
return list(self.store.keys())
def _sshuttle_cfg(**overrides):
cfg = {
'host': 'ssh.example.com',
'port': 22,
'user': 'tunnel',
'auth': 'key',
'known_hosts': VALID_KNOWN_HOSTS,
'exclude_subnets': ['10.0.0.0/8'],
}
cfg.update(overrides)
return cfg
def _proxy_cfg(**overrides):
cfg = {'scheme': 'http', 'host': 'proxy.example.com', 'port': 3128}
cfg.update(overrides)
return cfg
class _Base(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.mkdtemp()
self.cfg_file = os.path.join(self.tmp, 'cell_config.json')
self.data_dir = os.path.join(self.tmp, 'data')
os.makedirs(self.data_dir, exist_ok=True)
self.cm = ConfigManager(self.cfg_file, self.data_dir)
self.vault = FakeVault()
self.peer_registry = MagicMock()
self.peer_registry.list_peers.return_value = []
with patch.object(ConnectivityManager, '_subscribe_to_events',
lambda self: None):
self.mgr = ConnectivityManager(
config_manager=self.cm,
peer_registry=self.peer_registry,
vault_manager=self.vault,
data_dir=self.data_dir,
config_dir=self.tmp,
)
# Start CRUD tests from a clean migrated v2 (no legacy exits). Migration
# is exercised separately in TestMigration; here we want an empty slate
# so the always-configured Tor default doesn't get auto-migrated in.
self.cm.configs['connectivity'] = {'version': 2, 'connections': []}
# P5 gate: a connection's backing store service must be installed before
# it can be created. Mark all connectivity store services installed so
# the CRUD tests below exercise allocation/persistence, not the gate
# (the gate itself is tested explicitly in TestStoreServiceGate). The
# per-instance container up/down is a no-op here: the manager has no
# service_composer wired (service_composer=None), so up_connection_instance
# short-circuits without touching Docker.
for _svc in ('wireguard-ext', 'openvpn-client', 'tor', 'sshuttle', 'proxy'):
self.cm.set_installed_service(_svc, {'id': _svc, 'manifest': {}})
self.cm._save_all_configs()
def tearDown(self):
shutil.rmtree(self.tmp, ignore_errors=True)
# ---------------------------------------------------------------------------
# Resource allocator
# ---------------------------------------------------------------------------
class TestAllocator(_Base):
def test_iface_types_get_iface_not_port(self):
for ctype in ('wireguard_ext', 'openvpn'):
cid = 'conn_deadbeef'
mark, table, iface, port = self.mgr._allocate_resources(ctype, cid)
self.assertIsNotNone(iface)
self.assertIsNone(port)
self.assertTrue(iface.startswith(
self.mgr.IFACE_PREFIXES[ctype]))
self.assertLessEqual(len(iface), 15)
def test_redirect_types_get_port_not_iface(self):
for ctype in ('tor', 'sshuttle', 'proxy'):
cid = 'conn_abcd1234'
mark, table, iface, port = self.mgr._allocate_resources(ctype, cid)
self.assertIsNone(iface)
self.assertIsNotNone(port)
self.assertTrue(
self.mgr.REDIRECT_PORT_BASE <= port <= self.mgr.REDIRECT_PORT_MAX)
def test_iface_uses_8hex_id(self):
_m, _t, iface, _p = self.mgr._allocate_resources(
'wireguard_ext', 'conn_0123456789')
self.assertEqual(iface, 'wgext_01234567')
def test_no_collisions_across_many_connections(self):
created = []
# Mix of iface + redirect types; tor is single-instance so exclude it.
plan = (['wireguard_ext', 'openvpn', 'sshuttle', 'proxy'] * 4)
for i, ctype in enumerate(plan):
if ctype == 'sshuttle':
res = self.mgr.create_connection(
ctype, f'ssh-{i}', _sshuttle_cfg(),
secrets={'private_key': VALID_KEY})
elif ctype == 'proxy':
res = self.mgr.create_connection(ctype, f'px-{i}', _proxy_cfg())
elif ctype == 'wireguard_ext':
res = self.mgr.create_connection(
ctype, f'wg-{i}', {},
secrets={'conf': '[Interface]\nPrivateKey = x\n'})
else:
res = self.mgr.create_connection(
ctype, f'ov-{i}', {},
secrets={'conf': 'client\nremote vpn.example.com 1194\n'})
self.assertTrue(res['ok'], res)
created.append(res['connection'])
marks = [c['mark'] for c in created]
tables = [c['table'] for c in created]
ifaces = [c['iface'] for c in created if c['iface']]
ports = [c['redirect_port'] for c in created if c['redirect_port']]
self.assertEqual(len(marks), len(set(marks)))
self.assertEqual(len(tables), len(set(tables)))
self.assertEqual(len(ifaces), len(set(ifaces)))
self.assertEqual(len(ports), len(set(ports)))
for m in marks:
self.assertEqual(m % self.mgr.MARK_STRIDE, 0)
self.assertTrue(self.mgr.MARK_BASE <= m <= self.mgr.MARK_MAX)
for t in tables:
self.assertGreaterEqual(t, self.mgr.TABLE_BASE)
# ---------------------------------------------------------------------------
# Connection CRUD
# ---------------------------------------------------------------------------
class TestCreateConnection(_Base):
def test_create_proxy_allocates_and_persists(self):
res = self.mgr.create_connection('proxy', 'Work proxy', _proxy_cfg())
self.assertTrue(res['ok'], res)
rec = res['connection']
self.assertTrue(rec['id'].startswith('conn_'))
self.assertEqual(rec['type'], 'proxy')
self.assertIsNotNone(rec['redirect_port'])
self.assertIsNone(rec['iface'])
# Persisted to config_manager.
self.assertEqual(len(self.cm.list_connections()), 1)
def test_secret_stored_as_ref_not_value(self):
res = self.mgr.create_connection(
'sshuttle', 'tun', _sshuttle_cfg(),
secrets={'private_key': VALID_KEY})
self.assertTrue(res['ok'], res)
rec = res['connection']
cid = rec['id']
# Refs recorded, values absent from the record + config.
self.assertIn(f'{cid}_private_key', rec['secret_refs'])
self.assertIn(f'{cid}_known_hosts', rec['secret_refs'])
self.assertNotIn('private_key', rec['config'])
stored = self.cm.get_connection(cid)
self.assertNotIn('private_key', stored['config'])
# Secret value lives only in the vault.
self.assertEqual(self.vault.get_secret(f'{cid}_private_key'), VALID_KEY)
def test_invalid_type_rejected(self):
res = self.mgr.create_connection('bogus', 'x', {})
self.assertFalse(res['ok'])
def test_invalid_name_rejected(self):
res = self.mgr.create_connection('proxy', '', _proxy_cfg())
self.assertFalse(res['ok'])
def test_invalid_config_rejected(self):
res = self.mgr.create_connection('proxy', 'p', _proxy_cfg(scheme='ftp'))
self.assertFalse(res['ok'])
self.assertEqual(len(self.cm.list_connections()), 0)
def test_single_tor_enforced(self):
first = self.mgr.create_connection('tor', 'Tor 1', {})
self.assertTrue(first['ok'], first)
second = self.mgr.create_connection('tor', 'Tor 2', {})
self.assertFalse(second['ok'])
self.assertIn('single', second['error'].lower())
def test_duplicate_name_rejected(self):
a = self.mgr.create_connection('proxy', 'Dup', _proxy_cfg())
self.assertTrue(a['ok'])
b = self.mgr.create_connection('proxy', 'dup', _proxy_cfg(port=8080))
self.assertFalse(b['ok'])
self.assertIn('already exists', b['error'])
def test_state_configured_when_complete(self):
res = self.mgr.create_connection('proxy', 'full', _proxy_cfg())
self.assertEqual(res['connection']['status']['state'], 'configured')
self.assertEqual(res['connection']['status']['health'], 'unknown')
def test_state_added_when_missing_secret(self):
# wireguard_ext with no conf secret → required secret missing → added.
# (validator requires conf, so create one that omits it must fail; to
# exercise the 'added' path we build a record directly via migration of
# an installed-but-unconfigured exit — see TestStatusState.)
res = self.mgr.create_connection(
'wireguard_ext', 'wgfull', {},
secrets={'conf': '[Interface]\nPrivateKey = x\n'})
self.assertTrue(res['ok'], res)
self.assertEqual(res['connection']['status']['state'], 'configured')
class TestDeleteConnection(_Base):
def test_delete_frees_resources_and_secret(self):
res = self.mgr.create_connection(
'sshuttle', 'tun', _sshuttle_cfg(),
secrets={'private_key': VALID_KEY})
cid = res['connection']['id']
self.assertIn(f'{cid}_private_key', self.vault.list_secrets())
out = self.mgr.delete_connection(cid)
self.assertTrue(out['ok'], out)
self.assertEqual(len(self.cm.list_connections()), 0)
self.assertNotIn(f'{cid}_private_key', self.vault.list_secrets())
self.assertNotIn(f'{cid}_known_hosts', self.vault.list_secrets())
def test_freed_resources_are_reusable(self):
a = self.mgr.create_connection('proxy', 'A', _proxy_cfg())
mark_a = a['connection']['mark']
self.mgr.delete_connection(a['connection']['id'])
b = self.mgr.create_connection('proxy', 'B', _proxy_cfg())
self.assertEqual(b['connection']['mark'], mark_a)
def test_delete_blocked_when_referenced(self):
res = self.mgr.create_connection('proxy', 'ref', _proxy_cfg())
cid = res['connection']['id']
self.peer_registry.list_peers.return_value = [
{'peer': 'alice', 'exit_via': cid}]
out = self.mgr.delete_connection(cid)
self.assertFalse(out['ok'])
self.assertIn('in use', out['error'])
# Still present, secret intact.
self.assertEqual(len(self.cm.list_connections()), 1)
def test_delete_unknown_returns_error(self):
out = self.mgr.delete_connection('conn_nope')
self.assertFalse(out['ok'])
class TestUpdateConnection(_Base):
def test_update_name(self):
res = self.mgr.create_connection('proxy', 'old', _proxy_cfg())
cid = res['connection']['id']
out = self.mgr.update_connection(cid, name='new')
self.assertTrue(out['ok'], out)
self.assertEqual(self.cm.get_connection(cid)['name'], 'new')
def test_update_duplicate_name_rejected(self):
a = self.mgr.create_connection('proxy', 'A', _proxy_cfg())
b = self.mgr.create_connection('proxy', 'B', _proxy_cfg(port=8080))
out = self.mgr.update_connection(b['connection']['id'], name='A')
self.assertFalse(out['ok'])
def test_update_secret_repoints_vault(self):
res = self.mgr.create_connection(
'sshuttle', 'tun', _sshuttle_cfg(),
secrets={'private_key': VALID_KEY})
cid = res['connection']['id']
new_key = VALID_KEY.replace('AAAAB', 'BBBBB')
out = self.mgr.update_connection(
cid, config=_sshuttle_cfg(), secrets={'private_key': new_key})
self.assertTrue(out['ok'], out)
self.assertEqual(self.vault.get_secret(f'{cid}_private_key'), new_key)
# ---------------------------------------------------------------------------
# list/get enrichment
# ---------------------------------------------------------------------------
class TestListGet(_Base):
def test_list_and_get_strip_secret_values(self):
res = self.mgr.create_connection(
'sshuttle', 'tun', _sshuttle_cfg(),
secrets={'private_key': VALID_KEY})
cid = res['connection']['id']
listed = self.mgr.list_connections()
self.assertEqual(len(listed), 1)
self.assertNotIn('private_key', listed[0]['config'])
self.assertEqual(listed[0]['status']['health'], 'unknown')
got = self.mgr.get_connection(cid)
self.assertEqual(got['id'], cid)
self.assertNotIn('private_key', got['config'])
def test_get_unknown_returns_none(self):
self.assertIsNone(self.mgr.get_connection('conn_missing'))
# ---------------------------------------------------------------------------
# Computed status.state
# ---------------------------------------------------------------------------
class TestStatusState(_Base):
def test_compute_state_proxy(self):
self.assertEqual(
self.mgr._compute_state('proxy', _proxy_cfg(), []), 'configured')
self.assertEqual(
self.mgr._compute_state('proxy', {'scheme': 'http'}, []), 'added')
def test_compute_state_sshuttle_password_auth(self):
cfg = _sshuttle_cfg(auth='password')
cid = 'conn_x'
self.assertEqual(
self.mgr._compute_state(cfg['auth'] and 'sshuttle', cfg,
[f'{cid}_known_hosts', f'{cid}_password']),
'configured')
self.assertEqual(
self.mgr._compute_state('sshuttle', cfg, [f'{cid}_known_hosts']),
'added')
def test_compute_state_tor_always_configured(self):
self.assertEqual(self.mgr._compute_state('tor', {}, []), 'configured')
# ---------------------------------------------------------------------------
# v1 → v2 migration
# ---------------------------------------------------------------------------
class TestMigration(_Base):
def test_empty_legacy_becomes_version2_empty(self):
# No exits configured: _exit_status reports nothing configured.
self.cm.configs['connectivity'] = {'exits': {}, 'peer_exit_map': {}}
with patch.object(self.mgr, '_exit_status',
return_value={'configured': False}):
v2 = self.cm.get_connectivity()
self.assertEqual(v2['version'], 2)
self.assertEqual(v2['connections'], [])
def test_each_configured_exit_becomes_one_connection(self):
configured = {'sshuttle', 'proxy', 'tor'}
def fake_status(exit_type):
return {'configured': exit_type in configured}
# Seed legacy config + secrets.
self.cm.configs['connectivity'] = {
'exits': {
'sshuttle': {'host': 'h', 'port': 22, 'user': 'u',
'auth': 'key', 'exclude_subnets': []},
'proxy': {'scheme': 'http', 'host': 'p', 'port': 3128, 'user': ''},
},
'peer_exit_map': {},
}
self.vault.store_secret('connectivity_sshuttle_key', VALID_KEY)
with patch.object(self.mgr, '_exit_status', side_effect=fake_status):
v2 = self.cm.get_connectivity()
self.assertEqual(v2['version'], 2)
types = sorted(c['type'] for c in v2['connections'])
self.assertEqual(types, ['proxy', 'sshuttle', 'tor'])
# Exactly one per type.
self.assertEqual(len(v2['connections']), 3)
def test_secret_repointed_not_lost(self):
self.cm.configs['connectivity'] = {
'exits': {'sshuttle': {'host': 'h', 'port': 22, 'user': 'u',
'auth': 'key', 'exclude_subnets': []}},
'peer_exit_map': {},
}
self.vault.store_secret('connectivity_sshuttle_key', VALID_KEY)
def fake_status(exit_type):
return {'configured': exit_type == 'sshuttle'}
with patch.object(self.mgr, '_exit_status', side_effect=fake_status):
v2 = self.cm.get_connectivity()
conn = next(c for c in v2['connections'] if c['type'] == 'sshuttle')
cid = conn['id']
new_ref = f'{cid}_private_key'
# Old name gone, new name holds the value — nothing lost.
self.assertIsNone(self.vault.get_secret('connectivity_sshuttle_key'))
self.assertEqual(self.vault.get_secret(new_ref), VALID_KEY)
self.assertIn(new_ref, conn['secret_refs'])
def test_migration_idempotent(self):
def fake_status(exit_type):
return {'configured': exit_type == 'proxy'}
self.cm.configs['connectivity'] = {
'exits': {'proxy': {'scheme': 'http', 'host': 'p', 'port': 3128}},
'peer_exit_map': {},
}
with patch.object(self.mgr, '_exit_status', side_effect=fake_status):
first = self.cm.get_connectivity()
ids_first = [c['id'] for c in first['connections']]
# Second access must not re-run migration / duplicate.
second = self.cm.get_connectivity()
ids_second = [c['id'] for c in second['connections']]
self.assertEqual(ids_first, ids_second)
self.assertEqual(len(second['connections']), 1)
def test_migration_allocates_distinct_resources(self):
configured = {'wireguard_ext', 'openvpn', 'sshuttle', 'proxy', 'tor'}
self.cm.configs['connectivity'] = {'exits': {}, 'peer_exit_map': {}}
def fake_status(exit_type):
return {'configured': exit_type in configured}
with patch.object(self.mgr, '_exit_status', side_effect=fake_status):
v2 = self.cm.get_connectivity()
conns = v2['connections']
marks = [c['mark'] for c in conns]
tables = [c['table'] for c in conns]
self.assertEqual(len(marks), len(set(marks)))
self.assertEqual(len(tables), len(set(tables)))
for c in conns:
if c['type'] in ('wireguard_ext', 'openvpn'):
self.assertIsNotNone(c['iface'])
self.assertIsNone(c['redirect_port'])
else:
self.assertIsNone(c['iface'])
self.assertIsNotNone(c['redirect_port'])
# ---------------------------------------------------------------------------
# P5 — per-instance container lifecycle wiring
# ---------------------------------------------------------------------------
_PROXY_TEMPLATE = (
'services:\n'
' proxy:\n'
' image: git.pic.ngo/roof/svc-redsocks:latest\n'
' container_name: cell-proxy-${INSTANCE_ID}\n'
' network_mode: host\n'
' cap_add:\n'
' - NET_ADMIN\n'
' environment:\n'
' - REDSOCKS_LOCAL_PORT=${REDIRECT_PORT}\n'
' volumes:\n'
' - ${PIC_DATA_DIR}/services/proxy/${INSTANCE_ID}/config:/config\n'
)
_SSHUTTLE_TEMPLATE = (
'services:\n'
' sshuttle:\n'
' image: git.pic.ngo/roof/svc-sshuttle:latest\n'
' container_name: cell-sshuttle-${INSTANCE_ID}\n'
' network_mode: host\n'
' cap_add:\n'
' - NET_ADMIN\n'
' environment:\n'
' - SSHUTTLE_LISTEN_PORT=${REDIRECT_PORT}\n'
' volumes:\n'
' - ${PIC_DATA_DIR}/services/sshuttle/${INSTANCE_ID}/config:/config\n'
)
class _ComposerBase(_Base):
"""Base that wires a real ServiceComposer (subprocess mocked) + install records
that carry a compose template, so create/delete drive per-instance up/down."""
TEMPLATES = {
'proxy': _PROXY_TEMPLATE,
'sshuttle': _SSHUTTLE_TEMPLATE,
}
def setUp(self):
super().setUp()
from service_composer import ServiceComposer
cm = MagicMock()
ident = {'cell_name': 'c', 'domain': 'cell.local', 'domain_mode': 'lan'}
cm.get_identity.return_value = ident
cm.get_effective_domain.return_value = 'cell.local'
cm.configs = {}
self.composer = ServiceComposer(config_manager=cm, data_dir=self.data_dir)
self.mgr.service_composer = self.composer
# Re-record install records with compose templates so up_connection_instance
# finds an image + template (the _Base records had empty manifests).
svc_template = {
'wireguard-ext': _SSHUTTLE_TEMPLATE, # content irrelevant; render only
'openvpn-client': _SSHUTTLE_TEMPLATE,
'sshuttle': _SSHUTTLE_TEMPLATE,
'proxy': _PROXY_TEMPLATE,
'tor': None,
}
for svc, tmpl in svc_template.items():
rec = {'id': svc, 'manifest': {'instanceable': bool(tmpl),
'requires_host_network': True}}
if tmpl:
rec['compose_template'] = tmpl
self.cm.set_installed_service(svc, rec)
self.cm._save_all_configs()
class TestCreateBringsUpInstance(_ComposerBase):
@patch('service_composer.subprocess.run')
def test_create_proxy_triggers_up_and_writes_compose(self, mock_run):
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
res = self.mgr.create_connection('proxy', 'Work proxy', _proxy_cfg())
self.assertTrue(res['ok'], res)
cid = res['connection']['id']
inst = cid.split('_')[-1][:12]
# up_instance ran docker compose up for the per-instance project.
cmds = [c.args[0] for c in mock_run.call_args_list]
self.assertTrue(any('up' in cmd and f'pic-conn-{inst}' in cmd
for cmd in cmds), cmds)
# Per-instance compose written to its own dir, not the type-level path.
self.assertTrue(self.composer.has_instance_compose('proxy', inst))
compose_path = self.composer._instance_compose_path('proxy', inst)
with open(compose_path) as f:
content = f.read()
self.assertIn(f'cell-proxy-{inst}', content)
self.assertIn(str(res['connection']['redirect_port']), content)
# redsocks.conf materialized in the per-instance config dir.
self.assertTrue(os.path.exists(os.path.join(
self.composer.instance_config_dir('proxy', inst), 'redsocks.conf')))
@patch('service_composer.subprocess.run')
def test_two_proxies_distinct_containers_and_ports(self, mock_run):
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
r1 = self.mgr.create_connection('proxy', 'p1', _proxy_cfg())
r2 = self.mgr.create_connection('proxy', 'p2', _proxy_cfg())
self.assertTrue(r1['ok'] and r2['ok'])
n1 = self.mgr.instance_container_name(r1['connection'])
n2 = self.mgr.instance_container_name(r2['connection'])
self.assertNotEqual(n1, n2)
self.assertNotEqual(r1['connection']['redirect_port'],
r2['connection']['redirect_port'])
@patch('service_composer.subprocess.run')
def test_delete_brings_down_and_cleans_instance(self, mock_run):
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
res = self.mgr.create_connection('sshuttle', 'tun', _sshuttle_cfg(),
secrets={'private_key': VALID_KEY})
cid = res['connection']['id']
inst = cid.split('_')[-1][:12]
self.assertTrue(self.composer.has_instance_compose('sshuttle', inst))
inst_dir = self.composer._instance_dir('sshuttle', inst)
mock_run.reset_mock()
d = self.mgr.delete_connection(cid)
self.assertTrue(d['ok'], d)
cmds = [c.args[0] for c in mock_run.call_args_list]
self.assertTrue(any('down' in cmd and f'pic-conn-{inst}' in cmd
for cmd in cmds), cmds)
self.assertFalse(os.path.exists(inst_dir))
class TestStoreServiceGate(_Base):
def test_create_errors_when_store_service_not_installed(self):
# _Base marks all installed; remove proxy to hit the gate.
self.cm.remove_installed_service('proxy')
res = self.mgr.create_connection('proxy', 'p', _proxy_cfg())
self.assertFalse(res['ok'])
self.assertIn('Service Store', res['error'])
self.assertEqual(len(self.cm.list_connections()), 0)
class TestTorSingleInstance(_ComposerBase):
@patch('service_composer.subprocess.run')
def test_tor_uses_fixed_container_and_no_instance_up(self, mock_run):
mock_run.return_value = MagicMock(returncode=0, stdout='', stderr='')
res = self.mgr.create_connection('tor', 'Tor')
self.assertTrue(res['ok'], res)
rec = res['connection']
# Single fixed container name, not cell-tor-<id>.
self.assertEqual(self.mgr.instance_container_name(rec), 'cell-tor')
# No per-instance compose written for tor.
inst = rec['id'].split('_')[-1][:12]
self.assertFalse(self.composer.has_instance_compose('tor', inst))
# A second tor is rejected (single instance).
res2 = self.mgr.create_connection('tor', 'Tor2')
self.assertFalse(res2['ok'])
if __name__ == '__main__':
unittest.main()
Reference in New Issue View Git Blame Copy Permalink