Hitonabi
46a8f50c34
Ampel / ampel (push) Successful in 28s
Vorfall 24.07.: Beim API-Start blockierte der synchrone CIFS-Schreibtest in alle_remounten()/mounten() im Kernel (wait_for_response), als der SMB-Server langsam war -> ~5 min "Waiting for application startup", kein Endpoint bedient (bis der soft-Mount per Timeout abbrach). startup_event() lief isoliert sauber, also war es der blockierende Netz-Mount, nicht die App-Logik. Fix: remount als Hintergrund-Task (asyncio.create_task) statt await -> die API kommt sofort hoch, die Mounts stellen sich her sobald der Server antwortet. Test (test_api_smoke.py): haelt den Nicht-blockierend-Vertrag per Quelltext- Inspektion fest, im Stil der anderen Verdrahtungs-Tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
"""Import-Smoke-Test: bricht, wenn main.py kaputte Imports oder Verdrahtung hat.
|
|
|
|
Warum: Kein anderer Test importiert main.py — ein Tippfehler dort fiele sonst
|
|
erst beim Container-Start auf (und die Ampel bliebe fälschlich grün).
|
|
Läuft nur unter Linux (detection.py nutzt fcntl/ioctl), also genau dort,
|
|
wo auch die Ampel läuft.
|
|
"""
|
|
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
if sys.platform == "win32": # pragma: no cover
|
|
pytest.skip("detection.py braucht fcntl (Linux)", allow_module_level=True)
|
|
|
|
|
|
def test_main_importierbar_und_routen_verdrahtet():
|
|
from main import app
|
|
|
|
routen = {route.path for route in app.routes}
|
|
for pfad in ("/health", "/jobs", "/devices", "/logs", "/settings", "/prescan"):
|
|
assert pfad in routen, f"Route {pfad} fehlt"
|
|
|
|
|
|
def test_dateiname_validierung_blockt_pfad_tricks():
|
|
"""Download-Endpoint: nur nackte Dateinamen — kein .., kein Slash, kein Dotfile."""
|
|
from main import _sicherer_dateiname
|
|
|
|
assert _sicherer_dateiname("film.mkv") is True
|
|
assert _sicherer_dateiname("../../etc/passwd") is False
|
|
assert _sicherer_dateiname("a/b.mkv") is False
|
|
assert _sicherer_dateiname("a\\b.mkv") is False
|
|
assert _sicherer_dateiname(".versteckt") is False
|
|
assert _sicherer_dateiname("") is False
|
|
|
|
|
|
def test_worker_task_name_passt_zum_celery_client():
|
|
"""API schickt an 'worker.tasks.rip_disc' — der Name ist Vertrag mit dem Worker."""
|
|
import inspect
|
|
|
|
import celery_client
|
|
|
|
quelle = inspect.getsource(celery_client.start_rip)
|
|
assert '"worker.tasks.rip_disc"' in quelle
|
|
|
|
|
|
def test_remount_blockiert_den_api_start_nicht():
|
|
"""Regression (Vorfall 24.07.): ein haengender Netz-Mount (CIFS-Schreibtest kann
|
|
im Kernel haengen, wait_for_response) darf den API-Start NICHT blockieren. remount
|
|
muss als Hintergrund-Task laufen (create_task), nicht direkt awaited werden."""
|
|
import inspect
|
|
|
|
import main
|
|
|
|
quelle = inspect.getsource(main.startup_event)
|
|
assert "create_task(asyncio.to_thread(remount))" in quelle, \
|
|
"remount muss als Hintergrund-Task laufen (nicht blockierend)"
|
|
assert "await asyncio.to_thread(remount)" not in quelle, \
|
|
"remount darf nicht mehr direkt awaited werden (blockiert sonst den Start)"
|