gx
afc2dd7fff
#199 DLPack export: - frame.dlpack_y() / .dlpack_uv() — explicit multi-plane access для NV12 - frame.__dlpack__() / __dlpack_device__() — protocol для torch/cupy - Capsule deleter правильно держит refcount на frame_keep_alive, releases shape/strides arrays. CUDA pointer принадлежит frame. #200 Health/stats counters: - frames_received, timeouts, errors — per-call counters - last_seq, gap_count — proxy для drop count (NEWEST_ONLY mode) - last_frame_pts_ns - stats() — snapshot dict для MQTT health publish - counted в pybind layer т.к. C API не expose'ит ring_occupancy #201 Per-subscriber CUDA stream + thread-safety: - consumer_stream kwarg в subscribe() — int (cudaStream_t pointer) - subscriber.consumer_stream property - Thread-safety contract в docstring CuframesSubscriber - next_frame() передаёт consumer_stream_ в cuframes_subscriber_next #202 Smoke test + docs: - 10/10 pytest passed (расширен +2 теста на consumer_stream) - docs/python.md (~250 строк): quick start, API reference, integration с PyTorch/CuPy, reconnect-loop pattern, per-stream usage, pitch alignment, thread-safety, error taxonomy, backpressure, Phase 0 limitations Verify build + tests: cmake -B build-python -DBUILD_PYTHON_BINDINGS=ON cmake --build build-python -j pytest python/tests/ -v # 10/10 Закрывает Phase 0 issue gx/cuframes#6. Разблокирует goldix-smart-home/yolo-world-detector Phase 1. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
"""Smoke tests для cuframes Python bindings.
|
||
|
||
В Phase 0 (skeleton) проверяем что:
|
||
- модуль импортируется
|
||
- версия читается
|
||
- error классы существуют и являются нормальной иерархией
|
||
|
||
Subscriber / DLPack тесты появятся в следующих фазах
|
||
(см. issue gx/cuframes#6, tasks #198+).
|
||
"""
|
||
|
||
import cuframes
|
||
|
||
|
||
def test_version_format():
|
||
v = cuframes.version_string()
|
||
assert isinstance(v, str)
|
||
parts = v.split(".")
|
||
assert len(parts) >= 3
|
||
assert all(p.isdigit() for p in parts[:3])
|
||
|
||
|
||
def test_protocol_version_is_uint():
|
||
pv = cuframes.protocol_version()
|
||
assert isinstance(pv, int)
|
||
assert pv >= 0
|
||
|
||
|
||
def test_pixel_format_enum_members():
|
||
assert cuframes.PixelFormat.NV12.value == 0
|
||
assert cuframes.PixelFormat.YUV420P.value == 1
|
||
|
||
|
||
def test_subscriber_mode_enum_members():
|
||
assert cuframes.SubscriberMode.NEWEST_ONLY.value == 0
|
||
assert cuframes.SubscriberMode.STRICT_ORDER.value == 1
|
||
|
||
|
||
def test_error_hierarchy():
|
||
"""Все subtype'ы наследуются от CuframesError."""
|
||
for sub in [
|
||
cuframes.CuframesPublisherGone,
|
||
cuframes.CuframesFrameTimeout,
|
||
cuframes.CuframesDeviceLost,
|
||
cuframes.CuframesShmError,
|
||
cuframes.CuframesProtocolMismatch,
|
||
cuframes.CuframesInvalidArgument,
|
||
cuframes.CuframesOutOfMemory,
|
||
cuframes.CuframesInternal,
|
||
]:
|
||
assert issubclass(sub, cuframes.CuframesError)
|
||
|
||
|
||
def test_subscriber_class_exposed():
|
||
"""CuframesSubscriber/CuframesFrame exposed как public classes."""
|
||
assert hasattr(cuframes, "CuframesSubscriber")
|
||
assert hasattr(cuframes, "CuframesFrame")
|
||
assert hasattr(cuframes, "subscribe")
|
||
|
||
|
||
def test_subscribe_to_missing_publisher_raises():
|
||
"""Subscribe к несуществующему publisher → CuframesError (subclass)
|
||
после connect_timeout_ms.
|
||
|
||
Этот тест работает на любом хосте (без живого cuframes-pub) — мы
|
||
верифицируем что error path работает и маппит CUFRAMES_ERR_*
|
||
в правильный Python exception.
|
||
"""
|
||
import pytest
|
||
with pytest.raises(cuframes.CuframesError):
|
||
cuframes.subscribe(
|
||
"definitely-not-existing-publisher-xyz",
|
||
connect_timeout_ms=100,
|
||
)
|
||
|
||
|
||
def test_subscriber_repr_when_unable_to_connect():
|
||
"""Лёгкий тест что repr не падает и close idempotent."""
|
||
import pytest
|
||
try:
|
||
sub = cuframes.subscribe("nope-xyz", connect_timeout_ms=100)
|
||
except cuframes.CuframesError:
|
||
return # ожидаемо
|
||
pytest.fail("subscribe должно было выкинуть exception")
|
||
|
||
|
||
def test_subscribe_accepts_consumer_stream_param():
|
||
"""consumer_stream — uintptr (cudaStream_t).
|
||
|
||
Проверяем что параметр accepted; реальное использование требует
|
||
cuda-python / torch.cuda.Stream — это в integration тестах
|
||
yolo-world-detector'а.
|
||
"""
|
||
import pytest
|
||
with pytest.raises(cuframes.CuframesError):
|
||
cuframes.subscribe(
|
||
"nope-xyz",
|
||
connect_timeout_ms=100,
|
||
consumer_stream=0, # 0 = default stream
|
||
)
|
||
|
||
|
||
def test_subscribe_kwargs_signature():
|
||
"""Проверяем что у subscribe правильный набор kwargs."""
|
||
import inspect
|
||
# Pybind11-обёртки не дают inspect.signature, но help_doc отражает их.
|
||
doc = cuframes.subscribe.__doc__
|
||
assert "consumer_name" in doc
|
||
assert "mode" in doc
|
||
assert "cuda_device" in doc
|
||
assert "connect_timeout_ms" in doc
|
||
assert "consumer_stream" in doc
|