mirror of
https://github.com/azaion/detections.git
synced 2026-04-22 22:26:33 +00:00
Oleksandr Bezdieniezhnykh
129 lines
3.5 KiB
Python
129 lines
3.5 KiB
Python
import base64
|
|
import json
|
|
import time
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
|
|
def _access_jwt(sub: str = "u1") -> str:
|
|
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 test_token_manager_decode_user_id_sub():
|
|
# Arrange
|
|
from main import TokenManager
|
|
|
|
raw = json.dumps(
|
|
{"sub": "user-abc", "exp": int(time.time()) + 3600}, separators=(",", ":")
|
|
).encode()
|
|
payload = base64.urlsafe_b64encode(raw).decode().rstrip("=")
|
|
token = f"hdr.{payload}.sig"
|
|
# Act
|
|
uid = TokenManager.decode_user_id(token)
|
|
# Assert
|
|
assert uid == "user-abc"
|
|
|
|
|
|
def test_token_manager_decode_user_id_invalid():
|
|
# Arrange
|
|
from main import TokenManager
|
|
|
|
# Act
|
|
uid = TokenManager.decode_user_id("not-a-jwt")
|
|
# Assert
|
|
assert uid is None
|
|
|
|
|
|
def test_merged_annotation_settings_pascal_case():
|
|
# Arrange
|
|
from main import _merged_annotation_settings_payload
|
|
|
|
raw = {
|
|
"FramePeriodRecognition": 5,
|
|
"ProbabilityThreshold": 0.4,
|
|
"Altitude": 300,
|
|
"FocalLength": 35,
|
|
"SensorWidth": 36,
|
|
}
|
|
# Act
|
|
out = _merged_annotation_settings_payload(raw)
|
|
# Assert
|
|
assert out["frame_period_recognition"] == 5
|
|
assert out["probability_threshold"] == 0.4
|
|
assert out["altitude"] == 300
|
|
|
|
|
|
def test_merged_annotation_nested_sections():
|
|
# Arrange
|
|
from main import _merged_annotation_settings_payload
|
|
|
|
raw = {
|
|
"aiRecognitionSettings": {"modelBatchSize": 4},
|
|
"cameraSettings": {"altitude": 100},
|
|
}
|
|
# Act
|
|
out = _merged_annotation_settings_payload(raw)
|
|
# Assert
|
|
assert out["model_batch_size"] == 4
|
|
assert out["altitude"] == 100
|
|
|
|
|
|
def test_resolve_media_for_detect_uses_api_path_and_defaults_when_api_empty():
|
|
# Arrange
|
|
import main
|
|
|
|
tm = main.TokenManager(_access_jwt(), "")
|
|
mock_ann = MagicMock()
|
|
mock_ann.fetch_user_ai_settings.return_value = None
|
|
mock_ann.fetch_media_path.return_value = "/m/file.jpg"
|
|
with patch("main.annotations_client", mock_ann):
|
|
# Act
|
|
cfg, path = main._resolve_media_for_detect("mid-1", tm, None)
|
|
# Assert
|
|
assert path == "/m/file.jpg"
|
|
assert "paths" not in cfg
|
|
assert "probability_threshold" not in cfg
|
|
|
|
|
|
def test_resolve_media_for_detect_override_wins():
|
|
# Arrange
|
|
import main
|
|
|
|
tm = main.TokenManager(_access_jwt(), "")
|
|
override = main.AIConfigDto(probability_threshold=0.99)
|
|
mock_ann = MagicMock()
|
|
mock_ann.fetch_user_ai_settings.return_value = {
|
|
"probabilityThreshold": 0.2,
|
|
"altitude": 500,
|
|
}
|
|
mock_ann.fetch_media_path.return_value = "/m/v.mp4"
|
|
with patch("main.annotations_client", mock_ann):
|
|
# Act
|
|
cfg, path = main._resolve_media_for_detect("vid-1", tm, override)
|
|
# Assert
|
|
assert cfg["probability_threshold"] == 0.99
|
|
assert cfg["altitude"] == 500
|
|
assert path == "/m/v.mp4"
|
|
assert "paths" not in cfg
|
|
|
|
|
|
def test_resolve_media_for_detect_raises_when_no_media_path():
|
|
# Arrange
|
|
import main
|
|
|
|
tm = main.TokenManager(_access_jwt(), "")
|
|
mock_ann = MagicMock()
|
|
mock_ann.fetch_user_ai_settings.return_value = {}
|
|
mock_ann.fetch_media_path.return_value = None
|
|
with patch("main.annotations_client", mock_ann):
|
|
# Act / Assert
|
|
with pytest.raises(HTTPException) as exc:
|
|
main._resolve_media_for_detect("missing", tm, None)
|
|
assert exc.value.status_code == 503
|