mirror of
https://github.com/azaion/detections.git
synced 2026-04-22 22:36:32 +00:00
Oleksandr Bezdieniezhnykh
834f846dc8
- Added a new Cython extension for the engine factory to the setup configuration. - Updated the inference module to include additional logging for video batch processing and annotation callbacks. - Refactored test cases to standardize the detection endpoint responses and include channel IDs in headers for better event handling.
89 lines
2.5 KiB
Python
89 lines
2.5 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
|
|
|
|
_TEST_JWT_SECRET = "az-test-secret-for-unit-tests-only-32b"
|
|
|
|
|
|
def _access_jwt(sub: str = "u1") -> str:
|
|
return pyjwt.encode(
|
|
{"exp": int(time.time()) + 3600, "sub": sub},
|
|
_TEST_JWT_SECRET,
|
|
algorithm="HS256",
|
|
)
|
|
|
|
|
|
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(main, "JWT_SECRET", _TEST_JWT_SECRET),
|
|
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}", "X-Channel-Id": "test-channel"},
|
|
)
|
|
# Assert
|
|
assert r.status_code == 202
|
|
assert wb_hits.count(expected_path) == 1
|
|
with real_open(expected_path, "rb") as f:
|
|
assert f.read() == img
|