| from __future__ import annotations |
|
|
| from threading import Event |
| import time |
|
|
| from starlette.middleware.gzip import GZipMiddleware |
|
|
| from api.app import _start_image_cleanup, create_app |
| from services.config import config |
|
|
|
|
| def test_image_cleanup_runs_without_blocking_startup(monkeypatch) -> None: |
| cleanup_started = Event() |
| allow_cleanup_to_finish = Event() |
|
|
| def slow_cleanup() -> int: |
| cleanup_started.set() |
| allow_cleanup_to_finish.wait(timeout=2) |
| return 0 |
|
|
| monkeypatch.setattr(config, "cleanup_old_images", slow_cleanup) |
|
|
| started_at = time.monotonic() |
| thread = _start_image_cleanup() |
| elapsed = time.monotonic() - started_at |
|
|
| assert elapsed < 0.25 |
| assert cleanup_started.wait(timeout=1) |
| assert thread.is_alive() |
|
|
| allow_cleanup_to_finish.set() |
| thread.join(timeout=1) |
| assert not thread.is_alive() |
|
|
|
|
| def test_app_compresses_large_json_responses() -> None: |
| app = create_app() |
| gzip_middleware = next( |
| item for item in app.user_middleware if item.cls is GZipMiddleware |
| ) |
|
|
| assert gzip_middleware.kwargs["minimum_size"] == 1024 |
| assert gzip_middleware.kwargs["compresslevel"] == 1 |
|
|
