petr.polezhaev
b84bd8d555
- C1: единственный владелец stop+destroy — reader-поток; close() только выставляет флаг и ждёт; повторный close и close из колбэка безопасны (регресс-тесты; TSan на close с мёртвым модулем — 0 гонок) - M1: wheel собирается с bundled mbedtls (нет внешней soname mbedcrypto); попутно исправлена bundled-ветка CMake (-lSTATIC) - C-API: счётчики потерянных событий/логов, лог-кольцо без union-балласта, атомарные fn/ctx лог-синка - provision: валидация формата облачных ответов; CLI: человеческие ошибки (exit 2), --out с правами 0600, console_script pyfglair - упаковка: MANIFEST.in (sdist с C++-исходниками), очистка stale .so, readme в метаданных; find_library без ложных .bak - тесты: close-сценарии, доставка логов, ошибки CLI/provision, C-API очередь
56 lines
1.8 KiB
Python
56 lines
1.8 KiB
Python
"""Низкоуровневые проверки C-API: очередь событий, счётчики, ошибки входа."""
|
|
from __future__ import annotations
|
|
|
|
from pyfglair._cffi import ffi, lib
|
|
|
|
|
|
def _make_session():
|
|
cfg = ffi.new("fgl_config_t*")
|
|
cfg.host = ffi.new("char[]", b"127.0.0.1")
|
|
cfg.device_port = 9
|
|
cfg.dsn = ffi.new("char[]", b"AC000W00MOCK0001")
|
|
cfg.lanip_key = ffi.new("char[]", b"key")
|
|
cfg.lanip_key_id = 1
|
|
cfg.tmpl = 0
|
|
cfg.listen_port = 0
|
|
cfg.keepalive_ms = 1000
|
|
cfg.max_queue = 40
|
|
session = lib.fgl_session_create(cfg, ffi.NULL)
|
|
assert session != ffi.NULL
|
|
return session
|
|
|
|
|
|
def test_poll_wait_empty():
|
|
session = _make_session()
|
|
try:
|
|
out = ffi.new("fgl_event_t[]", 4)
|
|
assert lib.fgl_session_poll_events(session, out, 4) == 0
|
|
assert lib.fgl_session_wait_events(session, out, 4, 0) == 0
|
|
assert lib.fgl_session_events_dropped(session) == 0
|
|
finally:
|
|
lib.fgl_session_destroy(session)
|
|
|
|
|
|
def test_poll_invalid_args():
|
|
session = _make_session()
|
|
try:
|
|
out = ffi.new("fgl_event_t[]", 4)
|
|
assert lib.fgl_session_poll_events(session, ffi.NULL, 4) == 0
|
|
assert lib.fgl_session_poll_events(session, out, 0) == 0
|
|
assert lib.fgl_session_wait_events(session, out, 0, 0) == 0
|
|
assert lib.fgl_session_poll_events(ffi.NULL, out, 4) == 0
|
|
assert lib.fgl_session_events_dropped(ffi.NULL) == 0
|
|
finally:
|
|
lib.fgl_session_destroy(session)
|
|
|
|
|
|
def test_create_null_config():
|
|
assert lib.fgl_session_create(ffi.NULL, ffi.NULL) == ffi.NULL
|
|
|
|
|
|
def test_log_poll_empty():
|
|
out = ffi.new("fgl_log_event_t[]", 4)
|
|
assert lib.fgl_log_poll_events(out, 4) >= 0
|
|
assert lib.fgl_log_wait_events(out, 4, 0) >= 0
|
|
assert lib.fgl_log_events_dropped() >= 0
|