Update .gitignore and refine documentation for execution environment

- Added Cython generated files to .gitignore to prevent unnecessary tracking.
- Updated paths in `inference.c` and `coreml_engine.c` to reflect the correct virtual environment.
- Revised the execution environment documentation to clarify hardware dependency checks and local execution instructions, ensuring accurate guidance for users.
- Removed outdated Docker suitability checks and streamlined the assessment process for test execution environments.
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-03-30 00:53:46 +03:00
parent 27f4aceb52
commit 5a968edcba
12 changed files with 328 additions and 401 deletions
-137
View File
@@ -1,27 +1,7 @@
import json
import os
import threading
import time
import uuid
import pytest
import requests
_DETECT_TIMEOUT = 60
_MEDIA = os.environ.get("MEDIA_DIR", "/media")
def _ai_config_video() -> dict:
return {
"probability_threshold": 0.25,
"tracking_intersection_threshold": 0.6,
"altitude": 400,
"focal_length": 24,
"sensor_width": 23.5,
"paths": [f"{_MEDIA}/video_test01.mp4"],
"frame_period_recognition": 4,
"frame_recognition_seconds": 2,
}
def test_ft_n_06_loader_unreachable_during_init_health(
@@ -44,62 +24,6 @@ def test_ft_n_06_loader_unreachable_during_init_health(
assert d.get("errorMessage") is None
@pytest.mark.skip(reason="video resilience covered by test_ft_p09_sse_event_delivery")
@pytest.mark.slow
@pytest.mark.timeout(300)
def test_ft_n_07_annotations_unreachable_detection_continues(
warm_engine,
http_client,
jwt_token,
mock_annotations_url,
sse_client_factory,
):
requests.post(
f"{mock_annotations_url}/mock/config", json={"mode": "error"}, timeout=10
).raise_for_status()
media_id = f"res-n07-{uuid.uuid4().hex}"
body = _ai_config_video()
headers = {"Authorization": f"Bearer {jwt_token}"}
collected = []
thread_exc = []
done = threading.Event()
def _listen():
try:
with sse_client_factory() as sse:
time.sleep(0.3)
for event in sse.events():
if not event.data or not str(event.data).strip():
continue
data = json.loads(event.data)
if data.get("mediaId") != media_id:
continue
collected.append(data)
if (
data.get("mediaStatus") == "AIProcessed"
and data.get("mediaPercent") == 100
):
break
except BaseException as e:
thread_exc.append(e)
finally:
done.set()
th = threading.Thread(target=_listen, daemon=True)
th.start()
time.sleep(0.5)
pr = http_client.post(f"/detect/{media_id}", json=body, headers=headers)
assert pr.status_code == 200
ok = done.wait(timeout=290)
assert ok
th.join(timeout=5)
assert not thread_exc
assert any(
e.get("mediaStatus") == "AIProcessed" and e.get("mediaPercent") == 100
for e in collected
)
def test_nft_res_01_loader_outage_after_init(
warm_engine, http_client, mock_loader_url, image_small
):
@@ -117,62 +41,6 @@ def test_nft_res_01_loader_outage_after_init(
assert hd.get("errorMessage") is None
@pytest.mark.skip(reason="Single video run — covered by test_ft_p09_sse_event_delivery")
@pytest.mark.slow
@pytest.mark.timeout(300)
def test_nft_res_02_annotations_outage_during_async_detection(
warm_engine,
http_client,
jwt_token,
mock_annotations_url,
sse_client_factory,
):
media_id = f"res-n02-{uuid.uuid4().hex}"
body = _ai_config_video()
headers = {"Authorization": f"Bearer {jwt_token}"}
collected = []
thread_exc = []
done = threading.Event()
def _listen():
try:
with sse_client_factory() as sse:
time.sleep(0.3)
for event in sse.events():
if not event.data or not str(event.data).strip():
continue
data = json.loads(event.data)
if data.get("mediaId") != media_id:
continue
collected.append(data)
if (
data.get("mediaStatus") == "AIProcessed"
and data.get("mediaPercent") == 100
):
break
except BaseException as e:
thread_exc.append(e)
finally:
done.set()
th = threading.Thread(target=_listen, daemon=True)
th.start()
time.sleep(0.5)
pr = http_client.post(f"/detect/{media_id}", json=body, headers=headers)
assert pr.status_code == 200
requests.post(
f"{mock_annotations_url}/mock/config", json={"mode": "error"}, timeout=10
).raise_for_status()
ok = done.wait(timeout=290)
assert ok
th.join(timeout=5)
assert not thread_exc
assert any(
e.get("mediaStatus") == "AIProcessed" and e.get("mediaPercent") == 100
for e in collected
)
def test_nft_res_03_transient_loader_first_fail(
mock_loader_url, http_client, image_small
):
@@ -188,8 +56,3 @@ def test_nft_res_03_transient_loader_first_fail(
assert r1.status_code != 500
@pytest.mark.skip(
reason="Requires docker compose restart capability not available in e2e-runner"
)
def test_nft_res_04_service_restart():
pass