Spaces:
Running
Running
File size: 2,173 Bytes
f2bf02d 50ffab2 605dd3b 50ffab2 605dd3b 50ffab2 f2bf02d 50ffab2 605dd3b 50ffab2 605dd3b 50ffab2 f2bf02d 50ffab2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | from services.pipeline import ProcessingPipeline
import pytest
import numpy as np
from unittest.mock import MagicMock, AsyncMock, patch
@pytest.fixture
def mock_deps():
return {
"detector": MagicMock(),
"safety": MagicMock(),
"depth": MagicMock(),
"redis": AsyncMock(),
}
@pytest.mark.asyncio
async def test_pipeline_success(mock_deps):
mock_detection = MagicMock()
mock_detection.xyxy = [10, 10, 20, 20]
mock_detection.class_name = "person"
# Settings mock models values
mock_deps["detector"].detect.return_value.detections = [mock_detection]
mock_deps["safety"].detect.return_value = [] # No dangers
mock_config = MagicMock()
mock_config.experiment = False
pipeline = ProcessingPipeline(
detector=mock_deps["detector"],
safety_detector=mock_deps["safety"],
depth_model=mock_deps["depth"],
redis=mock_deps["redis"],
config=mock_config,
)
with (
patch("cv2.imdecode") as mock_decode,
patch("utils.profiling.mlflow") as _,
):
mock_decode.return_value = np.zeros((100, 100, 3), dtype=np.uint8)
result = await pipeline.run(
camera_id="test", frame_bytes=b"test", frame_count=1
)
assert result["status"] == 200
mock_deps["depth"].calculate_depth.assert_called_once()
mock_deps["redis"].publish.assert_called_once()
@pytest.mark.asyncio
async def test_pipeline_no_detections(mock_deps):
mock_deps["detector"].detect.return_value.detections = []
mock_config = MagicMock()
mock_config.experiment = False
pipeline = ProcessingPipeline(
detector=mock_deps["detector"],
safety_detector=mock_deps["safety"],
depth_model=mock_deps["depth"],
redis=mock_deps["redis"],
config=mock_config,
)
with (
patch("cv2.imdecode") as mock_decode,
patch("utils.profiling.mlflow") as _,
):
mock_decode.return_value = np.zeros((100, 100, 3), dtype=np.uint8)
await pipeline.run(camera_id="test", frame_bytes=b"test", frame_count=1)
mock_deps["depth"].calculate_depth.assert_not_called()
|