""" Phase 4 tests for the generic connection CRUD REST API and the per-peer fail-open endpoint. Logic lives in ConnectivityManager (mocked here); these tests assert the thin route layer: status codes, error mapping (404/409/400), that secrets are never echoed, and that admin-only enforcement applies. The `client` fixture sets TESTING=True (bypassing auth/CSRF) for happy-path status-code checks; admin-only enforcement is verified separately against a real seeded AuthManager with TESTING off. """ import os import sys import json from pathlib import Path from unittest.mock import patch, MagicMock import pytest sys.path.insert(0, str(Path(__file__).parent.parent / 'api')) import app as app_module from app import app from auth_manager import AuthManager @pytest.fixture def client(): app.config['TESTING'] = True with app.test_client() as c: yield c # --------------------------------------------------------------------------- # GET /api/connectivity/connections # --------------------------------------------------------------------------- class TestListConnections: def test_list_returns_connections_with_status(self, client): cm = MagicMock() cm.list_connections.return_value = [ {'id': 'conn_a', 'type': 'proxy', 'name': 'P', 'secret_refs': ['conn_a_password'], 'status': {'state': 'configured', 'health': 'working'}, 'config': {'host': 'p', 'port': 3128}}, ] with patch.object(app_module, 'connectivity_manager', cm): resp = client.get('/api/connectivity/connections') assert resp.status_code == 200 body = resp.get_json() assert body['connections'][0]['id'] == 'conn_a' assert body['connections'][0]['status']['health'] == 'working' def test_list_never_returns_secret_values(self, client): cm = MagicMock() # _public_record strips secret values; the manager is what enforces it. cm.list_connections.return_value = [ {'id': 'conn_a', 'type': 'sshuttle', 'name': 'S', 'secret_refs': ['conn_a_private_key'], 'config': {'host': 'h', 'user': 'u', 'auth': 'key'}, 'status': {}}, ] with patch.object(app_module, 'connectivity_manager', cm): resp = client.get('/api/connectivity/connections') raw = resp.get_data(as_text=True) assert 'PRIVATE KEY' not in raw assert 'private_key' not in json.loads(raw)['connections'][0]['config'] def test_list_500_on_exception(self, client): cm = MagicMock() cm.list_connections.side_effect = RuntimeError('boom') with patch.object(app_module, 'connectivity_manager', cm): resp = client.get('/api/connectivity/connections') assert resp.status_code == 500 # --------------------------------------------------------------------------- # POST /api/connectivity/connections # --------------------------------------------------------------------------- class TestCreateConnection: def test_create_201_with_record(self, client): cm = MagicMock() cm.create_connection.return_value = { 'ok': True, 'connection': {'id': 'conn_x', 'type': 'proxy', 'secret_refs': []}} with patch.object(app_module, 'connectivity_manager', cm): resp = client.post('/api/connectivity/connections', json={'type': 'proxy', 'name': 'My Proxy', 'config': {'scheme': 'http', 'host': 'p', 'port': 3128}}) assert resp.status_code == 201 assert resp.get_json()['connection']['id'] == 'conn_x' def test_create_passes_secrets_to_manager(self, client): cm = MagicMock() cm.create_connection.return_value = {'ok': True, 'connection': {'id': 'c'}} with patch.object(app_module, 'connectivity_manager', cm): client.post('/api/connectivity/connections', json={'type': 'sshuttle', 'name': 'S', 'config': {'host': 'h'}, 'secrets': {'private_key': 'SECRET'}}) _, kwargs = cm.create_connection.call_args assert kwargs['secrets'] == {'private_key': 'SECRET'} def test_create_does_not_echo_secret_value(self, client): cm = MagicMock() cm.create_connection.return_value = { 'ok': True, 'connection': {'id': 'c', 'secret_refs': ['c_private_key'], 'config': {}}} with patch.object(app_module, 'connectivity_manager', cm): resp = client.post('/api/connectivity/connections', json={'type': 'sshuttle', 'name': 'S', 'secrets': {'private_key': 'TOPSECRET'}}) assert 'TOPSECRET' not in resp.get_data(as_text=True) def test_create_missing_type_400(self, client): resp = client.post('/api/connectivity/connections', json={'name': 'x'}) assert resp.status_code == 400 def test_create_missing_name_400(self, client): resp = client.post('/api/connectivity/connections', json={'type': 'tor'}) assert resp.status_code == 400 def test_create_validation_error_400(self, client): cm = MagicMock() cm.create_connection.return_value = {'ok': False, 'error': 'invalid host'} with patch.object(app_module, 'connectivity_manager', cm): resp = client.post('/api/connectivity/connections', json={'type': 'proxy', 'name': 'P', 'config': {}}) assert resp.status_code == 400 assert 'invalid host' in resp.get_json()['error'] # --------------------------------------------------------------------------- # PUT /api/connectivity/connections/ # --------------------------------------------------------------------------- class TestUpdateConnection: def test_update_200(self, client): cm = MagicMock() cm.update_connection.return_value = { 'ok': True, 'connection': {'id': 'conn_a', 'name': 'New'}} with patch.object(app_module, 'connectivity_manager', cm): resp = client.put('/api/connectivity/connections/conn_a', json={'name': 'New'}) assert resp.status_code == 200 def test_update_not_found_404(self, client): cm = MagicMock() cm.update_connection.return_value = { 'ok': False, 'error': "connection 'conn_z' not found"} with patch.object(app_module, 'connectivity_manager', cm): resp = client.put('/api/connectivity/connections/conn_z', json={'name': 'x'}) assert resp.status_code == 404 def test_update_validation_400(self, client): cm = MagicMock() cm.update_connection.return_value = {'ok': False, 'error': 'invalid name'} with patch.object(app_module, 'connectivity_manager', cm): resp = client.put('/api/connectivity/connections/conn_a', json={'name': '!!!'}) assert resp.status_code == 400 # --------------------------------------------------------------------------- # DELETE /api/connectivity/connections/ # --------------------------------------------------------------------------- class TestDeleteConnection: def test_delete_200(self, client): cm = MagicMock() cm.delete_connection.return_value = {'ok': True} with patch.object(app_module, 'connectivity_manager', cm): resp = client.delete('/api/connectivity/connections/conn_a') assert resp.status_code == 200 def test_delete_referenced_409(self, client): cm = MagicMock() cm.delete_connection.return_value = { 'ok': False, 'error': "connection is in use by peer 'alice'; detach it first"} with patch.object(app_module, 'connectivity_manager', cm): resp = client.delete('/api/connectivity/connections/conn_a') assert resp.status_code == 409 assert 'in use by' in resp.get_json()['error'] def test_delete_not_found_404(self, client): cm = MagicMock() cm.delete_connection.return_value = { 'ok': False, 'error': "connection 'conn_z' not found"} with patch.object(app_module, 'connectivity_manager', cm): resp = client.delete('/api/connectivity/connections/conn_z') assert resp.status_code == 404 # --------------------------------------------------------------------------- # GET /api/connectivity/connections//health # --------------------------------------------------------------------------- class TestConnectionHealthEndpoint: def test_health_returns_probe_result(self, client): cm = MagicMock() cm.get_connection.return_value = {'id': 'conn_a', 'type': 'proxy'} cm.probe_health.return_value = ('working', 'reachable') with patch.object(app_module, 'connectivity_manager', cm): resp = client.get('/api/connectivity/connections/conn_a/health') assert resp.status_code == 200 body = resp.get_json() assert body['health'] == 'working' assert body['detail'] == 'reachable' def test_health_unknown_connection_404(self, client): cm = MagicMock() cm.get_connection.return_value = None with patch.object(app_module, 'connectivity_manager', cm): resp = client.get('/api/connectivity/connections/conn_z/health') assert resp.status_code == 404 # --------------------------------------------------------------------------- # PUT /api/connectivity/peers//failopen # --------------------------------------------------------------------------- class TestSetPeerFailopen: def test_set_true_200(self, client): cm = MagicMock() cm.set_peer_failopen.return_value = { 'ok': True, 'peer': 'alice', 'exit_failopen': True} with patch.object(app_module, 'connectivity_manager', cm): resp = client.put('/api/connectivity/peers/alice/failopen', json={'failopen': True}) assert resp.status_code == 200 cm.set_peer_failopen.assert_called_once_with('alice', True) def test_clear_with_null_200(self, client): cm = MagicMock() cm.set_peer_failopen.return_value = { 'ok': True, 'peer': 'alice', 'exit_failopen': None} with patch.object(app_module, 'connectivity_manager', cm): resp = client.put('/api/connectivity/peers/alice/failopen', json={'failopen': None}) assert resp.status_code == 200 cm.set_peer_failopen.assert_called_once_with('alice', None) def test_non_bool_400(self, client): resp = client.put('/api/connectivity/peers/alice/failopen', json={'failopen': 'yes'}) assert resp.status_code == 400 def test_unknown_peer_404(self, client): cm = MagicMock() cm.set_peer_failopen.return_value = { 'ok': False, 'error': "peer 'ghost' not found"} with patch.object(app_module, 'connectivity_manager', cm): resp = client.put('/api/connectivity/peers/ghost/failopen', json={'failopen': True}) assert resp.status_code == 404 # --------------------------------------------------------------------------- # admin-only enforcement (mutating connection routes) # --------------------------------------------------------------------------- def _seed_auth(tmp_path): data_dir = str(tmp_path / 'data') config_dir = str(tmp_path / 'config') os.makedirs(data_dir, exist_ok=True) os.makedirs(config_dir, exist_ok=True) mgr = AuthManager(data_dir=data_dir, config_dir=config_dir) mgr.create_user('admin', 'AdminPass123!', 'admin') mgr.create_user('alice', 'AlicePass123!', 'peer') return mgr class TestAdminOnly: def _login(self, c, user, pw): return c.post('/api/auth/login', data=json.dumps({'username': user, 'password': pw}), content_type='application/json') def test_peer_role_forbidden_on_create(self, tmp_path): auth = _seed_auth(tmp_path) app.config['TESTING'] = False app.config['SECRET_KEY'] = 'test-secret' try: import auth_routes with patch.object(app_module, 'auth_manager', auth), \ patch.object(auth_routes, 'auth_manager', auth, create=True), \ patch.object(app_module.setup_manager, 'is_setup_complete', return_value=True): with app.test_client() as c: assert self._login(c, 'alice', 'AlicePass123!').status_code == 200 # CSRF token from session for the mutating request. with c.session_transaction() as sess: token = sess.get('csrf_token') resp = c.post('/api/connectivity/connections', json={'type': 'tor', 'name': 'T'}, headers={'X-CSRF-Token': token or ''}) assert resp.status_code == 403 finally: app.config['TESTING'] = True def test_admin_role_allowed_on_create(self, tmp_path): auth = _seed_auth(tmp_path) cm = MagicMock() cm.create_connection.return_value = {'ok': True, 'connection': {'id': 'c'}} app.config['TESTING'] = False app.config['SECRET_KEY'] = 'test-secret' try: import auth_routes with patch.object(app_module, 'auth_manager', auth), \ patch.object(auth_routes, 'auth_manager', auth, create=True), \ patch.object(app_module, 'connectivity_manager', cm), \ patch.object(app_module.setup_manager, 'is_setup_complete', return_value=True): with app.test_client() as c: assert self._login(c, 'admin', 'AdminPass123!').status_code == 200 with c.session_transaction() as sess: token = sess.get('csrf_token') resp = c.post('/api/connectivity/connections', json={'type': 'tor', 'name': 'T'}, headers={'X-CSRF-Token': token or ''}) assert resp.status_code == 201 finally: app.config['TESTING'] = True if __name__ == '__main__': import pytest as _pytest _pytest.main([__file__, '-q'])