roof
a338836bb8
Security fixes: - Replace debug=True with env-driven FLASK_DEBUG in app.py - Add _safe_path helper and path-traversal protection to all 6 file routes in file_manager.py - Add peer_name regex and input validation (public_key, name, endpoint_ip) in wireguard_manager.py - Stop returning private key from GET /api/wireguard/keys; return only public_key + has_private_key boolean - Fix is_local_request() XFF bypass by checking remote_addr only, ignoring X-Forwarded-For - Remove duplicate get_all_configs / get_config_summary methods from config_manager.py DevOps: - Bind 6 internal service ports to 127.0.0.1 in docker-compose.yml (radicale, webdav, api, webui, rainloop, filegator) - Move WebDAV credentials to env vars (WEBDAV_USER, WEBDAV_PASS) - Pin flask, flask-cors, requests, cryptography, docker to secure minimum versions in requirements.txt QA (560 tests, 0 failures): - tests/test_wireguard_endpoints.py: 18 new endpoint tests - tests/test_file_endpoints.py: 24 new endpoint tests incl. path traversal - tests/test_container_manager.py: expanded from 2 to 30 tests - tests/test_config_backup_restore_http.py: 25 new tests (new file) - tests/test_config_apply.py: 9 new tests (new file) Docs: - Rewrite README.md with accurate architecture, ports, env vars, security notes - Rewrite QUICKSTART.md with verified commands Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
324 lines
13 KiB
Python
324 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Unit tests for ContainerManager (api/container_manager.py).
|
|
"""
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock, PropertyMock
|
|
|
|
api_dir = Path(__file__).parent.parent / 'api'
|
|
sys.path.insert(0, str(api_dir))
|
|
|
|
from container_manager import ContainerManager
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper to build a ContainerManager with a pre-wired mock Docker client
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_manager(mock_from_env):
|
|
"""Return a ContainerManager whose Docker client is mock_from_env's return."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
return ContainerManager(), mock_client
|
|
|
|
|
|
class TestListContainers(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_list_containers(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_container = MagicMock()
|
|
mock_container.id = 'abc'
|
|
mock_container.name = 'test'
|
|
mock_container.status = 'running'
|
|
mock_container.image.tags = ['img']
|
|
mock_container.labels = {}
|
|
mock_client.containers.list.return_value = [mock_container]
|
|
result = mgr.list_containers()
|
|
self.assertEqual(result[0]['name'], 'test')
|
|
|
|
|
|
class TestStartStopRestart(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_start_stop_restart_container(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_container = MagicMock()
|
|
mock_client.containers.get.return_value = mock_container
|
|
# Start
|
|
self.assertTrue(mgr.start_container('test'))
|
|
mock_container.start.assert_called_once()
|
|
# Stop
|
|
self.assertTrue(mgr.stop_container('test'))
|
|
mock_container.stop.assert_called_once()
|
|
# Restart
|
|
self.assertTrue(mgr.restart_container('test'))
|
|
mock_container.restart.assert_called_once()
|
|
# Exception cases
|
|
mock_client.containers.get.side_effect = Exception('fail')
|
|
self.assertFalse(mgr.start_container('bad'))
|
|
self.assertFalse(mgr.stop_container('bad'))
|
|
self.assertFalse(mgr.restart_container('bad'))
|
|
|
|
|
|
class TestGetContainerLogs(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_get_container_logs_returns_string(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_container = MagicMock()
|
|
mock_container.logs.return_value = b'log line 1\nlog line 2\n'
|
|
mock_client.containers.get.return_value = mock_container
|
|
result = mgr.get_container_logs('mycontainer')
|
|
self.assertIsInstance(result, str)
|
|
self.assertIn('log line 1', result)
|
|
|
|
@patch('docker.from_env')
|
|
def test_get_container_logs_uses_tail_parameter(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_container = MagicMock()
|
|
mock_container.logs.return_value = b''
|
|
mock_client.containers.get.return_value = mock_container
|
|
mgr.get_container_logs('mycontainer', tail=50)
|
|
mock_container.logs.assert_called_once_with(tail=50)
|
|
|
|
@patch('docker.from_env')
|
|
def test_get_container_logs_raises_when_docker_unavailable(self, mock_from_env):
|
|
mock_from_env.side_effect = Exception('docker not found')
|
|
with self.assertRaises(Exception):
|
|
mgr = ContainerManager()
|
|
mgr.get_container_logs('test')
|
|
|
|
|
|
class TestGetContainerStats(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_get_container_stats_returns_dict(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_container = MagicMock()
|
|
mock_container.stats.return_value = {
|
|
'cpu_stats': {'cpu_usage': {'total_usage': 123}},
|
|
'memory_stats': {'usage': 4096},
|
|
}
|
|
mock_client.containers.get.return_value = mock_container
|
|
result = mgr.get_container_stats('mycontainer')
|
|
self.assertIsInstance(result, dict)
|
|
self.assertIn('cpu_stats', result)
|
|
self.assertIn('memory_stats', result)
|
|
|
|
@patch('docker.from_env')
|
|
def test_get_container_stats_returns_error_dict_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.containers.get.side_effect = Exception('not found')
|
|
result = mgr.get_container_stats('nonexistent')
|
|
self.assertIsInstance(result, dict)
|
|
self.assertIn('error', result)
|
|
|
|
|
|
class TestCreateContainer(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_create_container_returns_id_and_name(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_container = MagicMock()
|
|
mock_container.id = 'cid123'
|
|
mock_container.name = 'myapp'
|
|
mock_client.containers.create.return_value = mock_container
|
|
result = mgr.create_container(
|
|
image='nginx:latest',
|
|
name='myapp',
|
|
env={'ENV_VAR': 'value'},
|
|
volumes={'/host/path': '/container/path'},
|
|
command='nginx -g "daemon off;"',
|
|
ports={'80/tcp': 8080},
|
|
)
|
|
self.assertEqual(result['id'], 'cid123')
|
|
self.assertEqual(result['name'], 'myapp')
|
|
|
|
@patch('docker.from_env')
|
|
def test_create_container_returns_error_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.containers.create.side_effect = Exception('image not found')
|
|
result = mgr.create_container(image='nonexistent:latest', name='test')
|
|
self.assertIn('error', result)
|
|
|
|
@patch('docker.from_env')
|
|
def test_create_container_passes_env_to_docker(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_container = MagicMock()
|
|
mock_container.id = 'x'
|
|
mock_container.name = 'y'
|
|
mock_client.containers.create.return_value = mock_container
|
|
mgr.create_container(image='alpine', name='test', env={'KEY': 'VAL'})
|
|
_, kwargs = mock_client.containers.create.call_args
|
|
self.assertEqual(kwargs['environment'], {'KEY': 'VAL'})
|
|
|
|
|
|
class TestRemoveContainer(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_remove_container_returns_true_on_success(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_container = MagicMock()
|
|
mock_client.containers.get.return_value = mock_container
|
|
result = mgr.remove_container('mycontainer')
|
|
self.assertTrue(result)
|
|
mock_container.remove.assert_called_once()
|
|
|
|
@patch('docker.from_env')
|
|
def test_remove_container_returns_false_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.containers.get.side_effect = Exception('not found')
|
|
result = mgr.remove_container('ghost')
|
|
self.assertFalse(result)
|
|
|
|
|
|
class TestPullImage(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_pull_image_returns_id_and_tags(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_image = MagicMock()
|
|
mock_image.id = 'sha256:abc'
|
|
mock_image.tags = ['nginx:latest']
|
|
mock_client.images.pull.return_value = mock_image
|
|
result = mgr.pull_image('nginx:latest')
|
|
self.assertEqual(result['id'], 'sha256:abc')
|
|
self.assertEqual(result['tags'], ['nginx:latest'])
|
|
|
|
@patch('docker.from_env')
|
|
def test_pull_image_returns_error_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.images.pull.side_effect = Exception('pull access denied')
|
|
result = mgr.pull_image('private/image:latest')
|
|
self.assertIn('error', result)
|
|
|
|
|
|
class TestRemoveImage(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_remove_image_returns_true_on_success(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
result = mgr.remove_image('nginx:latest')
|
|
self.assertTrue(result)
|
|
mock_client.images.remove.assert_called_once_with(image='nginx:latest', force=False)
|
|
|
|
@patch('docker.from_env')
|
|
def test_remove_image_returns_false_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.images.remove.side_effect = Exception('image in use')
|
|
result = mgr.remove_image('nginx:latest')
|
|
self.assertFalse(result)
|
|
|
|
|
|
class TestCreateVolume(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_create_volume_returns_name_and_mountpoint(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_vol = MagicMock()
|
|
mock_vol.name = 'myvolume'
|
|
mock_vol.attrs = {'Mountpoint': '/var/lib/docker/volumes/myvolume/_data'}
|
|
mock_client.volumes.create.return_value = mock_vol
|
|
result = mgr.create_volume('myvolume')
|
|
self.assertEqual(result['name'], 'myvolume')
|
|
self.assertIn('mountpoint', result)
|
|
self.assertIn('myvolume', result['mountpoint'])
|
|
|
|
@patch('docker.from_env')
|
|
def test_create_volume_returns_error_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.volumes.create.side_effect = Exception('no space left')
|
|
result = mgr.create_volume('bigvolume')
|
|
self.assertIn('error', result)
|
|
|
|
|
|
class TestRemoveVolume(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_remove_volume_returns_true_on_success(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_vol = MagicMock()
|
|
mock_client.volumes.get.return_value = mock_vol
|
|
result = mgr.remove_volume('myvolume')
|
|
self.assertTrue(result)
|
|
mock_vol.remove.assert_called_once()
|
|
|
|
@patch('docker.from_env')
|
|
def test_remove_volume_returns_false_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.volumes.get.side_effect = Exception('volume not found')
|
|
result = mgr.remove_volume('ghostvolume')
|
|
self.assertFalse(result)
|
|
|
|
|
|
class TestListImages(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_list_images_returns_list(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_img = MagicMock()
|
|
mock_img.id = 'sha256:abc'
|
|
mock_img.tags = ['nginx:latest']
|
|
mock_img.short_id = 'abc123'
|
|
mock_client.images.list.return_value = [mock_img]
|
|
result = mgr.list_images()
|
|
self.assertIsInstance(result, list)
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0]['id'], 'sha256:abc')
|
|
|
|
@patch('docker.from_env')
|
|
def test_list_images_returns_empty_list_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.images.list.side_effect = Exception('daemon unreachable')
|
|
result = mgr.list_images()
|
|
self.assertEqual(result, [])
|
|
|
|
|
|
class TestListVolumes(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_list_volumes_returns_list(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_vol = MagicMock()
|
|
mock_vol.name = 'vol1'
|
|
mock_vol.attrs = {'Mountpoint': '/mnt/vol1'}
|
|
mock_client.volumes.list.return_value = [mock_vol]
|
|
result = mgr.list_volumes()
|
|
self.assertIsInstance(result, list)
|
|
self.assertEqual(result[0]['name'], 'vol1')
|
|
self.assertEqual(result[0]['mountpoint'], '/mnt/vol1')
|
|
|
|
@patch('docker.from_env')
|
|
def test_list_volumes_returns_empty_list_on_exception(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.volumes.list.side_effect = Exception('daemon unreachable')
|
|
result = mgr.list_volumes()
|
|
self.assertEqual(result, [])
|
|
|
|
|
|
class TestGetStatusWhenDockerUnavailable(unittest.TestCase):
|
|
@patch('docker.from_env')
|
|
def test_get_status_offline_when_docker_init_fails(self, mock_from_env):
|
|
"""ContainerManager.get_status() returns {running: False, status: 'offline'}
|
|
when Docker client could not be initialised."""
|
|
mock_from_env.side_effect = Exception('Cannot connect to Docker daemon')
|
|
mgr = ContainerManager()
|
|
self.assertFalse(mgr.docker_available)
|
|
status = mgr.get_status()
|
|
self.assertFalse(status['running'])
|
|
self.assertEqual(status['status'], 'offline')
|
|
|
|
@patch('docker.from_env')
|
|
def test_get_status_online_when_docker_available(self, mock_from_env):
|
|
mgr, mock_client = _make_manager(mock_from_env)
|
|
mock_client.containers.list.return_value = []
|
|
mock_client.images.list.return_value = []
|
|
mock_client.volumes.list.return_value = []
|
|
mock_client.info.return_value = {
|
|
'ServerVersion': '24.0.0',
|
|
'Containers': 0,
|
|
'Images': 0,
|
|
'Driver': 'overlay2',
|
|
'KernelVersion': '6.1.0',
|
|
'OperatingSystem': 'Linux',
|
|
}
|
|
status = mgr.get_status()
|
|
self.assertTrue(status['running'])
|
|
self.assertEqual(status['status'], 'online')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|