roof
aa1e5c41ec
Unit Tests / test (push) Successful in 12m6s
Coverage was below acceptable levels and several newly-added code paths (sshuttle egress, proxy egress, DDNS provider stubs, DNS overview route, peer-registry provisioning) had zero test coverage. ~250 new unit tests are added across 16 new test files. Existing test files are updated to match refactored interfaces (DHCP removed, constants introduced, network_manager restructured). .coveragerc is added to pin the source mapping and the 70% floor so regressions are caught at commit time. tests/test_enhanced_api.py was previously living in api/ (wrong location) and is moved to tests/ where it belongs. Integration test files are updated to remove references to DHCP endpoints and add coverage for the new DNS overview and DDNS sync endpoints. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
378 lines
16 KiB
Python
378 lines
16 KiB
Python
"""
|
|
Tests for routes/containers.py — container, image, and volume management endpoints.
|
|
All endpoints require is_local_request() to return True; non-local gets 403.
|
|
"""
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'api'))
|
|
|
|
import app as app_module
|
|
from app import app
|
|
|
|
|
|
@pytest.fixture
|
|
def local_client():
|
|
"""Flask test client that appears as local (loopback) request."""
|
|
app.config['TESTING'] = True
|
|
with app.test_client() as c:
|
|
# is_local_request is imported inside each route handler from app
|
|
with patch.object(app_module, 'is_local_request', return_value=True):
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
def remote_client():
|
|
"""Flask test client that appears as a non-local (remote) request."""
|
|
app.config['TESTING'] = True
|
|
with app.test_client() as c:
|
|
with patch.object(app_module, 'is_local_request', return_value=False):
|
|
yield c
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/containers — requires local
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListContainers:
|
|
def test_returns_200_from_local(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.list_containers.return_value = [{'name': 'cell-dns', 'status': 'running'}]
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.get('/api/containers')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_containers_list(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.list_containers.return_value = [{'name': 'cell-dns'}]
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.get('/api/containers')
|
|
data = json.loads(resp.data)
|
|
assert isinstance(data, list)
|
|
assert data[0]['name'] == 'cell-dns'
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.get('/api/containers')
|
|
assert resp.status_code == 403
|
|
|
|
def test_500_on_exception(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.list_containers.side_effect = Exception('docker error')
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.get('/api/containers')
|
|
assert resp.status_code == 500
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/containers/<name>/start
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestStartContainer:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.start_container.return_value = True
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/containers/cell-dns/start')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.post('/api/containers/cell-dns/start')
|
|
assert resp.status_code == 403
|
|
|
|
def test_response_shape(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.start_container.return_value = True
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/containers/cell-dns/start')
|
|
data = json.loads(resp.data)
|
|
assert 'started' in data
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/containers/<name>/stop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestStopContainer:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.stop_container.return_value = True
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/containers/cell-dns/stop')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.post('/api/containers/cell-dns/stop')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/containers/<name>/restart
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRestartContainer:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.restart_container.return_value = True
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/containers/cell-dns/restart')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.post('/api/containers/cell-dns/restart')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/containers/<name>/logs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetContainerLogs:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_container_logs.return_value = ['line1', 'line2']
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.get('/api/containers/cell-dns/logs')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_logs_in_response(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_container_logs.return_value = ['line1']
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.get('/api/containers/cell-dns/logs?tail=50')
|
|
data = json.loads(resp.data)
|
|
assert 'logs' in data
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.get('/api/containers/cell-dns/logs')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/containers/<name>/stats
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetContainerStats:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.get_container_stats.return_value = {'cpu': '5%', 'memory': '100MB'}
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.get('/api/containers/cell-dns/stats')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.get('/api/containers/cell-dns/stats')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/containers (create)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCreateContainer:
|
|
def test_returns_400_when_image_missing(self, local_client):
|
|
resp = local_client.post('/api/containers',
|
|
data=json.dumps({'name': 'test'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 400
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.post('/api/containers',
|
|
data=json.dumps({'image': 'nginx'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 403
|
|
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.create_container.return_value = {'id': 'abc123', 'name': 'mycontainer'}
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/containers',
|
|
data=json.dumps({'image': 'nginx', 'name': 'mycontainer'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_500_when_result_has_error(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.create_container.return_value = {'error': 'image not found'}
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/containers',
|
|
data=json.dumps({'image': 'badimage'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 500
|
|
|
|
def test_volume_outside_allowed_path_returns_403(self, local_client):
|
|
"""Volume mounts outside the allowed directories are blocked."""
|
|
mock_cm = MagicMock()
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/containers',
|
|
data=json.dumps({
|
|
'image': 'nginx',
|
|
'volumes': {'/etc/passwd': '/mnt/passwd'}
|
|
}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 403
|
|
|
|
def test_volume_in_allowed_path_passes(self, local_client):
|
|
"""Volume mounts under /tmp/ are permitted."""
|
|
mock_cm = MagicMock()
|
|
mock_cm.create_container.return_value = {'id': 'abc'}
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/containers',
|
|
data=json.dumps({
|
|
'image': 'nginx',
|
|
'volumes': {'/tmp/test': '/mnt/test'}
|
|
}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DELETE /api/containers/<name>
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRemoveContainer:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.remove_container.return_value = True
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.delete('/api/containers/mycontainer')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.delete('/api/containers/mycontainer')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/images
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListImages:
|
|
def test_returns_200(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.list_images.return_value = [{'tag': 'nginx:latest'}]
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.get('/api/images')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.get('/api/images')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/images/pull
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPullImage:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.pull_image.return_value = {'status': 'pulled'}
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/images/pull',
|
|
data=json.dumps({'image': 'nginx:latest'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_400_when_image_missing(self, local_client):
|
|
resp = local_client.post('/api/images/pull',
|
|
data=json.dumps({}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 400
|
|
|
|
def test_returns_500_when_pull_fails(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.pull_image.return_value = {'error': 'not found'}
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/images/pull',
|
|
data=json.dumps({'image': 'badimage'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 500
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.post('/api/images/pull',
|
|
data=json.dumps({'image': 'nginx'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DELETE /api/images/<image>
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRemoveImage:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.remove_image.return_value = True
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.delete('/api/images/nginx:latest')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.delete('/api/images/nginx:latest')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/volumes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListVolumes:
|
|
def test_returns_200(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.list_volumes.return_value = []
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.get('/api/volumes')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.get('/api/volumes')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/volumes (create)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCreateVolume:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.create_volume.return_value = {'name': 'myvolume'}
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.post('/api/volumes',
|
|
data=json.dumps({'name': 'myvolume'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_400_when_name_missing(self, local_client):
|
|
resp = local_client.post('/api/volumes',
|
|
data=json.dumps({}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 400
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.post('/api/volumes',
|
|
data=json.dumps({'name': 'v'}),
|
|
content_type='application/json')
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DELETE /api/volumes/<name>
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRemoveVolume:
|
|
def test_returns_200_on_success(self, local_client):
|
|
mock_cm = MagicMock()
|
|
mock_cm.remove_volume.return_value = True
|
|
with patch.object(app_module, 'container_manager', mock_cm):
|
|
resp = local_client.delete('/api/volumes/myvolume')
|
|
assert resp.status_code == 200
|
|
|
|
def test_returns_403_from_remote(self, remote_client):
|
|
resp = remote_client.delete('/api/volumes/myvolume')
|
|
assert resp.status_code == 403
|