mirror of
https://github.com/azaion/detections.git
synced 2026-04-22 21:56:33 +00:00
Oleksandr Bezdieniezhnykh
8baa96978b
- Updated the detection image endpoint to require a channel ID for event streaming. - Introduced a new endpoint for streaming detection events, allowing clients to receive real-time updates. - Enhanced the internal buffering mechanism for detection events to manage multiple channels. - Refactored the inference module to support the new event handling structure. Made-with: Cursor
137 lines
3.9 KiB
Python
137 lines
3.9 KiB
Python
"""
|
|
AZ-178: True streaming video detection — e2e tests.
|
|
|
|
Both tests upload video_test01.mp4 (12 MB), wait for the first SSE event,
|
|
then stop. The goal is to prove the service starts and produces detections,
|
|
not to process the whole file.
|
|
|
|
Run with: pytest e2e/tests/test_streaming_video_upload.py -s -v
|
|
"""
|
|
import json
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import sseclient
|
|
|
|
FIXTURES_DIR = Path(__file__).resolve().parent.parent / "fixtures"
|
|
_TIMEOUT = 5.0
|
|
_STOP_AFTER = 5
|
|
|
|
|
|
def _fixture_path(name: str) -> str:
|
|
p = FIXTURES_DIR / name
|
|
if not p.is_file():
|
|
pytest.skip(f"missing fixture {p}")
|
|
return str(p)
|
|
|
|
|
|
def _chunked_reader(path: str, chunk_size: int = 64 * 1024):
|
|
with open(path, "rb") as f:
|
|
while True:
|
|
chunk = f.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
yield chunk
|
|
|
|
|
|
def _start_sse_listener(
|
|
http_client, channel_id: str, auth_headers: dict
|
|
) -> tuple[list[dict], list[BaseException], threading.Event]:
|
|
events: list[dict] = []
|
|
errors: list[BaseException] = []
|
|
first_event = threading.Event()
|
|
connected = threading.Event()
|
|
|
|
def _listen():
|
|
try:
|
|
with http_client.get(
|
|
f"/detect/events/{channel_id}",
|
|
stream=True,
|
|
timeout=_TIMEOUT + 2,
|
|
headers=auth_headers,
|
|
) as resp:
|
|
resp.raise_for_status()
|
|
connected.set()
|
|
for event in sseclient.SSEClient(resp).events():
|
|
if not event.data or not str(event.data).strip():
|
|
continue
|
|
events.append(json.loads(event.data))
|
|
if len(events) >= _STOP_AFTER:
|
|
first_event.set()
|
|
break
|
|
except BaseException as exc:
|
|
errors.append(exc)
|
|
finally:
|
|
connected.set()
|
|
first_event.set()
|
|
|
|
th = threading.Thread(target=_listen, daemon=True)
|
|
th.start()
|
|
connected.wait(timeout=3)
|
|
return events, errors, first_event
|
|
|
|
|
|
@pytest.mark.timeout(10)
|
|
def test_streaming_video_detections_appear_during_upload(
|
|
warm_engine, http_client, auth_headers
|
|
):
|
|
# Arrange
|
|
video_path = _fixture_path("video_test01.mp4")
|
|
channel_id = str(uuid.uuid4())
|
|
events, errors, first_event = _start_sse_listener(http_client, channel_id, auth_headers)
|
|
|
|
# Act
|
|
r = http_client.post(
|
|
"/detect/video",
|
|
data=_chunked_reader(video_path),
|
|
headers={
|
|
**auth_headers,
|
|
"X-Channel-Id": channel_id,
|
|
"X-Filename": "video_test01.mp4",
|
|
"Content-Type": "application/octet-stream",
|
|
},
|
|
timeout=8,
|
|
)
|
|
assert r.status_code == 202
|
|
first_event.wait(timeout=_TIMEOUT)
|
|
|
|
# Assert
|
|
assert not errors, f"SSE thread error: {errors}"
|
|
assert len(events) >= 1, "Expected at least one SSE event within 5s"
|
|
print(f"\n First {len(events)} SSE events:")
|
|
for e in events:
|
|
print(f" {e}")
|
|
|
|
|
|
@pytest.mark.timeout(10)
|
|
def test_non_faststart_video_still_works(warm_engine, http_client, auth_headers):
|
|
# Arrange
|
|
video_path = _fixture_path("video_test01.mp4")
|
|
channel_id = str(uuid.uuid4())
|
|
events, errors, first_event = _start_sse_listener(http_client, channel_id, auth_headers)
|
|
|
|
# Act
|
|
r = http_client.post(
|
|
"/detect/video",
|
|
data=_chunked_reader(video_path),
|
|
headers={
|
|
**auth_headers,
|
|
"X-Channel-Id": channel_id,
|
|
"X-Filename": "video_test01_plain.mp4",
|
|
"Content-Type": "application/octet-stream",
|
|
},
|
|
timeout=8,
|
|
)
|
|
assert r.status_code == 202
|
|
first_event.wait(timeout=_TIMEOUT)
|
|
|
|
# Assert
|
|
assert not errors, f"SSE thread error: {errors}"
|
|
assert len(events) >= 1, "Expected at least one SSE event within 5s"
|
|
print(f"\n First {len(events)} SSE events:")
|
|
for e in events:
|
|
print(f" {e}")
|