mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 13:35:12 +09:00
Juan DenisandClaude Fable 5
0a82873c83
SHOW FULL PROCESSLIST / pg_stat_activity normalized per server or container, with admin-only kill/terminate and 5s auto-refresh tabs in the Database Explorer. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
343 lines
14 KiB
Python
343 lines
14 KiB
Python
"""Tests for the live processlist + kill feature (DbProcessService + API)."""
|
|
import pytest
|
|
|
|
from app.services.db_process_service import DbProcessService
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _stub_exec(monkeypatch, output='', success=True, error=None):
|
|
"""Replace the single _exec_sql choke-point, recording calls."""
|
|
calls = []
|
|
|
|
def fake_exec(target, sql):
|
|
calls.append({'target': target, 'sql': sql})
|
|
return {'success': success, 'output': output, 'error': error}
|
|
|
|
monkeypatch.setattr(DbProcessService, '_exec_sql', staticmethod(fake_exec))
|
|
return calls
|
|
|
|
|
|
MYSQL_OUTPUT = (
|
|
"Id\tUser\tHost\tdb\tCommand\tTime\tState\tInfo\n"
|
|
"5\troot\tlocalhost\tNULL\tSleep\t120\t\tNULL\n"
|
|
"12\tapp\t172.17.0.3:5544\tshop\tQuery\t3\texecuting\tSELECT * FROM orders WHERE id > 5\n"
|
|
)
|
|
|
|
PG_OUTPUT = (
|
|
"101|postgres|appdb|active|42|SELECT a || b FROM t WHERE x = 1\n"
|
|
"102|svc|appdb|idle|0|\n"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_processes — parsing / normalization
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestListProcessesMySQL:
|
|
def test_parses_and_normalizes_rows(self, monkeypatch):
|
|
calls = _stub_exec(monkeypatch, output=MYSQL_OUTPUT)
|
|
result = DbProcessService.list_processes({'engine': 'mysql'})
|
|
|
|
assert 'error' not in result
|
|
procs = result['processes']
|
|
assert len(procs) == 2 # header row skipped
|
|
|
|
sleep_row = procs[0]
|
|
assert sleep_row == {
|
|
'id': 5, 'user': 'root', 'db': None, 'command': 'Sleep',
|
|
'time_s': 120, 'state': '', 'query': '',
|
|
}
|
|
|
|
query_row = procs[1]
|
|
assert query_row['id'] == 12
|
|
assert query_row['db'] == 'shop'
|
|
assert query_row['command'] == 'Query'
|
|
assert query_row['time_s'] == 3
|
|
assert query_row['state'] == 'executing'
|
|
assert query_row['query'] == 'SELECT * FROM orders WHERE id > 5'
|
|
|
|
# Sent SHOW FULL PROCESSLIST through the choke-point
|
|
assert calls[0]['sql'] == 'SHOW FULL PROCESSLIST;'
|
|
|
|
def test_empty_output(self, monkeypatch):
|
|
_stub_exec(monkeypatch, output='')
|
|
result = DbProcessService.list_processes({'engine': 'mysql'})
|
|
assert result == {'processes': []}
|
|
|
|
|
|
class TestListProcessesPostgres:
|
|
def test_parses_pipe_separated_rows(self, monkeypatch):
|
|
calls = _stub_exec(monkeypatch, output=PG_OUTPUT)
|
|
result = DbProcessService.list_processes({'engine': 'postgresql'})
|
|
|
|
procs = result['processes']
|
|
assert len(procs) == 2
|
|
|
|
active = procs[0]
|
|
assert active['id'] == 101
|
|
assert active['user'] == 'postgres'
|
|
assert active['db'] == 'appdb'
|
|
assert active['state'] == 'active'
|
|
assert active['time_s'] == 42
|
|
# Query containing '|' survives (bounded split, query column last)
|
|
assert active['query'] == 'SELECT a || b FROM t WHERE x = 1'
|
|
|
|
idle = procs[1]
|
|
assert idle['id'] == 102
|
|
assert idle['time_s'] == 0
|
|
assert idle['query'] == ''
|
|
|
|
assert 'pg_stat_activity' in calls[0]['sql']
|
|
assert 'pg_backend_pid()' in calls[0]['sql']
|
|
|
|
def test_skips_malformed_lines(self, monkeypatch):
|
|
_stub_exec(monkeypatch, output='garbage\nnot|enough|fields\n')
|
|
result = DbProcessService.list_processes({'engine': 'postgresql'})
|
|
assert result == {'processes': []}
|
|
|
|
|
|
class TestListProcessesErrors:
|
|
@pytest.mark.parametrize('engine', ['sqlite', 'mongodb', 'redis', None, ''])
|
|
def test_unsupported_engine(self, engine):
|
|
result = DbProcessService.list_processes({'engine': engine})
|
|
assert result == {'error': 'unsupported engine'}
|
|
|
|
def test_exec_failure_surfaces_error(self, monkeypatch):
|
|
_stub_exec(monkeypatch, success=False, error='docker: command not found')
|
|
result = DbProcessService.list_processes({'engine': 'mysql', 'container': 'db1'})
|
|
assert result == {'error': 'docker: command not found'}
|
|
|
|
def test_exec_failure_without_message(self, monkeypatch):
|
|
_stub_exec(monkeypatch, success=False)
|
|
result = DbProcessService.list_processes({'engine': 'postgresql'})
|
|
assert result == {'error': 'failed to list processes'}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# kill_process — SQL generation + pid validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestKillProcess:
|
|
def test_mysql_kill_sql(self, monkeypatch):
|
|
calls = _stub_exec(monkeypatch)
|
|
result = DbProcessService.kill_process({'engine': 'mysql'}, 42)
|
|
assert result == {'success': True, 'pid': 42}
|
|
assert calls[0]['sql'] == 'KILL 42;'
|
|
|
|
def test_pg_terminate_sql(self, monkeypatch):
|
|
calls = _stub_exec(monkeypatch, output='t\n')
|
|
result = DbProcessService.kill_process({'engine': 'postgresql'}, '77')
|
|
assert result == {'success': True, 'pid': 77}
|
|
assert calls[0]['sql'] == 'SELECT pg_terminate_backend(77);'
|
|
|
|
def test_pg_terminate_false_result(self, monkeypatch):
|
|
_stub_exec(monkeypatch, output='f\n')
|
|
result = DbProcessService.kill_process({'engine': 'postgresql'}, 77)
|
|
assert 'error' in result
|
|
|
|
@pytest.mark.parametrize('bad_pid', ['abc', '42; DROP TABLE x', None, '1 OR 1=1', ''])
|
|
def test_pid_must_be_integer(self, monkeypatch, bad_pid):
|
|
calls = _stub_exec(monkeypatch)
|
|
result = DbProcessService.kill_process({'engine': 'mysql'}, bad_pid)
|
|
assert result == {'error': 'pid must be an integer'}
|
|
assert calls == [] # never reached the exec pathway
|
|
|
|
def test_unsupported_engine(self, monkeypatch):
|
|
calls = _stub_exec(monkeypatch)
|
|
result = DbProcessService.kill_process({'engine': 'sqlite'}, 1)
|
|
assert result == {'error': 'unsupported engine'}
|
|
assert calls == []
|
|
|
|
def test_exec_failure(self, monkeypatch):
|
|
_stub_exec(monkeypatch, success=False, error='access denied')
|
|
result = DbProcessService.kill_process({'engine': 'mysql'}, 5)
|
|
assert result == {'error': 'access denied'}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _exec_sql routing (host vs docker) — stub the DatabaseService pathways
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestExecSqlRouting:
|
|
def test_host_mysql_routes_to_mysql_execute(self, monkeypatch):
|
|
seen = {}
|
|
|
|
def fake(query, database=None, root_password=None):
|
|
seen.update({'query': query, 'root_password': root_password})
|
|
return {'success': True, 'output': '', 'error': None}
|
|
|
|
from app.services import db_process_service
|
|
monkeypatch.setattr(db_process_service.DatabaseService, 'mysql_execute', staticmethod(fake))
|
|
result = DbProcessService._exec_sql({'engine': 'mysql', 'password': 's3c'}, 'SELECT 1;')
|
|
assert result['success'] is True
|
|
assert seen == {'query': 'SELECT 1;', 'root_password': 's3c'}
|
|
|
|
def test_host_pg_routes_to_pg_execute(self, monkeypatch):
|
|
seen = {}
|
|
|
|
def fake(query, database='postgres', user='postgres'):
|
|
seen['query'] = query
|
|
return {'success': True, 'output': '', 'error': None}
|
|
|
|
from app.services import db_process_service
|
|
monkeypatch.setattr(db_process_service.DatabaseService, 'pg_execute', staticmethod(fake))
|
|
result = DbProcessService._exec_sql({'engine': 'postgresql'}, 'SELECT 2;')
|
|
assert result['success'] is True
|
|
assert seen['query'] == 'SELECT 2;'
|
|
|
|
def test_docker_mysql_routes_to_docker_execute(self, monkeypatch):
|
|
seen = {}
|
|
|
|
def fake(container_name, query, database=None, user='root', password=None):
|
|
seen.update({'container': container_name, 'query': query, 'user': user, 'password': password})
|
|
return {'success': True, 'output': '', 'error': None}
|
|
|
|
from app.services import db_process_service
|
|
monkeypatch.setattr(db_process_service.DatabaseService, 'docker_mysql_execute', staticmethod(fake))
|
|
result = DbProcessService._exec_sql(
|
|
{'engine': 'mysql', 'container': 'wp-db', 'user': 'wp', 'password': 'pw'}, 'SHOW FULL PROCESSLIST;')
|
|
assert result['success'] is True
|
|
assert seen == {'container': 'wp-db', 'query': 'SHOW FULL PROCESSLIST;', 'user': 'wp', 'password': 'pw'}
|
|
|
|
def test_docker_pg_clean_error_when_docker_missing(self, monkeypatch):
|
|
def boom(*args, **kwargs):
|
|
raise FileNotFoundError('docker not found')
|
|
|
|
from app.services import db_process_service
|
|
monkeypatch.setattr(db_process_service.subprocess, 'run', boom)
|
|
result = DbProcessService._exec_sql({'engine': 'postgresql', 'container': 'pg1'}, 'SELECT 1;')
|
|
assert result['success'] is False
|
|
assert 'docker not found' in result['error']
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# API endpoints — auth + wiring (service stubbed)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def user_headers(app):
|
|
"""Headers for a non-admin user (kill must be forbidden)."""
|
|
from app import db
|
|
from app.models import User
|
|
from flask_jwt_extended import create_access_token
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
with app.app_context():
|
|
user = User(
|
|
email='plainuser@test.local',
|
|
username='plainuser',
|
|
password_hash=generate_password_hash('testpass'),
|
|
role='user',
|
|
is_active=True,
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
token = create_access_token(identity=user.id)
|
|
return {'Authorization': f'Bearer {token}'}
|
|
|
|
|
|
class TestProcessAPI:
|
|
def test_list_requires_auth(self, client):
|
|
resp = client.get('/api/v1/databases/mysql/processes')
|
|
assert resp.status_code == 401
|
|
|
|
def test_list_host_processes(self, client, auth_headers, monkeypatch):
|
|
seen = {}
|
|
|
|
def fake(target):
|
|
seen.update(target)
|
|
return {'processes': [{'id': 1, 'user': 'root', 'db': None, 'state': '',
|
|
'command': 'Sleep', 'time_s': 9, 'query': ''}]}
|
|
|
|
monkeypatch.setattr(DbProcessService, 'list_processes', staticmethod(fake))
|
|
resp = client.get('/api/v1/databases/postgresql/processes', headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['processes'][0]['id'] == 1
|
|
assert seen['engine'] == 'postgresql'
|
|
assert seen.get('container') is None
|
|
|
|
def test_list_docker_processes_builds_target(self, client, auth_headers, monkeypatch):
|
|
seen = {}
|
|
|
|
def fake(target):
|
|
seen.update(target)
|
|
return {'processes': []}
|
|
|
|
monkeypatch.setattr(DbProcessService, 'list_processes', staticmethod(fake))
|
|
resp = client.get(
|
|
'/api/v1/databases/docker/wp-db/processes?type=mysql&user=wp',
|
|
headers={**auth_headers, 'X-DB-Password': 'pw'})
|
|
assert resp.status_code == 200
|
|
assert seen['engine'] == 'mysql'
|
|
assert seen['container'] == 'wp-db'
|
|
assert seen['user'] == 'wp'
|
|
assert seen['password'] == 'pw'
|
|
|
|
def test_list_unsupported_engine_is_400(self, client, auth_headers, monkeypatch):
|
|
monkeypatch.setattr(
|
|
DbProcessService, 'list_processes',
|
|
staticmethod(lambda target: {'error': 'unsupported engine'}))
|
|
resp = client.get('/api/v1/databases/docker/c1/processes?type=redis', headers=auth_headers)
|
|
assert resp.status_code == 400
|
|
assert resp.get_json() == {'error': 'unsupported engine'}
|
|
|
|
def test_list_exec_failure_is_502(self, client, auth_headers, monkeypatch):
|
|
monkeypatch.setattr(
|
|
DbProcessService, 'list_processes',
|
|
staticmethod(lambda target: {'error': 'connection refused'}))
|
|
resp = client.get('/api/v1/databases/mysql/processes', headers=auth_headers)
|
|
assert resp.status_code == 502
|
|
assert resp.get_json() == {'error': 'connection refused'}
|
|
|
|
def test_kill_requires_auth(self, client):
|
|
resp = client.post('/api/v1/databases/mysql/processes/5/kill')
|
|
assert resp.status_code == 401
|
|
|
|
def test_kill_requires_admin(self, client, user_headers, monkeypatch):
|
|
called = []
|
|
monkeypatch.setattr(
|
|
DbProcessService, 'kill_process',
|
|
staticmethod(lambda target, pid: called.append(pid) or {'success': True, 'pid': pid}))
|
|
resp = client.post('/api/v1/databases/mysql/processes/5/kill', headers=user_headers)
|
|
assert resp.status_code == 403
|
|
assert called == []
|
|
|
|
def test_kill_host_process_as_admin(self, client, auth_headers, monkeypatch):
|
|
seen = {}
|
|
|
|
def fake(target, pid):
|
|
seen.update({'target': target, 'pid': pid})
|
|
return {'success': True, 'pid': pid}
|
|
|
|
monkeypatch.setattr(DbProcessService, 'kill_process', staticmethod(fake))
|
|
resp = client.post('/api/v1/databases/mysql/processes/42/kill', headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {'success': True, 'pid': 42}
|
|
assert seen['target']['engine'] == 'mysql'
|
|
assert seen['pid'] == 42
|
|
|
|
def test_kill_docker_process_as_admin(self, client, auth_headers, monkeypatch):
|
|
seen = {}
|
|
|
|
def fake(target, pid):
|
|
seen.update({'target': target, 'pid': pid})
|
|
return {'success': True, 'pid': pid}
|
|
|
|
monkeypatch.setattr(DbProcessService, 'kill_process', staticmethod(fake))
|
|
resp = client.post(
|
|
'/api/v1/databases/docker/pg-db/processes/101/kill',
|
|
headers=auth_headers, json={'type': 'postgresql', 'user': 'svc'})
|
|
assert resp.status_code == 200
|
|
assert seen['target']['engine'] == 'postgresql'
|
|
assert seen['target']['container'] == 'pg-db'
|
|
assert seen['target']['user'] == 'svc'
|
|
assert seen['pid'] == 101
|
|
|
|
def test_kill_pid_must_be_int_in_url(self, client, auth_headers):
|
|
# <int:pid> converter rejects non-numeric pids at the routing layer
|
|
resp = client.post('/api/v1/databases/mysql/processes/abc/kill', headers=auth_headers)
|
|
assert resp.status_code in (404, 405)
|