mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-03 02:26:43 +09:00
Juan DenisandClaude Fable 5
9b9aa34f01
tests/factories.py owns make_user/headers_for/make_application/ make_server/make_workspace; conftest personas (scoping_rbac, new viewer_headers/developer_headers fixtures) ride them. The byte-identical _seed_app pair (container scale/sleep), two variants (image_update, waf) and three server fixtures (fleet doctor/repair) migrate as proof. Ratchet freezes the 44-file hand-minted-JWT and 65-file direct-Application() populations. Note: test_fleet_doctor.py::test_dns_fail_when_unresolved fails pre-existing on this machine (warn vs fail), unrelated to this change. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
261 lines
11 KiB
Python
261 lines
11 KiB
Python
"""REST surface for the fleet doctor (plan 72 A.2).
|
|
|
|
The sweep engine itself is covered by test_fleet_doctor.py; this file covers the
|
|
routes that make it reachable:
|
|
|
|
* auth + admin gating on all three routes,
|
|
* ``POST /doctor/fleet/run`` is a 202 + job id (never a synchronous fan-out —
|
|
the agent gateway registry is single-worker and in-memory),
|
|
* the report shape ``{'report': {'ran_at', 'servers': [...]}}`` including the
|
|
``key`` alias the panel renders findings by,
|
|
* an offline/unknown server degrades to a reported state rather than an error.
|
|
"""
|
|
import pytest
|
|
from flask_jwt_extended import create_access_token
|
|
from werkzeug.security import generate_password_hash
|
|
|
|
from app.services import agent_registry as agent_registry_mod
|
|
from app.services.fleet_doctor_service import FLEET_DOCTOR_JOB_KIND, FleetDoctorService
|
|
|
|
registry = agent_registry_mod.agent_registry
|
|
|
|
|
|
@pytest.fixture
|
|
def server(app):
|
|
from app import db
|
|
from factories import make_server
|
|
return make_server(db, hostname='box', ip_address='203.0.113.10')
|
|
|
|
|
|
@pytest.fixture
|
|
def user_headers(app):
|
|
"""A non-admin (viewer) JWT — the negative case for @admin_required."""
|
|
from app import db
|
|
from app.models import User
|
|
user = User(
|
|
email='fleetpleb@test.local',
|
|
username='fleetpleb',
|
|
password_hash=generate_password_hash('x'),
|
|
role=User.ROLE_VIEWER,
|
|
is_active=True,
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
return {'Authorization': f'Bearer {create_access_token(identity=user.id)}'}
|
|
|
|
|
|
def _seed_rows(server_id):
|
|
from app.services.fleet_doctor_service import _row
|
|
FleetDoctorService._persist(server_id, [
|
|
_row('service.nginx', 'nginx service', 'ok', 'Running.'),
|
|
_row('service.docker', 'docker service', 'fail', 'Not running.',
|
|
repairable=True,
|
|
repair_ref={'kind': 'fleet.service', 'server_id': server_id,
|
|
'name': 'docker'}),
|
|
])
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Gating
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class TestGating:
|
|
@pytest.mark.parametrize('method,path', [
|
|
('get', '/api/v1/doctor/fleet'),
|
|
('post', '/api/v1/doctor/fleet/run'),
|
|
('post', '/api/v1/doctor/fleet/repair'),
|
|
])
|
|
def test_anonymous_is_rejected(self, client, method, path):
|
|
resp = getattr(client, method)(path)
|
|
assert resp.status_code in (401, 422)
|
|
|
|
@pytest.mark.parametrize('method,path', [
|
|
('get', '/api/v1/doctor/fleet'),
|
|
('post', '/api/v1/doctor/fleet/run'),
|
|
('post', '/api/v1/doctor/fleet/repair'),
|
|
])
|
|
def test_non_admin_is_forbidden(self, client, user_headers, method, path):
|
|
resp = getattr(client, method)(path, headers=user_headers)
|
|
assert resp.status_code == 403
|
|
assert 'error' in resp.get_json()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# GET /doctor/fleet
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class TestReport:
|
|
def test_report_is_empty_but_well_shaped_with_no_servers(self, client, auth_headers):
|
|
resp = client.get('/api/v1/doctor/fleet', headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
report = resp.get_json()['report']
|
|
assert report['servers'] == []
|
|
assert report['ran_at'] is None
|
|
|
|
def test_report_lists_a_server_with_no_rows_yet(self, client, auth_headers,
|
|
server, monkeypatch):
|
|
monkeypatch.setattr(registry, 'is_agent_connected', lambda sid: True)
|
|
report = client.get('/api/v1/doctor/fleet',
|
|
headers=auth_headers).get_json()['report']
|
|
assert len(report['servers']) == 1
|
|
entry = report['servers'][0]
|
|
assert entry['server_id'] == server.id
|
|
assert entry['name'] == 'box1'
|
|
assert entry['connected'] is True
|
|
assert entry['checks'] == []
|
|
assert entry['ran_at'] is None
|
|
|
|
def test_report_returns_rows_counts_and_the_key_alias(self, client, auth_headers,
|
|
server, monkeypatch):
|
|
monkeypatch.setattr(registry, 'is_agent_connected', lambda sid: True)
|
|
_seed_rows(server.id)
|
|
|
|
report = client.get('/api/v1/doctor/fleet',
|
|
headers=auth_headers).get_json()['report']
|
|
entry = report['servers'][0]
|
|
assert entry['counts'] == {'ok': 1, 'warn': 0, 'fail': 1, 'error': 0}
|
|
assert entry['ran_at'] and report['ran_at']
|
|
|
|
checks = {c['key']: c for c in entry['checks']}
|
|
# The panel renders findings field-driven off key/repairable/repair_ref.
|
|
assert set(checks) == {'service.nginx', 'service.docker'}
|
|
assert checks['service.docker']['check_key'] == 'service.docker'
|
|
assert checks['service.docker']['repairable'] is True
|
|
assert checks['service.docker']['repair_ref'] == {
|
|
'kind': 'fleet.service', 'server_id': server.id, 'name': 'docker'}
|
|
assert checks['service.nginx']['repairable'] is False
|
|
|
|
def test_offline_server_is_reported_not_dropped(self, client, auth_headers,
|
|
server, monkeypatch):
|
|
"""An unreachable box keeps its last rows and reports connected=false."""
|
|
_seed_rows(server.id)
|
|
monkeypatch.setattr(registry, 'is_agent_connected', lambda sid: False)
|
|
|
|
report = client.get('/api/v1/doctor/fleet',
|
|
headers=auth_headers).get_json()['report']
|
|
entry = report['servers'][0]
|
|
assert entry['connected'] is False
|
|
assert len(entry['checks']) == 2
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# POST /doctor/fleet/run
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class TestRun:
|
|
def test_run_returns_202_with_a_job_id(self, client, auth_headers, monkeypatch):
|
|
from app.jobs.models import Job
|
|
|
|
# Never fan out on the request thread — if the route called the sweep
|
|
# directly this would blow up.
|
|
def _boom():
|
|
raise AssertionError('run_fleet_doctor must not run on a request thread')
|
|
|
|
monkeypatch.setattr(FleetDoctorService, 'run_fleet_doctor',
|
|
classmethod(lambda cls: _boom()))
|
|
|
|
resp = client.post('/api/v1/doctor/fleet/run', headers=auth_headers)
|
|
assert resp.status_code == 202
|
|
job_id = resp.get_json()['job_id']
|
|
assert job_id
|
|
|
|
job = Job.query.get(job_id)
|
|
assert job is not None
|
|
assert job.kind == FLEET_DOCTOR_JOB_KIND
|
|
assert job.status == Job.STATUS_PENDING
|
|
|
|
def test_the_job_kind_has_a_registered_handler(self, app):
|
|
from app.jobs import registry as job_registry
|
|
FleetDoctorService.register_jobs()
|
|
assert job_registry.is_registered(FLEET_DOCTOR_JOB_KIND)
|
|
|
|
def test_the_fleet_sweep_schedule_is_seeded_and_staggered(self, app):
|
|
from app.jobs.builtin_handlers import seed_builtin_schedules
|
|
from app.jobs.models import ScheduledJob
|
|
from app.services.doctor_service import DOCTOR_SCHEDULE_NAME
|
|
from app.services.fleet_doctor_service import FLEET_DOCTOR_SCHEDULE_NAME
|
|
|
|
seed_builtin_schedules()
|
|
fleet = ScheduledJob.query.filter_by(name=FLEET_DOCTOR_SCHEDULE_NAME).first()
|
|
host = ScheduledJob.query.filter_by(name=DOCTOR_SCHEDULE_NAME).first()
|
|
assert fleet is not None and host is not None
|
|
assert fleet.kind == FLEET_DOCTOR_JOB_KIND
|
|
assert fleet.interval_seconds == 86400
|
|
# Staggered: the fleet sweep must not fire alongside the host sweep.
|
|
assert fleet.next_run_at > host.next_run_at
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# POST /doctor/fleet/repair
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class TestRepair:
|
|
def test_empty_items_is_a_400(self, client, auth_headers):
|
|
resp = client.post('/api/v1/doctor/fleet/repair', headers=auth_headers,
|
|
json={'items': []})
|
|
assert resp.status_code == 400
|
|
assert 'error' in resp.get_json()
|
|
|
|
def test_non_allowlisted_kind_is_refused_as_a_result(self, client, auth_headers):
|
|
resp = client.post('/api/v1/doctor/fleet/repair', headers=auth_headers,
|
|
json={'items': [{'kind': 'fleet.rm-rf',
|
|
'server_id': 'x', 'name': 'nginx'}]})
|
|
assert resp.status_code == 200
|
|
result = resp.get_json()['results'][0]
|
|
assert result['success'] is False
|
|
assert result['code'] == 'NOT_ALLOWLISTED'
|
|
|
|
def test_offline_agent_is_refused_not_errored(self, client, auth_headers,
|
|
server, monkeypatch):
|
|
monkeypatch.setattr(registry, 'is_agent_connected', lambda sid: False)
|
|
resp = client.post('/api/v1/doctor/fleet/repair', headers=auth_headers,
|
|
json={'items': [{'kind': 'fleet.service',
|
|
'server_id': server.id,
|
|
'name': 'nginx'}]})
|
|
assert resp.status_code == 200
|
|
result = resp.get_json()['results'][0]
|
|
assert result['success'] is False
|
|
assert result['code'] == 'AGENT_OFFLINE'
|
|
|
|
def test_allowlisted_repair_dispatches_and_reports_success(
|
|
self, client, auth_headers, server, monkeypatch):
|
|
monkeypatch.setattr(registry, 'is_agent_connected', lambda sid: True)
|
|
monkeypatch.setattr(registry, 'has_capability', lambda sid, feat: True)
|
|
sent = {}
|
|
|
|
def _send(server_id, action, params, timeout=None, user_id=None):
|
|
sent.update(server_id=server_id, action=action, params=params)
|
|
return {'success': True}
|
|
|
|
monkeypatch.setattr(registry, 'send_command', _send)
|
|
|
|
resp = client.post('/api/v1/doctor/fleet/repair', headers=auth_headers,
|
|
json={'items': [{'kind': 'fleet.service',
|
|
'server_id': server.id,
|
|
'name': 'nginx'}]})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['results'][0]['success'] is True
|
|
assert sent['action'] == 'systemd:restart'
|
|
assert sent['params'] == {'unit': 'nginx'}
|
|
|
|
def test_one_bad_item_does_not_sink_the_batch(self, client, auth_headers,
|
|
server, monkeypatch):
|
|
monkeypatch.setattr(registry, 'is_agent_connected', lambda sid: True)
|
|
monkeypatch.setattr(registry, 'has_capability', lambda sid, feat: True)
|
|
monkeypatch.setattr(
|
|
registry, 'send_command',
|
|
lambda server_id, action, params, timeout=None, user_id=None:
|
|
{'success': True})
|
|
|
|
resp = client.post('/api/v1/doctor/fleet/repair', headers=auth_headers,
|
|
json={'items': [
|
|
{'kind': 'fleet.service', 'server_id': server.id,
|
|
'name': 'not-a-unit'},
|
|
{'kind': 'fleet.service', 'server_id': server.id,
|
|
'name': 'nginx'},
|
|
]})
|
|
assert resp.status_code == 200
|
|
results = resp.get_json()['results']
|
|
assert [r['success'] for r in results] == [False, True]
|
|
assert results[0]['code'] == 'NOT_ALLOWLISTED'
|