mirror of
https://github.com/azaion/detections.git
synced 2026-04-22 05:26:32 +00:00
Oleksandr Bezdieniezhnykh
097811a67b
- Pin all deps; h11==0.16.0 (CVE-2025-43859), python-multipart>=1.3.1 (CVE-2026-28356), PyJWT==2.12.1
- Add HMAC JWT verification (require_auth FastAPI dependency, JWT_SECRET-gated)
- Fix TokenManager._refresh() to use ADMIN_API_URL instead of ANNOTATIONS_URL
- Rename POST /detect → POST /detect/image (image-only, rejects video files)
- Replace global SSE stream with per-job SSE: GET /detect/{media_id} with event replay buffer
- Apply require_auth to all 4 protected endpoints
- Fix on_annotation/on_status closure to use mutable current_id for correct post-upload event routing
- Add non-root appuser to Dockerfile and Dockerfile.gpu
- Add JWT_SECRET to e2e/docker-compose.test.yml and run-tests.sh
- Update all e2e tests and unit tests for new endpoints and HMAC token signing
- 64/64 tests pass
Made-with: Cursor
92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
import builtins
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import cv2
|
|
import jwt as pyjwt
|
|
import numpy as np
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
pytest.importorskip("inference")
|
|
|
|
import inference as inference_mod
|
|
|
|
|
|
def _access_jwt(sub: str = "u1") -> str:
|
|
secret = os.environ.get("JWT_SECRET", "")
|
|
if secret:
|
|
return pyjwt.encode(
|
|
{"exp": int(time.time()) + 3600, "sub": sub},
|
|
secret,
|
|
algorithm="HS256",
|
|
)
|
|
import base64, json
|
|
raw = json.dumps({"exp": int(time.time()) + 3600, "sub": sub}, separators=(",", ":")).encode()
|
|
payload = base64.urlsafe_b64encode(raw).decode().rstrip("=")
|
|
return f"h.{payload}.s"
|
|
|
|
|
|
def _jpeg_bytes() -> bytes:
|
|
_, buf = cv2.imencode(".jpg", np.zeros((2, 2, 3), dtype=np.uint8))
|
|
return buf.tobytes()
|
|
|
|
|
|
class _FakeInfVideo:
|
|
def run_detect_image(self, image_bytes, ai_cfg, media_name, on_annotation, *args, **kwargs):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def reset_main_inference():
|
|
import main
|
|
|
|
main.inference = None
|
|
yield
|
|
main.inference = None
|
|
|
|
|
|
def test_auth_image_still_writes_once_before_detect(reset_main_inference):
|
|
# Arrange
|
|
import main
|
|
from media_hash import compute_media_content_hash
|
|
|
|
real_open = builtins.open
|
|
wb_hits = []
|
|
|
|
def tracking_open(file, mode="r", *args, **kwargs):
|
|
if mode == "wb":
|
|
wb_hits.append(os.path.abspath(str(file)))
|
|
return real_open(file, mode, *args, **kwargs)
|
|
|
|
img = _jpeg_bytes()
|
|
token = _access_jwt()
|
|
mock_post = MagicMock()
|
|
mock_post.return_value.status_code = 201
|
|
mock_put = MagicMock()
|
|
mock_put.return_value.status_code = 204
|
|
with tempfile.TemporaryDirectory() as idir:
|
|
os.environ["IMAGES_DIR"] = idir
|
|
content_hash = compute_media_content_hash(img)
|
|
expected_path = os.path.abspath(os.path.join(idir, f"{content_hash}.jpg"))
|
|
client = TestClient(main.app)
|
|
with (
|
|
patch.object(builtins, "open", tracking_open),
|
|
patch.object(main.http_requests, "post", mock_post),
|
|
patch.object(main.http_requests, "put", mock_put),
|
|
patch.object(main, "get_inference", return_value=_FakeInfVideo()),
|
|
):
|
|
# Act
|
|
r = client.post(
|
|
"/detect/image",
|
|
files={"file": ("p.jpg", img, "image/jpeg")},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
# Assert
|
|
assert r.status_code == 200
|
|
assert wb_hits.count(expected_path) == 1
|
|
with real_open(expected_path, "rb") as f:
|
|
assert f.read() == img
|