From 11f1bbdd12e2ebdfcff86c5e99e7d46e43cf4822 Mon Sep 17 00:00:00 2001 From: FoxVoland Date: Thu, 8 Oct 2026 19:35:14 +0300 Subject: [PATCH] [+!] sio legacy --- src/main.py | 12 +- src/session/session.py | 38 +- src/transport/__init__.py | 4 +- src/transport/sio_legacy/README.md | 110 +++++ src/transport/sio_legacy/__init__.py | 55 +++ src/transport/sio_legacy/connection.py | 157 ++++++ src/transport/sio_legacy/handler.py | 79 +++ src/transport/sio_legacy/requirements.txt | 2 + .../sio_legacy/tests/test_transport.py | 448 ++++++++++++++++++ 9 files changed, 891 insertions(+), 14 deletions(-) create mode 100644 src/transport/sio_legacy/README.md create mode 100644 src/transport/sio_legacy/__init__.py create mode 100644 src/transport/sio_legacy/connection.py create mode 100644 src/transport/sio_legacy/handler.py create mode 100644 src/transport/sio_legacy/requirements.txt create mode 100644 src/transport/sio_legacy/tests/test_transport.py diff --git a/src/main.py b/src/main.py index e1d16e8..5bd3e7a 100644 --- a/src/main.py +++ b/src/main.py @@ -7,7 +7,7 @@ Created on Mon Oct 5 11:47:32 2026 import asyncio import json -from transport import TCP +from transport import TCP, SIO from session import Sessions from router import Router @@ -20,13 +20,11 @@ router.sessions = sessions async def main(): - tcp = TCP( - "0.0.0.0", - 9000, - session=sessions, - ) + tasks = [] + tasks.append(TCP("0.0.0.0", 9000, session=sessions).run) + tasks.append(SIO("0.0.0.0", 9001, session=sessions).run) - await tcp.run() + await asyncio.gather(*[task() for task in tasks]) asyncio.run(main()) \ No newline at end of file diff --git a/src/session/session.py b/src/session/session.py index 654fa5d..2aeb03e 100644 --- a/src/session/session.py +++ b/src/session/session.py @@ -33,6 +33,10 @@ class Session: } self.last_data = {} + + self.sending = set() + self.retry_data = {} + async def feed (self, package): ''' @@ -45,10 +49,9 @@ class Session: if com is None : return {"type" : "disconnect"} #BAN if self.state == "new" and com == VERSION[self.pv]["CONNECT"]: - ... - #TODO - #process new connection? auth? get and fill some data - + + #get and fill some data + self.pv = data.get("connect_protocol", "v1") self.static_meta["name"] = data.get("name", "unnamed") self.static_meta["type"] = data.get("type", "unknown") @@ -62,6 +65,9 @@ class Session: self.static_meta["com"] = set(data.get("com", [])) self.static_meta["auth"] = data.get("token") + + #TODO + #auth with token self.state = "token" @@ -132,10 +138,30 @@ class Session: self.last_data[command] = data def send(self, package): + + if self.state in {"stop", "migrated"} : return + com = package["command"] + #TODO #decide if we want dialect pack here or not - if self.state in {"stop", "migrated"} : return - asyncio.create_task( self.connection.send(package) ) + + asyncio.create_task( self.process_send(com, package) ) + + async def process_send(self, com, package): + if com in self.sending: + self.retry_data[com] = package + return + self.sending.add(com) + try: + while True: + await self.connection.send(package) + if com not in self.retry_data : break + package = self.retry_data.pop(com) + + finally: + self.sending.discard(com) + self.retry_data.pop(com, None) + async def stop(self): self.state = "stop" diff --git a/src/transport/__init__.py b/src/transport/__init__.py index bb26864..63bfc5b 100644 --- a/src/transport/__init__.py +++ b/src/transport/__init__.py @@ -1 +1,3 @@ -from .tcp import TCP \ No newline at end of file +from .tcp import TCP + +from .sio_legacy import SIO \ No newline at end of file diff --git a/src/transport/sio_legacy/README.md b/src/transport/sio_legacy/README.md new file mode 100644 index 0000000..931159f --- /dev/null +++ b/src/transport/sio_legacy/README.md @@ -0,0 +1,110 @@ +# Socket.IO для старого Artex-robot + +Адаптер подключает старый `core/com/Connector` из Artex-robot к текущим +`SessionManager`, `Session` и `Router`. Код ядра, TCP и `main.py` не изменяется. +Все файлы адаптера, его зависимости и тесты находятся в этой папке. + +## Подключение + +Из папки `server` установите зависимости: + +```powershell +& 'D:\_Programs\anaconda\envs\sites\python.exe' -m pip install -r src/transport/sio_legacy/requirements.txt +``` + +Интерфейс совпадает с TCP: + +```python +from transport.sio_legacy import SIO + +# sessions — тот же менеджер, который используется TCP. +sio = SIO("0.0.0.0", 9001, sessions) +await sio.run() +``` + +TCP и Socket.IO могут работать одновременно с одним менеджером и роутером: +`await asyncio.gather(tcp.run(), sio.run())`. Им нужны разные порты. +Отмена задачи `run()` закрывает listener и активные соединения Socket.IO. +Для порта `0` выбранный порт доступен в `sio.bound_port` после +`await sio.started.wait()`. + +Сервер использует [python-socketio с aiohttp](https://python-socketio.readthedocs.io/en/stable/server.html), +поддерживает polling и WebSocket, namespace `/robot` и обычный путь `/socket.io`. +Это Socket.IO, не отдельный WebSocket-протокол. + +## Соответствие протоколов + +| Старый робот | Пакет для ядра / действие адаптера | +| --- | --- | +| Метаданные в `auth` при соединении | `connection`, диалект `v1` | +| Ответ ACK на `token_ask` | `token_answer` без старого token/upf | +| ACK на `heartbeat(["Ready", "d_meta"])`, `Ready == "ok"` | Успешный `handshake_answer` | +| ACK на `heartbeat(["d_meta"])` | `heartbeat_ask`, обновление срока в ядре | +| `telem` / `video` / другой event с `{"payload": value}` | `{"command": event, "payload": value}` | +| Отправка ядром `command` | Event `command` с тем же `payload` | +| Отправка ядром `com` / `force_com` / `sys_com` | Event `command`, `payload` содержит выбранную категорию | +| Остальные отправки ядра | Event с именем `command` пакета и `{"payload": value}` | + +`Ready` отправляется один раз при успешном handshake: старый клиент запускает +по этому запросу потоки телеметрии и видео. Последующие heartbeat запрашивают +только `d_meta`. Данные, отправленные этими потоками до ACK handshake, ждут +завершения handshake и затем поступают в ядро. + +Первый `Ready` задерживается на `heartbeat_interval`, чтобы исходный Connector +успел завершить инициализацию после `socketio.Client.connect()`. +По умолчанию `heartbeat_interval=1.0`, `ack_timeout=3.0` секунды. +Интервал вместе с ожиданием ACK должен быть меньше пяти секунд. +У ядра остаются прежние сроки: 60 секунд до первого heartbeat, 5 секунд после +heartbeat, 60 секунд хранения отключённой сессии. Истечение ACK также +закрывает соединение; удаление сессии и маршрутов выполняет менеджер. + +`video.payload`, включая JPEG `bytes`, сохраняется в `Session.last_data` +без преобразования. Отправка через Socket.IO передаёт тот же бинарный payload. +Текущий JSON-транспорт TCP не кодирует bytes; преобразование для других +транспортов здесь не добавлено. + +Исходные `telem_can`, `video_can` и другие поля `auth` доступны в +`Connection.metadata`; последнее `d_meta` — в `Connection.dynamic_meta`. +Они не превращаются в `in/out`, активные порты или маршруты автоматически. +Маршруты задаются существующими средствами Router. + +`legacy` пока нужен: у старого робота метаданные передаются через `auth`, +управляющие ответы приходят как Socket.IO ACK, а heartbeat инициирует сервер. +Это отличается от пакетов `v1`. Старый клиент возвращает только `sid`, без +`upf`, и в просмотренной реализации не сохраняет новый token. +Адаптер сохраняет ответ в `Connection.legacy_sid`, но каждое соединение +создаёт новую сессию. Маршруты и данные прежней сессии не восстанавливаются +по одному sid или имени робота. + +Ошибки приёма, отправки и heartbeat записываются в лог с sid/uid и закрывают +только соответствующее соединение. `Session.send()` продолжает работать +через фоновые задачи. Отправки одному SID упорядочены, чтобы бинарные +вложения разных пакетов не перемешивались. + +## Проверки + +Из `server`, после установки зависимостей: + +```powershell +$env:PYTHONDONTWRITEBYTECODE = '1' +& 'D:\_Programs\anaconda\envs\sites\python.exe' -B -m pytest -q -p no:cacheprovider tests src/transport/sio_legacy/tests +``` + +Тесты используют настоящий сервер, ядро и сетевые клиенты. Проверяются +polling/WebSocket, handshake, периодический heartbeat, ранние данные потоков, +bytes, команды, общая маршрутизация TCP/Socket.IO, ошибки фоновой отправки, +таймаут ACK, watcher ядра, повторное закрытие и остановка listener. +Серверные cleanup-таймеры отменяются только при завершении теста. + +Дополнительный `test_original_robot_connector` загружает исходные Connector, +handlers, parsers и метаданные из `E:/robot_src`, запускает его настоящие +сетевые обработчики и потоки. Только оборудование заменено источником +тестовых кадров и приёмником команд. Другой путь задаётся через +`$env:ARTEX_ROBOT_SRC = 'путь/к/artex-robot'`. При отсутствии исходников этот +дополнительный тест пропускается; остальные тесты воспроизводимы без них. +Ключ `-B` предотвращает запись pyc в каталог внешнего проекта. + +Не проверяются физическое оборудование робота, восстановление сессии без +upf, авторизация и доставка бинарных данных через TCP. API/GUI маршрутов, +автоматическое создание маршрутов и сохранение состояния после перезапуска +не реализуются этим адаптером. diff --git a/src/transport/sio_legacy/__init__.py b/src/transport/sio_legacy/__init__.py new file mode 100644 index 0000000..de8cd56 --- /dev/null +++ b/src/transport/sio_legacy/__init__.py @@ -0,0 +1,55 @@ +"""Socket.IO transport for the original Artex robot's /robot namespace.""" + +import asyncio + +from .handler import Handler + +__all__ = ["SIO"] + + +class SIO: + """Use SIO(addr, port, session) and await run(), just like TCP.""" + + def __init__(self, addr, port, session, *, heartbeat_interval=1.0, ack_timeout=3.0): + # Keep optional dependencies out of imports of the TCP transport. + import socketio + from aiohttp import web + + if heartbeat_interval <= 0 or ack_timeout <= 0: + raise ValueError("heartbeat_interval and ack_timeout must be positive") + if heartbeat_interval + ack_timeout >= 5: + raise ValueError("heartbeat_interval + ack_timeout must be below the core's 5-second deadline") + + self.addr = addr + self.port = port + self.session = session + self.sio = socketio.AsyncServer(async_mode="aiohttp", async_handlers=True) + self.app = web.Application() + self.sio.attach(self.app) + self.handler = Handler(self.sio, session, heartbeat_interval, ack_timeout) + self.started = asyncio.Event() + self.bound_port = None + self._running = False + + async def run(self): + """Serve polling/WebSocket connections until this coroutine is cancelled.""" + from aiohttp import web + + if self._running: + raise RuntimeError("SIO.run() is already running") + self._running = True + self.started.clear() + runner = web.AppRunner(self.app, shutdown_timeout=3) + try: + await runner.setup() + site = web.TCPSite(runner, self.addr, self.port) + await site.start() + self.bound_port = runner.addresses[0][1] + self.started.set() + await asyncio.Event().wait() + finally: + await self.handler.shutdown() + await self.sio.shutdown() + await runner.cleanup() + self._running = False + diff --git a/src/transport/sio_legacy/connection.py b/src/transport/sio_legacy/connection.py new file mode 100644 index 0000000..ac3aad3 --- /dev/null +++ b/src/transport/sio_legacy/connection.py @@ -0,0 +1,157 @@ +"""Translate the legacy robot's events and ACKs to the core's v1 packets.""" + +import asyncio +import logging +from collections.abc import Mapping + +NAMESPACE = "/robot" +logger = logging.getLogger(__name__) + + +class Connection: + def __init__(self, sio, sid, session, on_close, heartbeat_interval, ack_timeout): + self.sio = sio + self.sid = sid + self.uid = None + self.session = session + self.on_close = on_close + self.heartbeat_interval = heartbeat_interval + self.ack_timeout = ack_timeout + self.closed = False + self.metadata = {} + self.dynamic_meta = {} + self.legacy_sid = None + self._ready = asyncio.Event() + self._send_lock = asyncio.Lock() + self._feed_lock = asyncio.Lock() + self._tasks = set() + self._heartbeat_task = None + + def start(self, metadata): + self.metadata = metadata + task = asyncio.create_task(self._begin()) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + async def _begin(self): + try: + payload = {**self.metadata, "connect_protocol": "v1"} + await self._feed("connection", payload) + except Exception: + logger.exception("Legacy startup failed: sid=%s uid=%s", self.sid, self.uid) + await self.close() + + async def _feed(self, command, payload): + async with self._feed_lock: + if not self.closed: + await self.session.feed(self.uid, {"command": command, "payload": payload}) + + async def _call(self, event, data): + return await self.sio.call(event, data, to=self.sid, namespace=NAMESPACE, timeout=self.ack_timeout) + + async def send(self, packet): + """Called in the core's existing background send task.""" + task = asyncio.current_task() + self._tasks.add(task) + try: + async with self._send_lock: + if self.closed: + return + command = packet["command"] + if command == "token_ask": + self.legacy_sid = await self._call("token_ask", packet["payload"]) + # The old client sends no UPF. Never fabricate recovery proof. + await self._feed("token_answer", {"token": None, "upf": None}) + elif command == "handshake_ask": + await self._handshake() + elif command == "heartbeat_answer": + # Legacy heartbeat was already acknowledged by the robot. + return + else: + payload = packet["payload"] + if command in {"com", "force_com", "sys_com"}: + payload = {command: payload} + command = "command" + await self.sio.emit(command, {"payload": payload}, to=self.sid, namespace=NAMESPACE) + except Exception: + logger.exception("Legacy send failed: sid=%s uid=%s", self.sid, self.uid) + await self.close() + finally: + self._tasks.discard(task) + + def _read_dynamic_meta(self, answer): + if not isinstance(answer, Mapping) or not isinstance(answer.get("d_meta"), Mapping): + raise ValueError("Legacy heartbeat ACK must contain an object d_meta") + self.dynamic_meta = dict(answer["d_meta"]) + + async def _handshake(self): + # Ready starts the old robot's stream threads. Its Connector finishes + # initialization only after socketio.Client.connect() has returned. + await asyncio.sleep(self.heartbeat_interval) + if self.closed: + return + answer = await self._call("heartbeat", ["Ready", "d_meta"]) + self._read_dynamic_meta(answer) + await self._feed("handshake_answer", { + "status": answer.get("Ready") == "ok", "state": self.dynamic_meta, + }) + if self.closed or self.session[self.uid].state != "air": + return + await self._feed("heartbeat_ask", {}) + self._ready.set() + if self._heartbeat_task is None: + self._heartbeat_task = asyncio.create_task(self._heartbeat()) + self._tasks.add(self._heartbeat_task) + self._heartbeat_task.add_done_callback(self._tasks.discard) + + async def _heartbeat(self): + try: + while not self.closed: + await asyncio.sleep(self.heartbeat_interval) + async with self._send_lock: + if self.closed: + return + answer = await self._call("heartbeat", ["d_meta"]) + self._read_dynamic_meta(answer) + await self._feed("heartbeat_ask", {}) + except Exception: + logger.exception("Legacy heartbeat failed: sid=%s uid=%s", self.sid, self.uid) + await self.close() + + async def receive(self, event, args): + task = asyncio.current_task() + self._tasks.add(task) + try: + if len(args) != 1 or not isinstance(args[0], Mapping) or "payload" not in args[0]: + raise ValueError("Legacy data event must have one object argument containing payload") + # Ready may emit telemetry/video before its handshake ACK arrives. + await self._ready.wait() + await self._feed(event, args[0]["payload"]) + except Exception: + logger.exception("Legacy packet failed: sid=%s uid=%s event=%s", self.sid, self.uid, event) + await self.close() + finally: + self._tasks.discard(task) + + async def close(self): + """Close once; let SessionManager retain the record for its usual 60 s.""" + if self.closed: + return + self.closed = True + self._ready.set() + self.on_close(self) + current = asyncio.current_task() + tasks = [task for task in self._tasks if task is not current and not task.done()] + for task in tasks: + task.cancel() + eio_sid = self.sio.manager.eio_sid_from_sid(self.sid, NAMESPACE) + try: + if self.sio.manager.is_connected(self.sid, NAMESPACE): + await self.sio.disconnect(self.sid, namespace=NAMESPACE) + if eio_sid is not None: + await asyncio.wait_for(self.sio.eio.disconnect(eio_sid), self.ack_timeout) + except Exception: + logger.exception("Legacy transport close failed: sid=%s uid=%s", self.sid, self.uid) + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + diff --git a/src/transport/sio_legacy/handler.py b/src/transport/sio_legacy/handler.py new file mode 100644 index 0000000..98a4e72 --- /dev/null +++ b/src/transport/sio_legacy/handler.py @@ -0,0 +1,79 @@ +"""Bind Socket.IO events to unchanged SessionManager methods.""" + +import asyncio +import logging +from collections.abc import Mapping + +from .connection import Connection, NAMESPACE + +logger = logging.getLogger(__name__) + + +class Handler: + def __init__(self, sio, session, heartbeat_interval, ack_timeout): + self.sio = sio + self.session = session + self.heartbeat_interval = heartbeat_interval + self.ack_timeout = ack_timeout + self.connections = {} + self.cleanup_tasks = set() + self._stopping = False + sio.on("connect", self.connect, namespace=NAMESPACE) + sio.on("disconnect", self.disconnect, namespace=NAMESPACE) + sio.on("*", self.message, namespace=NAMESPACE) + + async def connect(self, sid, environ, auth=None): + if self._stopping or (auth is not None and not isinstance(auth, Mapping)): + return False + connection = Connection( + self.sio, sid, self.session, self.release, + self.heartbeat_interval, self.ack_timeout, + ) + try: + connection.uid = await self.session.new(connection) + self.connections[sid] = connection + # Start after the Socket.IO namespace-connect reply is sent. + # The legacy client also initializes max_freq after connect(). + connection.start(dict(auth or {})) + except Exception: + logger.exception("Legacy connect failed: sid=%s uid=%s", sid, connection.uid) + await connection.close() + return False + return True + + async def disconnect(self, sid, reason=None): + connection = self.connections.get(sid) + if connection is not None: + await connection.close() + + async def message(self, event, sid, *args): + connection = self.connections.get(sid) + if connection is not None: + await connection.receive(event, args) + + def release(self, connection): + if self.connections.get(connection.sid) is connection: + del self.connections[connection.sid] + if connection.uid is None: + return + record = self.session.sessions.get(connection.uid) + # Session.stop already owns cleanup when the core watcher closes us. + if record is None or record.state in {"stop", "migrated"}: + return + task = asyncio.create_task(self.session.close(connection.uid)) + self.cleanup_tasks.add(task) + task.add_done_callback(self._cleanup_finished) + + def _cleanup_finished(self, task): + self.cleanup_tasks.discard(task) + if not task.cancelled() and task.exception() is not None: + error = task.exception() + logger.error("Legacy session cleanup failed", exc_info=(type(error), error, error.__traceback__)) + + async def shutdown(self): + self._stopping = True + await asyncio.gather(*(connection.close() for connection in list(self.connections.values()))) + # Let manager.close enter stop and start its ordinary recovery window. + # Do not cancel these timers: the same core can still serve TCP clients. + await asyncio.sleep(0) + diff --git a/src/transport/sio_legacy/requirements.txt b/src/transport/sio_legacy/requirements.txt new file mode 100644 index 0000000..daf779f --- /dev/null +++ b/src/transport/sio_legacy/requirements.txt @@ -0,0 +1,2 @@ +python-socketio>=5.12,<6 +aiohttp>=3.9,<4 diff --git a/src/transport/sio_legacy/tests/test_transport.py b/src/transport/sio_legacy/tests/test_transport.py new file mode 100644 index 0000000..37ec569 --- /dev/null +++ b/src/transport/sio_legacy/tests/test_transport.py @@ -0,0 +1,448 @@ +"""Real Socket.IO, real core, and clients speaking the old /robot protocol.""" + +import asyncio +from contextlib import asynccontextmanager +import importlib.util +import os +from pathlib import Path +import sys +from time import monotonic + +import pytest +import socketio + +from router import Router +from session import Sessions +from transport.sio_legacy import SIO + +NS = "/robot" + + +def run(scenario): + asyncio.run(asyncio.wait_for(scenario(), 10)) + + +async def until(predicate, timeout=2): + async with asyncio.timeout(timeout): + while not predicate(): + await asyncio.sleep(0.005) + + +class RobotPeer: + """Wire-level legacy client; the server and core are never mocked.""" + + def __init__(self, token=None, burst=()): + self.sio = socketio.AsyncClient(reconnection=False) + self.token = token + self.burst = burst + self.ready_count = 0 + self.heartbeats = 0 + self.ignore_token = False + self.ignore_heartbeat = False + self.received = [] + self.meta = {"sid": token, "ready": "init"} + + @self.sio.on("token_ask", namespace=NS) + async def token_ask(data): + if self.ignore_token: + await asyncio.Event().wait() + return self.token + + @self.sio.on("heartbeat", namespace=NS) + async def heartbeat(requests): + answer = {} + if "Ready" in requests: + self.ready_count += 1 + self.meta["ready"] = "ok" + for event, payload in self.burst: + await self.sio.emit(event, {"payload": payload}, namespace=NS) + answer["Ready"] = "ok" + elif self.ignore_heartbeat: + await asyncio.Event().wait() + if "d_meta" in requests: + self.heartbeats += 1 + answer["d_meta"] = dict(self.meta) + return answer + + @self.sio.on("*", namespace=NS) + async def incoming(event, data): + self.received.append((event, data)) + + async def emit(self, event, payload): + await self.sio.emit(event, {"payload": payload}, namespace=NS) + + +class Server: + def __init__(self, interval, timeout): + self.router = Router() + self.sessions = Sessions(self.router) + self.router.sessions = self.sessions + self.transport = SIO("127.0.0.1", 0, self.sessions, + heartbeat_interval=interval, ack_timeout=timeout) + self.clients = [] + + async def connect(self, peer=None, transports=None, ready=True): + peer = peer or RobotPeer() + self.clients.append(peer.sio) + await peer.sio.connect( + f"http://127.0.0.1:{self.transport.bound_port}", namespaces=[NS], + transports=transports, auth={"name": "robot", "type": "Robot", "telem_can": True, "video_can": True}, + ) + connection = self.transport.handler.connections[peer.sio.get_sid(NS)] + if ready: + await until(lambda: self.sessions[connection.uid].state == "air") + return peer, connection + + +@asynccontextmanager +async def serving(interval=0.03, timeout=0.2): + initial = asyncio.all_tasks() + server = Server(interval, timeout) + task = asyncio.create_task(server.transport.run()) + try: + await asyncio.wait_for(server.transport.started.wait(), 2) + yield server + finally: + await asyncio.gather(*(client.disconnect() for client in server.clients if client.connected), + return_exceptions=True) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + pending = asyncio.all_tasks() - initial + for created in pending: + created.cancel() + await asyncio.sleep(0) + for created in pending: + if not created.done(): + created.cancel() + await asyncio.wait_for(asyncio.gather(*pending, return_exceptions=True), 2) + + +@pytest.mark.parametrize("transport", ["polling", "websocket"]) +def test_legacy_handshake_and_periodic_heartbeat(transport): + """Проверяет /robot, auth, token ACK, Ready ровно один раз и heartbeat обоих HTTP-транспортов.""" + async def scenario(): + async with serving() as server: + peer, connection = await server.connect(transports=[transport]) + record = server.sessions[connection.uid] + first_deadline = record.timeout + await until(lambda: peer.heartbeats >= 3) + assert peer.ready_count == 1 + assert record.timeout > first_deadline + assert record.static_meta["name"] == "robot" + assert record.static_meta["type"] == "Robot" + assert record.static_meta["in"] == record.static_meta["out"] == set() + assert connection.metadata["video_can"] is True + assert connection.dynamic_meta["ready"] == "ok" + assert connection.uid in server.router.session_routes + run(scenario) + + +def test_data_emitted_before_ready_ack_waits_for_handshake(): + """Проверяет данные потоков, запущенных Ready до ACK: нет бана и сохраняется последнее значение.""" + async def scenario(): + burst = [("telem", {"index": n}) for n in range(10)] + [("video", {"camera": b"jpeg"})] + async with serving() as server: + peer, connection = await server.connect(RobotPeer(burst=burst)) + record = server.sessions[connection.uid] + await until(lambda: record.last_data.get("telem") == {"index": 9} and "video" in record.last_data) + assert record.err_count == 0 + assert record.last_data["video"] == {"camera": b"jpeg"} + run(scenario) + + +def test_binary_video_is_preserved_and_routed_over_socketio(): + """Проверяет сохранение JPEG bytes и передачу нескольких бинарных вложений без изменения.""" + async def scenario(): + async with serving() as server: + sender, source = await server.connect() + receiver, target = await server.connect() + server.router.session_routes[source.uid].add(target.uid) + payload = {"front": b"\xff\xd8\x00\xff\xd9", "rear": b"second-jpeg"} + await sender.emit("video", payload) + await until(lambda: bool(receiver.received)) + assert server.sessions[source.uid].last_data["video"] == payload + assert receiver.received == [("video", {"payload": payload})] + run(scenario) + + +@pytest.mark.parametrize("command", ["com", "force_com", "sys_com"]) +def test_command_categories_are_wrapped_for_legacy_robot(command): + """Проверяет преобразование команды ядра в legacy event command с payload нужной категории.""" + async def scenario(): + async with serving() as server: + peer, connection = await server.connect() + server.sessions[connection.uid].send({"command": command, "payload": "forward"}) + await until(lambda: bool(peer.received)) + assert peer.received == [("command", {"payload": {command: "forward"}})] + run(scenario) + + +def test_tcp_and_legacy_socketio_route_telemetry_and_commands(): + """Проверяет настоящий TCP↔Socket.IO через общий Router: телеметрию и команду роботу.""" + async def scenario(): + from transport.tcp.handler import handle + + client_path = Path(__file__).resolve().parents[4] / "examples" / "client.py" + spec = importlib.util.spec_from_file_location("sio_tcp_client", client_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + async with serving() as server: + robot, connection = await server.connect() + tcp = await asyncio.start_server(lambda r, w: handle(r, w, server.sessions), "127.0.0.1", 0) + client = module.Client("operator", port=tcp.sockets[0].getsockname()[1], timeout=1) + try: + await client.connect() + server.router.session_routes[connection.uid].add(client.uid) + server.router.session_routes[client.uid].add(connection.uid) + await robot.emit("telem", {"temperature": "23 °C"}) + assert await client.recv() == {"command": "telem", "payload": {"temperature": "23 °C"}} + await client.send("command", {"com": "forward"}) + await until(lambda: bool(robot.received)) + assert robot.received == [("command", {"payload": {"com": "forward"}})] + finally: + await client.close() + tcp.close() + await tcp.wait_closed() + run(scenario) + + +@pytest.mark.parametrize("bad_payload", [None, [], {}, "bad"]) +def test_malformed_event_disconnects_only_its_robot(bad_payload): + """Проверяет локальное отключение плохого event payload, окно восстановления и работу соседа.""" + async def scenario(): + async with serving() as server: + healthy, good = await server.connect() + bad, connection = await server.connect() + await bad.sio.emit("telem", bad_payload, namespace=NS) + await until(lambda: server.sessions[connection.uid].state == "stop") + assert connection.closed + assert connection.sid not in server.transport.handler.connections + assert connection.uid in server.router.session_routes + count = healthy.heartbeats + await until(lambda: healthy.heartbeats > count) + await healthy.emit("telem", {"ok": True}) + await until(lambda: server.sessions[good.uid].last_data.get("telem") == {"ok": True}) + run(scenario) + + +def test_invalid_auth_is_rejected_without_creating_session(): + """Проверяет отказ несловарному auth без новой сессии и без нарушения другого соединения.""" + async def scenario(): + async with serving() as server: + healthy, good = await server.connect() + bad = socketio.AsyncClient(reconnection=False) + server.clients.append(bad) + with pytest.raises(socketio.exceptions.ConnectionError): + await bad.connect(f"http://127.0.0.1:{server.transport.bound_port}", namespaces=[NS], auth=["bad"]) + assert len(server.sessions.sessions) == 1 + count = healthy.heartbeats + await until(lambda: healthy.heartbeats > count) + assert server.sessions[good.uid].state == "air" + run(scenario) + + +def test_missing_token_ack_closes_transport(): + """Проверяет ограниченное ожидание token ACK и закрытие TCP/WebSocket-транспорта при тишине.""" + async def scenario(): + async with serving() as server: + peer = RobotPeer() + peer.ignore_token = True + peer, connection = await server.connect(peer, ready=False) + await until(lambda: server.sessions[connection.uid].state == "stop") + await until(lambda: not peer.sio.connected) + assert connection.closed + run(scenario) + + +def test_missing_heartbeat_ack_closes_once_and_isolates_peer(monkeypatch): + """Проверяет таймаут heartbeat ACK, одно отключение и непрерывную работу другого робота.""" + async def scenario(): + async with serving() as server: + healthy, good = await server.connect() + bad, connection = await server.connect() + calls = [] + original = server.transport.sio.disconnect + + async def counted(sid, **kwargs): + calls.append(sid) + await original(sid, **kwargs) + + monkeypatch.setattr(server.transport.sio, "disconnect", counted) + bad.ignore_heartbeat = True + await until(lambda: server.sessions[connection.uid].state == "stop") + await connection.close() + assert calls.count(connection.sid) == 1 + count = healthy.heartbeats + await until(lambda: healthy.heartbeats > count) + assert server.sessions[good.uid].state == "air" + run(scenario) + + +def test_core_watcher_can_close_a_pending_legacy_ack(): + """Проверяет закрытие Socket.IO собственным watcher ядра во время ожидания ACK.""" + async def scenario(): + async with serving(timeout=2) as server: + peer = RobotPeer() + peer.ignore_token = True + peer, connection = await server.connect(peer, ready=False) + server.sessions[connection.uid].timeout = monotonic() - 1 + await until(lambda: server.sessions[connection.uid].state == "stop") + await until(lambda: not peer.sio.connected) + assert connection.closed + assert connection.sid not in server.transport.handler.connections + run(scenario) + + +def test_legacy_sid_without_upf_does_not_restore_an_old_session(): + """Проверяет новый UID при legacy реконекте: один sid без upf не переносит данные/маршруты.""" + async def scenario(): + async with serving() as server: + old, connection = await server.connect() + await old.emit("telem", {"old": True}) + await until(lambda: bool(server.sessions[connection.uid].last_data)) + await old.sio.disconnect() + await until(lambda: server.sessions[connection.uid].state == "stop") + fresh, new = await server.connect(RobotPeer(token=connection.uid)) + assert new.uid != connection.uid + assert server.sessions[new.uid].last_data == {} + assert server.sessions[connection.uid].state == "stop" + assert new.legacy_sid == connection.uid + run(scenario) + + +def test_background_send_error_is_logged_and_closes_its_peer(monkeypatch, caplog): + """Проверяет сетевую ошибку фоновой отправки: лог с UID, отключение и работу соседнего робота.""" + async def scenario(): + async with serving() as server: + healthy, good = await server.connect() + bad, connection = await server.connect() + failure = ConnectionResetError("legacy send regression") + original = server.transport.sio.emit + + async def failing(event, data=None, **kwargs): + if kwargs.get("to") == connection.sid: + raise failure + return await original(event, data, **kwargs) + + monkeypatch.setattr(server.transport.sio, "emit", failing) + server.sessions[connection.uid].send({"command": "command", "payload": {"com": "stop"}}) + await until(lambda: server.sessions[connection.uid].state == "stop") + assert str(connection.uid) in caplog.text + assert any(record.exc_info and record.exc_info[1] is failure for record in caplog.records) + count = healthy.heartbeats + await until(lambda: healthy.heartbeats > count) + assert server.sessions[good.uid].state == "air" + run(scenario) + + +def test_concurrent_binary_sends_keep_packet_attachments_together(): + """Проверяет сериализацию фоновых emit одному SID: бинарные вложения не перемешиваются.""" + async def scenario(): + async with serving() as server: + peer, connection = await server.connect() + packets = [{"index": n, "one": bytes([n]), "two": bytes([n + 1])} for n in range(10)] + for payload in packets: + server.sessions[connection.uid].send({"command": "video", "payload": payload}) + await until(lambda: len(peer.received) == len(packets)) + assert peer.received == [("video", {"payload": payload}) for payload in packets] + run(scenario) + + +def test_run_cancellation_closes_listener_and_active_sessions(): + """Проверяет отмену SIO.run: listener закрыт, робот отключён, сессия оставлена в stop.""" + async def scenario(): + async with serving() as server: + peer, connection = await server.connect() + # Find the public run coroutine started by the test fixture. + task = next(task for task in asyncio.all_tasks() + if task.get_coro().__qualname__ == "SIO.run") + task.cancel() + await asyncio.gather(task, return_exceptions=True) + assert server.sessions[connection.uid].state == "stop" + assert connection.closed + assert not server.transport.handler.connections + with pytest.raises(OSError): + reader, writer = await asyncio.open_connection("127.0.0.1", server.transport.bound_port) + run(scenario) + + +def test_original_robot_connector(): + """Проверяет исходный Connector робота: реальные потоки, heartbeat, bytes и получение команды без оборудования.""" + source = Path(os.environ.get("ARTEX_ROBOT_SRC", "E:/robot_src")) + connector_path = source / "core" / "com" / "__init__.py" + meta_path = source / "core" / "vault" / "meta.py" + if not connector_path.is_file() or not meta_path.is_file(): + pytest.skip("Original robot source unavailable; set ARTEX_ROBOT_SRC to run this additional check") + + # Load the original network package without importing hardware or cv2. + package_name = "sio_original_robot_connector" + spec = importlib.util.spec_from_file_location( + package_name, connector_path, submodule_search_locations=[str(connector_path.parent)], + ) + original = importlib.util.module_from_spec(spec) + sys.modules[package_name] = original + spec.loader.exec_module(original) + meta_spec = importlib.util.spec_from_file_location("sio_original_robot_meta", meta_path) + meta = importlib.util.module_from_spec(meta_spec) + meta_spec.loader.exec_module(meta) + + video = {"front": b"\xff\xd8\x00\xff\xd9"} + telemetry = {"battery": 12.5} + + class Vault(meta.Vault_Meta_mixin): + def __init__(self): + super().__init__() + self.smeta.update(name="original-robot", type="Robot", telem_can=True, video_can=True) + self.dmeta.update(sid=None, ready="init", telem_run=False, video_run=False) + self.frames = {"video": video, "telem": telemetry} + + def read_video(self): + return self.frames.pop("video", None) + + def read_telem(self): + return self.frames.pop("telem", None) + + class Conductor: + def __init__(self): + self.commands = [] + + def pass_command(self, command): + self.commands.append(command) + + async def scenario(): + async with serving(interval=0.15, timeout=0.5) as server: + vault, conductor = Vault(), Conductor() + connector = await asyncio.to_thread( + original.Connector, conductor, vault, "http://127.0.0.1", + server.transport.bound_port, max_freq=0.03, + ) + try: + connection = server.transport.handler.connections[connector.sio.get_sid(NS)] + record = server.sessions[connection.uid] + await until(lambda: record.state == "air" and "video" in record.last_data + and "telem" in record.last_data) + assert record.last_data == {"video": video, "telem": telemetry} + assert record.static_meta["name"] == "original-robot" + assert record.err_count == 0 + first_deadline = record.timeout + threads = (connector.telem_thread, connector.video_thread) + await until(lambda: record.timeout > first_deadline) + assert (connector.telem_thread, connector.video_thread) == threads + record.send({"command": "command", "payload": {"com": "forward"}}) + await until(lambda: bool(conductor.commands)) + assert conductor.commands == ["forward"] + finally: + await asyncio.to_thread(connector.disconnect) + for name in ("telem_thread", "video_thread"): + thread = getattr(connector, name, None) + if thread is not None: + await asyncio.to_thread(thread.join, 1) + assert not thread.is_alive() + await until(lambda: record.state == "stop") + + try: + run(scenario) + finally: + for name in list(sys.modules): + if name == package_name or name.startswith(package_name + "."): + del sys.modules[name]