mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 05:24:49 +09:00
Juan Denis
ed2c9c3426
Block stopping/restarting/removing ServerKit's own Docker containers and surface that in the UI. Backend: add DockerService helpers (is_protected_name / is_protected_container), route guard (_reject_if_protected) to reject destructive actions, annotate list_containers with a protected flag, and update cleanup logic. Add tests (backend/tests/test_docker_protection.py). Frontend: hide/replace lifecycle controls for protected containers (lock badge + styles), add Organization tab group (ORG_TABS) and migrate Projects/Shared Variables/Workspaces into a TabGroupLayout with topbar action hooks; update related styles and sidebar item matching.
98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
"""Tests for ServerKit system-container protection on the Docker page.
|
|
|
|
ServerKit's own containers (serverkit-frontend / serverkit-backend) run the
|
|
panel itself, so the Docker API must refuse to stop/restart/remove them and the
|
|
container list must flag them so the UI can hide lifecycle controls.
|
|
|
|
The predicate and the list annotation are pure (subprocess is mocked); the
|
|
route guard tests drive the real blueprint with an admin JWT.
|
|
"""
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from app.services.docker_service import DockerService
|
|
|
|
|
|
class TestIsProtectedName:
|
|
@pytest.mark.parametrize('name', [
|
|
'serverkit-frontend',
|
|
'serverkit_frontend',
|
|
'serverkit-backend',
|
|
'serverkit_backend',
|
|
'serverkit',
|
|
'/serverkit-backend', # docker inspect Name has a leading slash
|
|
'SERVERKIT-FRONTEND', # case-insensitive
|
|
])
|
|
def test_serverkit_names_are_protected(self, name):
|
|
assert DockerService.is_protected_name(name) is True
|
|
|
|
@pytest.mark.parametrize('name', [
|
|
'nginx',
|
|
'my-app_web_1',
|
|
'postgres',
|
|
'',
|
|
None,
|
|
])
|
|
def test_other_names_are_not_protected(self, name):
|
|
assert DockerService.is_protected_name(name) is False
|
|
|
|
|
|
class TestListContainersAnnotation:
|
|
def _fake_ps(self, names):
|
|
lines = '\n'.join(
|
|
'{"ID": "abc%d", "Names": "%s", "Image": "img", "Status": "Up", '
|
|
'"State": "running", "Ports": "", "CreatedAt": "now", "Size": "0B"}'
|
|
% (i, n)
|
|
for i, n in enumerate(names)
|
|
)
|
|
|
|
class _Result:
|
|
returncode = 0
|
|
stdout = lines
|
|
|
|
return _Result()
|
|
|
|
def test_protected_flag_set_per_container(self):
|
|
with patch('subprocess.run', return_value=self._fake_ps(['serverkit-backend', 'nginx'])):
|
|
containers = DockerService.list_containers()
|
|
|
|
by_name = {c['name']: c for c in containers}
|
|
assert by_name['serverkit-backend']['protected'] is True
|
|
assert by_name['nginx']['protected'] is False
|
|
|
|
|
|
class TestIsProtectedContainer:
|
|
def test_resolves_name_from_inspect(self):
|
|
# docker inspect returns the canonical Name with a leading slash.
|
|
with patch.object(DockerService, 'get_container', return_value={'Name': '/serverkit-frontend'}):
|
|
assert DockerService.is_protected_container('abc123') is True
|
|
|
|
def test_unknown_container_not_protected(self):
|
|
with patch.object(DockerService, 'get_container', return_value=None):
|
|
assert DockerService.is_protected_container('abc123') is False
|
|
|
|
|
|
class TestRouteGuards:
|
|
"""The destructive lifecycle routes must 403 on a protected container."""
|
|
|
|
@pytest.mark.parametrize('method,path', [
|
|
('post', '/api/v1/docker/containers/abc123/stop'),
|
|
('post', '/api/v1/docker/containers/abc123/restart'),
|
|
('delete', '/api/v1/docker/containers/abc123'),
|
|
])
|
|
def test_protected_container_is_rejected(self, client, auth_headers, method, path):
|
|
with patch.object(DockerService, 'is_protected_container', return_value=True):
|
|
resp = getattr(client, method)(path, headers=auth_headers, json={})
|
|
assert resp.status_code == 403
|
|
assert 'system container' in resp.get_json()['error'].lower()
|
|
|
|
def test_normal_container_stop_passes_through(self, client, auth_headers):
|
|
with patch.object(DockerService, 'is_protected_container', return_value=False), \
|
|
patch.object(DockerService, 'stop_container', return_value={'success': True}) as stop:
|
|
resp = client.post(
|
|
'/api/v1/docker/containers/abc123/stop', headers=auth_headers, json={}
|
|
)
|
|
assert resp.status_code == 200
|
|
stop.assert_called_once()
|