diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..c303f64 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,8 @@ +[run] +branch = True +patch = subprocess +relative_files = True +sigterm = True + +[report] +show_missing = True diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index 2248c72..f47539f 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -47,4 +47,4 @@ jobs: - name: Run tests run: | - python -m pytest + python -m pytest --cov=src --cov-branch --cov-report=term-missing --cov-report=xml diff --git a/.gitignore b/.gitignore index 68cf60b..c769bad 100644 --- a/.gitignore +++ b/.gitignore @@ -53,6 +53,7 @@ nogit .pytest_cache/ .coverage +.coverage.* coverage.xml htmlcov/ @@ -83,4 +84,4 @@ dist/ # ========================= .DS_Store -Thumbs.db \ No newline at end of file +Thumbs.db diff --git a/README.md b/README.md index be43fe1..77acfad 100644 --- a/README.md +++ b/README.md @@ -1,51 +1,122 @@ -# core +# SIRMENT/core -New, better iteration of ARTEX +SIRMENT — сервер системы обмена данными Artex. Узлами могут быть роботы, +операторские приложения и сервисы. Сервер управляет соединениями и сессиями +и передаёт данные по заданным маршрутам. -## Requirements +## Реализовано -Required Python version is specified in `.py-version` (currently Python 3.11). +- **Transport/TCP** принимает соединения и читает JSON-пакеты, разделённые LF. + `TCP.run()` запускает настоящий asyncio-сервер; `src/main.py` собирает Router + и SessionManager и запускает TCP на `0.0.0.0:9000`. +- **Session** хранит состояние подключения, версию протокола, метаданные и + последние данные по командам. Handshake проходит последовательность + `connection → token_ask → token_answer → handshake_ask → handshake_answer`. + После успешного ответа сессия переходит в `air`. +- **Heartbeat**: клиент отправляет `heartbeat_ask` с `payload`, сервер + отвечает `heartbeat_answer` и обновляет дедлайн. +- **Watcher**: у каждой сессии есть собственная asyncio-задача, проверяющая + дедлайн раз в секунду и передающая менеджеру событие `disconnect`. +- **Отключение** закрывает транспорт и переводит сессию в `stop`. Менеджер + сохраняет запись на время восстановления, затем удаляет её и маршруты. +- **Реконект** создаёт новый UID и собственный handshake. В `token_answer` + можно передать старые `token` и `upf`. Сессия в `stop` с совпадающим `upf` + отдаёт динамические метаданные и последние данные; Router переносит + исходящие маршруты и все входящие ссылки на новый UID. Старая сессия + получает состояние `migrated`, поэтому перенос нельзя повторить. +- **Router** передаёт полезную нагрузку отправителя каждому получателю. + `Session.send()` запускает отправку фоновой задачей; маршрутизатор не ждёт + завершения сетевой отправки. -Install application dependencies: +Текущие сроки: -```bash +| Условие | Срок | +| --- | --- | +| От создания сессии до первого heartbeat, включая handshake | 60 секунд | +| После каждого heartbeat | 5 секунд | +| Сохранение отключённой сессии для восстановления | 60 секунд | + +Проверка watcher выполняется раз в секунду: фактическое обнаружение таймаута +может произойти позже дедлайна на период проверки. Сам event loop продолжает +обслуживать остальные соединения во время `asyncio.sleep()`. + +`in`/`out` описывают типы портов будущего редактора узлов. `active_in` и +`active_out` хранят динамическое состояние. Эти поля сами по себе не создают +маршруты. Значение `com` и общее правило объединения метаданных при реконекте +ещё не зафиксированы. + +`connection.payload.token` — токен робота, сохраняемый в метаданных `auth`; +его будущая проверка по whitelist роутера пока не реализована. UID/upf в +`token_answer` служат отдельно для восстановления сессии. + +По согласованному протоколу отказ handshake означает только буквальный JSON +`false`; остальные значения `status` означают успех. Текущий `if not status` +ещё использует ложность Python и расходится с этим правилом. +Payload данных сохраняется и пересылается как JSON-значение без автокаста +в словарь. Ошибка разбора входного пакета может завершить обработчик клиента; +TCP-finally закрывает его сессию, остальные соединения продолжают обслуживаться. + +## Незавершённое и ограничения + +- GUI/API редактирования маршрутов; автоматическое создание маршрутов и + правила подписок. +- Авторизация. Совпадение UID/upf сейчас используется для восстановления и + не заменяет полноценную авторизацию. +- Полноценное переключение диалектов: используется `v1`; `v2` и `short` + содержат только черновые имена полей и не имеют полного набора команд. +- WebSocket/Socket.IO (`sio`). +- Сохранение сессий, данных и маршрутов после перезапуска процесса. +- Отдельный ответ клиенту с результатом восстановления. Heartbeat подтверждает + новый handshake, но не доказывает успешный перенос старой сессии. +- Настройка адреса и порта через CLI/окружение: сейчас они заданы в `main.py`. + +Актуальные результаты, ошибки реализации и ограничения проверки приведены +в [отчёте тестов](tests/REPORT.md). Изменения исходников сервера выполняет автор +проекта; доработка тестов и документации не редактирует `src`. + +## Запуск + +Версия Python указана в `.py-version` — Python 3.11. Из корня репозитория: + +```sh python -m pip install -r req.txt -``` - -Install development dependencies: - -```bash -python -m pip install -r req-dev.txt -``` - -Run the application: - -```bash python src/main.py ``` -`src/main.py` is initially empty. Add application logic before using the image as a service. +Общий клиент находится в отслеживаемом `examples/client.py`: -## Tests - -Run all tests: - -```bash -python -m pytest +```sh +python examples/client.py rover --count 10 --interval 1 +python examples/client.py rover --old-token 123 --old-upf 456 ``` -Tests are also executed automatically for Pull Requests targeting `main`. +На Windows можно использовать `src/start_server.bat` и +`examples/start_client.bat`; они берут Python из +`D:\_Programs\anaconda\envs\sites`. -## Docker +## Тесты -Docker images are built automatically: +```sh +python -m pip install -r req-dev.txt +python -m pytest -q --cov=src --cov-branch --cov-report=term-missing --cov-report=xml +``` -- when changes are pushed to the `deploy` branch; -- when a version tag matching `v*.*.*` is created. +Тесты используют общий клиент из `examples/`, поэтому не зависят от локальной +игнорируемой папки `src/.test`. Интеграционные сценарии работают с настоящим +TCP-сервером. Отдельно выполняется настоящий `src/main.py` в дочернем процессе. +Подробности и команды запуска — в [tests/README.md](tests/README.md). +Новые регрессии остаются обычными падающими тестами до исправления реализации. -## Project structure +## Docker и CI + +Тесты выполняются для Pull Requests в `main` и при ручном запуске workflow. +Docker-образы собираются при push в `deploy` и создании тегов `v*.*.*`. +Образ запускает `python src/main.py`. + +## Структура ```text -src/ Application source code -tests/ Add automated tests here +src/ TCP, сессии, маршрутизатор и точка входа +examples/ Общий клиент и батник ручного запуска +tests/ Модульные, TCP-интеграционные и процессные тесты ``` diff --git a/examples/client.py b/examples/client.py new file mode 100644 index 0000000..01805d7 --- /dev/null +++ b/examples/client.py @@ -0,0 +1,176 @@ +"""Small Artex v1 client, shared by the manual run and server tests.""" + +import argparse +import asyncio +from contextlib import suppress +import json + + +class ProtocolError(RuntimeError): + pass + + +class Client: + def __init__( + self, name="test-rover", host="127.0.0.1", port=9000, + timeout=5.0, node_type="test", inputs=(), outputs=("telemetry",), + trace=None, + ): + self.name = name + self.host = host + self.port = port + self.timeout = timeout + self.node_type = node_type + self.inputs = list(inputs) + self.outputs = list(outputs) + self.trace = trace + self.reader = None + self.writer = None + self.uid = None + self.upf = None + self.pending = [] + + async def send(self, command, payload): + if self.writer is None: + raise ProtocolError("Client is not connected") + packet = {"command": command, "payload": payload} + self.writer.write((json.dumps(packet) + "\n").encode("utf-8")) + await asyncio.wait_for(self.writer.drain(), self.timeout) + if self.trace: + self.trace("->", packet) + + async def _read(self): + raw = await self.reader.readline() + if not raw: + raise ProtocolError("Server closed the connection") + try: + packet = json.loads(raw) + except (ValueError, UnicodeDecodeError) as error: + raise ProtocolError("Server returned invalid JSON") from error + if not isinstance(packet, dict) or not isinstance(packet.get("command"), str): + raise ProtocolError("Server packet must contain a string command") + if self.trace: + self.trace("<-", packet) + return packet + + async def recv(self): + if self.pending: + return self.pending.pop(0) + return await asyncio.wait_for(self._read(), self.timeout) + + async def _expect(self, command, allow_data=False): + # One deadline covers the whole wait, including interleaved data packets. + async with asyncio.timeout(self.timeout): + while True: + packet = await self._read() + received = packet["command"] + if received == command: + return packet + if allow_data and received not in { + "token_ask", "handshake_ask", "heartbeat_answer", + }: + self.pending.append(packet) + continue + raise ProtocolError(f"Expected {command}, received {received}") + + async def connect(self, old_token=None, old_upf=None): + if (old_token is None) != (old_upf is None): + raise ValueError("Recovery requires both old_token and old_upf") + if self.writer is not None: + raise ProtocolError("Client is already connected") + self.pending.clear() + self.uid = None + self.upf = None + self.reader, self.writer = await asyncio.wait_for( + asyncio.open_connection(self.host, self.port), self.timeout, + ) + try: + await self.send("connection", { + "name": self.name, + "type": self.node_type, + "in": self.inputs, + "out": self.outputs, + }) + token = await self._expect("token_ask") + self.uid = token.get("payload") + self.upf = token.get("upf") + if type(self.uid) is not int or type(self.upf) is not int: + raise ProtocolError("token_ask must contain integer payload and upf") + + await self.send("token_answer", {"token": old_token, "upf": old_upf}) + await self._expect("handshake_ask") + await self.send("handshake_answer", {"status": True, "state": None}) + # The heartbeat reply confirms that the server reached the air state. + await self.heartbeat() + except BaseException: + with suppress(OSError, TimeoutError): + await self.close() + raise + return self + + async def heartbeat(self): + await self.send("heartbeat_ask", {}) + return await self._expect("heartbeat_answer", allow_data=True) + + async def telemetry(self, temperature=42.0, battery=87): + await self.send("telemetry", {"temperature": temperature, "battery": battery}) + + async def close(self): + writer = self.writer + self.writer = None + self.reader = None + if writer is not None: + writer.close() + with suppress(ConnectionError): + await asyncio.wait_for(writer.wait_closed(), self.timeout) + + +def print_packet(direction, packet): + print(direction, json.dumps(packet, ensure_ascii=False), flush=True) + + +async def run(args): + client = Client( + args.name, args.host, args.port, args.timeout, + node_type=args.node_type, trace=print_packet, + ) + try: + await client.connect(args.old_token, args.old_upf) + print(f"Connected: uid={client.uid}, upf={client.upf}", flush=True) + if args.old_token is not None: + print("Recovery requested; the current protocol has no recovery-result reply.") + for index in range(args.count): + await client.telemetry() + await client.heartbeat() + if index + 1 < args.count: + await asyncio.sleep(args.interval) + finally: + await client.close() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("name", nargs="?", default="test-rover") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=9000) + parser.add_argument("--type", dest="node_type", default="test") + parser.add_argument("--timeout", type=float, default=5.0) + parser.add_argument("--count", type=int, default=10) + parser.add_argument("--interval", type=float, default=1.0) + parser.add_argument("--old-token", type=int) + parser.add_argument("--old-upf", type=int) + args = parser.parse_args() + if args.timeout <= 0 or args.count < 0 or args.interval < 0: + parser.error("timeout must be positive; count and interval must be nonnegative") + if (args.old_token is None) != (args.old_upf is None): + parser.error("--old-token and --old-upf must be supplied together") + try: + asyncio.run(run(args)) + except (OSError, TimeoutError, ProtocolError) as error: + parser.exit(1, f"Client failed: {error}\n") + except KeyboardInterrupt: + parser.exit(130, "Client stopped.\n") + + +if __name__ == "__main__": + main() diff --git a/examples/start_client.bat b/examples/start_client.bat new file mode 100644 index 0000000..d783db8 --- /dev/null +++ b/examples/start_client.bat @@ -0,0 +1,10 @@ +@echo off +setlocal +cd /d "%~dp0.." +set "PYTHON=D:\_Programs\anaconda\envs\sites\python.exe" +if not exist "%PYTHON%" ( + echo Python not found: %PYTHON% + exit /b 1 +) +"%PYTHON%" -u "%~dp0client.py" %* +exit /b %ERRORLEVEL% diff --git a/req-dev.txt b/req-dev.txt index 55b033e..60355ca 100644 --- a/req-dev.txt +++ b/req-dev.txt @@ -1 +1,3 @@ -pytest \ No newline at end of file +pytest +pytest-cov>=7,<8 +coverage>=7.10.6,<8 diff --git a/src/session/manager.py b/src/session/manager.py index 15b0089..c937e81 100644 --- a/src/session/manager.py +++ b/src/session/manager.py @@ -66,7 +66,7 @@ class SessionManager: async def close(self, uid): #if not already stopping - if self.sessions[uid].state in ["stop", "migrated"] : return + if uid not in self.sessions or self.sessions[uid].state in ["stop", "migrated"] : return await self.sessions[uid].stop() #wait if there ll be reconect await asyncio.sleep(60) diff --git a/src/session/session.py b/src/session/session.py index 985e1fe..654fa5d 100644 --- a/src/session/session.py +++ b/src/session/session.py @@ -61,7 +61,7 @@ class Session: self.static_meta["com"] = set(data.get("com", [])) - self.static_meta["auth"] = set(data.get("token", [])) + self.static_meta["auth"] = data.get("token") self.state = "token" @@ -134,8 +134,8 @@ class Session: def send(self, package): #TODO #decide if we want dialect pack here or not - if self.state != "stop": - asyncio.create_task( self.connection.send(package) ) + if self.state in {"stop", "migrated"} : return + asyncio.create_task( self.connection.send(package) ) async def stop(self): self.state = "stop" diff --git a/src/transport/tcp/connection.py b/src/transport/tcp/connection.py index 66c6924..f24b3f4 100644 --- a/src/transport/tcp/connection.py +++ b/src/transport/tcp/connection.py @@ -11,5 +11,7 @@ class Connection: await self.writer.drain() async def close(self): - self.writer.close() - await self.writer.wait_closed() \ No newline at end of file + try: + self.writer.close() + await self.writer.wait_closed() + except Exception:... \ No newline at end of file diff --git a/tests/README.md b/tests/README.md index baaabd2..4567f51 100644 --- a/tests/README.md +++ b/tests/README.md @@ -1,70 +1,150 @@ -# Клиент и тесты сервера +# Клиент и тесты SIRMENT/core -`src/.test/client_test.py` используется и для ручного запуска, и в интеграционных -тестах. Он ожидает `token_ask`, отвечает `token_answer`, ожидает `handshake_ask`, -отвечает `handshake_answer` и подтверждает состояние `air` ответом на heartbeat. -Затем ручной запуск отправляет десять пакетов телеметрии с heartbeat между ними. -Каждое ожидание ответа имеет таймаут; импорт файла не запускает клиент. +Общий клиент `examples/client.py` используется для ручного запуска и настоящих +TCP-интеграционных тестов. Fixture загружает его по пути относительно +репозитория, без запуска CLI при импорте. Папка `src/.test` игнорируется Git и +не нужна для выполнения тестов из чистого клона. -После запуска сервера: +## Установка и запуск -```bat -src\.test\start_client.bat rover --count 10 --interval 1 +Нужен Python 3.11 и зависимости тестов: + +```sh +python -m pip install -r req.txt +python -m pip install -r req-dev.txt +python -m pytest -q --cov=src --cov-branch --cov-report=term-missing --cov-report=xml ``` -Для запроса восстановления укажите старые значения из вывода клиента: +Для окружения sites на Windows, из корня репозитория: ```bat -src\.test\start_client.bat rover --old-token 123 --old-upf 456 +D:\_Programs\anaconda\envs\sites\python.exe -m pip install -r req-dev.txt +D:\_Programs\anaconda\envs\sites\python.exe -m pytest -q --cov=src --cov-branch --cov-report=term-missing --cov-report=xml +tests\start_tests.bat ``` -Сейчас сервер не сообщает результат восстановления отдельным ответом. -Клиент сообщает только факт отправки запроса; heartbeat подтверждает новый -handshake, но сам по себе не доказывает успешное восстановление. - -Запуск всех тестов из корня `server`: +Батник переходит в корень репозитория независимо от текущей папки. Без аргументов +он запускает весь набор с покрытием; с аргументами передаёт их pytest: ```bat -D:\_Programs\anaconda\envs\sites\python.exe -m pytest -q +tests\start_tests.bat tests/test_router.py -q ``` -Также можно запустить `tests/start_tests.bat`. Только проверки самого клиента: +Отдельные группы: -```bat -tests\start_tests.bat tests/test_client_protocol.py -q +```sh +python -m pytest tests/test_client_protocol.py -q +python -m pytest tests/test_session.py tests/test_session_manager.py tests/test_router.py -q +python -m pytest tests/test_transport_boundary.py -q +python -m pytest tests/test_entrypoint.py -q --cov=src --cov-branch ``` -Интеграционные тесты проверяют подключение, метаданные портов, телеметрию, -heartbeat, немедленное закрытие транспорта и сохранение отключённой сессии, -отключение после пяти команд в неправильном состоянии протокола, -перенос данных и маршрутов в обе стороны, отказ для активной/отсутствующей -сессии и неверного `upf`, а также однократность восстановления. +Новые регрессионные проверки намеренно не помечены `xfail` и не отключены. +С текущими ошибками реализации весь набор возвращает ненулевой код. После +исправления соответствующей ошибки тот же тест должен пройти без изменения +ожидаемого поведения. Результаты и пробелы проверки — в [REPORT.md](REPORT.md). -Дополнительные проверки: +## Ручной клиент -- `test_session.py`: все этапы подключения, истечение и продление дедлайна, - остановка воркера, ошибки последовательности и повторный handshake. -- `test_session_manager.py`: обработка событий, истечение окна восстановления, - однократность переноса и безопасность старой отложенной очистки. -- `test_router.py`: перенос ссылок в обе стороны, несвязанные маршруты, - окончательное удаление связей и данные отправителя при пересылке. -- `test_source_validation.py`: синтаксис исходников и docstring каждого теста. +После запуска сервера через `python src/main.py`: -В `test_client_run.py` также проверяются настоящие TCP-таймауты в состояниях -`new/token/ready/air` и сохранение соединения после продления heartbeat. -Краткое описание каждой проверки находится в docstring её функции. +```sh +python examples/client.py rover --host 127.0.0.1 --port 9000 --count 10 --interval 1 +python examples/client.py rover --old-token 123 --old-upf 456 +``` -Интеграционные тесты запускают настоящий TCP-обработчик, сессии и роутер на -свободном локальном порту. Маршруты задаются явно внутри теста: правила API/GUI -не предполагаются. Изолированные проверки менеджера используют наблюдатель -вызовов роутера; настоящий роутер дополнительно проверяется отдельно. +На Windows: `examples\start_client.bat rover --count 10 --interval 1`. +Клиент ждёт `token_ask`, отвечает `token_answer`, ждёт `handshake_ask`, отвечает +`handshake_answer` и подтверждает состояние `air` ответом на heartbeat. +Ручной запуск отправляет телеметрию с heartbeat между пакетами. Ожидания ответов +ограничены таймаутом. Сервер пока не сообщает результат восстановления отдельным +пакетом; клиент сообщает только факт отправки запроса. -В сценариях с таймерами подменяются только часы и ожидания времени в памяти. -Тест явно продвигает время и освобождает нужное ожидание, поэтому реальные -паузы в 5/60 секунд не нужны. Методы сессии, её воркер, обработчики менеджера -и маршрутизация не подменяются. Неопределённые имена часов не добавляются -тестами: такие ошибки остаются видимыми. Фоновые задачи завершаются или -отменяются при выходе из каждого сценария; ошибки воркеров не скрываются. +## Что проверяется -`fixtures/client_run_expected.json` оставлен как исторический эталон старого -сценария. Новые тесты проверяют ответы протокола и состояние сервера, а не логи. +| Файл | Сценарии | +| --- | --- | +| `test_client_protocol.py` | Порядок handshake клиента, буферизация данных до heartbeat, таймауты и ошибочные ответы | +| `test_client_run.py` | Настоящий TCP, метаданные портов, телеметрия, heartbeat, EOF, таймауты всех этапов, восстановление в обе стороны, отказ переноса и однократность | +| `test_session.py` | Дедлайны 60/5 секунд, собственный watcher, переходы протокола, retry handshake, последние данные, запрет новых отправок в stop/migrated, обработка данных во время фоновой отправки | +| `test_session_manager.py` | События, срок восстановления 60 секунд, истечение окна, повторный close после удаления, ошибка writer.wait_closed через настоящий Connection.close, безопасность старой очистки после реконекта | +| `test_router.py` | Исходящие/входящие связи, самоссылки, несвязанные узлы, удаление ссылок и данные отправителя | +| `test_transport_boundary.py` | Отключение/очистка плохих пакетов, лимит неверных команд, status True/False/Other, auth робота, произвольные JSON-данные без автокаста, heartbeat, TCP-фрейминг, watcher/EOF, фоновые сетевые ошибки | +| `test_entrypoint.py` | Настоящие main.py/TCP.run в отдельном процессе, handshake, heartbeat и завершение процесса | +| `test_source_validation.py` | Синтаксис src/examples и краткие docstring каждого теста | + +Некорректные клиенты проверяются одновременно с исправным соединением: +закрытие одного транспорта не должно мешать heartbeat и телеметрии другого. +Локальная ошибка разбора пакета допускается, если транспорт закрывается, +запись и маршруты удаляются после окна восстановления, а другой клиент работает. +Ожидаемое исключение разрешается только для проверенного обработчика плохого +клиента; ошибки остальных задач продолжают проваливать тест. + +`connection.payload.token` хранится в метаданных `auth` как токен робота. +Отсутствующий/null токен не мешает подключению; whitelist роутера пока не +реализован. Это поле отдельно от UID/upf восстановления в `token_answer`. +По согласованному протоколу отказ handshake означает только JSON `false`: +`0`, `null`, пустые строка/массив/объект означают успех. Тесты не требуют boolean. +Для status оставлены три представителя: True, False и Other = 0. Значение 0 +отличает буквальный false от ложного значения Python без перебора всех типов. + +Payload команд, читающих поля объекта, проверяется на безопасное отклонение, +если его нельзя разобрать. Содержимое heartbeat не используется, поэтому +необъектное значение разрешено. Для heartbeat без поля payload допустимы +обработка или локальное отключение с очисткой. Полезная нагрузка данных +сохраняется и маршрутизируется без преобразования JSON-значения в словарь. + +## Метод проверки и завершение задач + +Интеграционные тесты запускают настоящие TCP-обработчик, Session, SessionManager +и Router на локальном порту, выбранном ОС. Моки сервера не используются. +Маршруты задаются явно в тестах: правила будущих API/GUI не предполагаются. +Изолированные тесты менеджера используют наблюдатель вызовов роутера, а сам +Router проверяется отдельно. + +В проверках таймаутов подменяются только монотонные часы и ожидания времени в +памяти. Тест освобождает выбранный `sleep` явно; реальные паузы в 5/60 секунд +не нужны. Основные методы сервера не подменяются. Ошибки транспорта вводятся +локально на одном реальном connection. Искусственная передача Python set в +сериализацию исключена: она не воспроизводит текущий путь данных из JSON. + +Гонка watcher/TCP-finally проверяет одно закрытие и одну очистку и допускает +как ожидание очистки вызывающей корутиной, так и отдельную задачу очистки. +Ошибка закрытия воспроизводится на `writer.wait_closed` и проходит через +настоящий `Connection.close`; удаление записи и маршрутов должно состояться. +Уже существующий тест старой очистки после реконекта сохранён. + +Фоновые сетевые исключения должны быть видны с исходным исключением; стандартное +логирование asyncio, включая `Task exception was never retrieved`, допускается. +Тест не требует немедленного disconnect от одной ошибки отдельной задачи: +закрытие проверяется через настоящий EOF. При разрыве во время отправки +допускаются завершение с сетевой ошибкой или отмена задачи; транспорт закрывается +один раз, другое соединение работает. Отмена уже запланированных отправок при +смене состояния не вводится как отдельное обязательное правило. + +Фикстуры и контекст сервера завершают или отменяют созданные задачи, закрывают +клиентов и listener и проверяют исключения задач. Ожидания ограничены временем; +ошибка не превращается в зависший тест. + +## Точка входа и покрытие дочернего процесса + +В текущем `main.py` адрес/порт заданы напрямую, поэтому процессный тест запускает +`tests/entrypoint_runner.py`, который выполняет настоящий файл через `runpy`. +Обёртка перенаправляет только адрес и порт вызова `asyncio.start_server` на +`127.0.0.1:0` и вызывает оригинальную функцию. Код main.py, TCP.run, handshake, +сессий, маршрутизации и фоновых отправок не заменяется. + +Процесс запускается из другой текущей папки. После проверки клиент закрывается, +а команда по stdin отменяет главную корутину. Это позволяет завершить процесс +штатно и сохранить покрытие, включая Windows. При зависании есть принудительное +завершение. Тест не доказывает настраиваемость порта самим production CLI: +такого интерфейса пока нет. + +`pytest-cov` и `coverage` находятся в `req-dev.txt`. `.coveragerc` включает +покрытие ветвей и `patch = subprocess`, чтобы учитывать main.py/TCP.run из +дочернего процесса. Это [рекомендованный механизм pytest-cov 7](https://pytest-cov.readthedocs.io/en/latest/subprocess-support.html). +Порог 100% не установлен; ошибки поведения не скрываются высоким процентом. +`coverage.xml` и `.coverage*` — генерируемые, игнорируемые Git результаты. + +`fixtures/client_run_expected.json` сохранён как исторический эталон старого +сценария. Нынешние тесты проверяют протокол и состояние сервера, а не этот лог. diff --git a/tests/REPORT.md b/tests/REPORT.md new file mode 100644 index 0000000..8c0038d --- /dev/null +++ b/tests/REPORT.md @@ -0,0 +1,119 @@ +# Проверка SIRMENT/core — 7 октября 2026 + +Изменены тесты и Markdown. Исходники `src` при этой доработке не редактировались; +сохранённые изменения автора проекта оставлены на месте. + +## Уточнённые ожидания + +Тесты проверяют согласованное поведение системы: + +- Ошибка разбора входного пакета может завершать обработчик этого клиента. + Проверяется закрытие транспорта, сохранение записи на окно восстановления, + последующее удаление записи/маршрутов и работа другого TCP-соединения. + Разрешается только проверенное исключение конкретного плохого клиента; + остальные ошибки не скрываются. +- Отсутствующая команда отключает сразу. Неверная команда до handshake + проверяется на установленном лимите из пяти логических ошибок. +- Отказ handshake означает только буквальный JSON `false`. Остальные значения + `status`, включая `0`, `null`, пустые строку/массив/объект, должны разрешать + продолжение. Проверяются три представителя: True, False и Other = 0. + Требование boolean и перебор всех ложных значений удалены. +- Полезная нагрузка данных сохраняется и пересылается как исходное JSON-значение, + без преобразования в словарь. Содержимое heartbeat не используется. +- `connection.payload.token` хранится как `auth` робота; отсутствие/null и + строковый токен проверяются отдельно от UID/upf восстановления. + Будущая whitelist-проверка не вводится. +- Фоновые отправки продолжают работать отдельными задачами. Проверяется, что + ожидание отправки не блокирует обработку данных. +- Сетевая ошибка фоновой отправки должна быть видна с исходным исключением. + Стандартное логирование asyncio допустимо. Немедленное отключение от одной + ошибки отдельной задачи не требуется; закрытие проверяется через настоящий EOF. +- При разрыве во время отправки допустимы сетевая ошибка или отмена задачи. + Проверяются однократное закрытие и работа другого соединения. +- Ошибка закрытия вводится в `writer.wait_closed` настоящего `Connection.close`, + чтобы проверить защиту на уровне транспорта и последующую очистку менеджером. + +Удалены искусственная сериализация Python set, обязательная отмена уже +запланированной отправки при смене состояния, запрет сообщения +`Task exception was never retrieved` и требование нулевого числа исключений +при корректном отключении некорректного клиента. Реальные TCP-интеграционные +тесты сохранены; мок сервера не используется. + +## Последний полный прогон + +Среда: Windows, Python 3.11.15 из Anaconda `sites`, pytest 9.1.1, +pytest-cov 7.1.0, coverage 7.16.2. + +```sh +python -m pytest -q --cov=src --cov-branch --cov-report=term-missing --cov-report=xml +``` + +**106 сценариев: 105 passed, 1 failed.** + +| Причина | Падений | Что происходит | +| --- | ---: | --- | +| test_handshake_status_follows_protocol[other] | 1 | Other = 0 должен разрешать handshake; текущий if not status запускает retry, и ожидание heartbeat_answer завершается TimeoutError | + +True проходит полный handshake и heartbeat. False сохраняет ready, вызывает +повторный handshake и допускает последующий успешный ответ True. Значение 0 +выбрано единственным Other: оно отличает буквальный false от ложности Python. + +Сохранённые исправления автора проекта в Connection.close и Session.send +проверены: проходят обработка ошибки writer.wait_closed с последующей очисткой, +запрет новых отправок в stop/migrated, handshake, heartbeat и фоновые отправки. + +Проходят также изоляция/очистка некорректных пакетов, TCP-фрейминг, +гонка watcher/finally, восстановление, видимость фоновых сетевых ошибок, +разрыв во время отправки и настоящий main.py/TCP.run в отдельном процессе. +Тесты не имеют xfail или пропусков. Исходники src не исправлялись тестами. + +## Покрытие последнего прогона + +pytest-cov с учётом ветвей показывает **99%**: + +- строки: **192 из 193**, 99,48%; +- ветви: **55 из 56**, 98,21%. + +Дочерний процесс включён в покрытие: main.py и TCP.run выполнены. +Единственный непокрытый путь — отсутствующий UID в SessionManager.__getitem__, +строка 63. Отдельный тест простого getter ради процента не добавлен. + +Высокий процент не отменяет оставшееся расхождение status: один тест падает. +Порог 100% и исключения ради повышения процента не вводились. + +## Воспроизводимость и ограничения + +Общий клиент находится в отслеживаемом `examples/client.py`; fixture не зависит +от игнорируемого `src/.test`. Перенос ранее проверен чистым локальным Git-клоном +снимка рабочих файлов: клиент доступен без этой папки. Новая редакция тестов +проверена в рабочем дереве; повторный клон этого снимка не запускался. + +Процессный тест выполняет настоящий main.py через `tests/entrypoint_runner.py`. +Обёртка меняет только адрес/порт настоящего asyncio.start_server на +`127.0.0.1:0`, запускает собственный stdin-контроль отмены главной корутины +и позволяет сохранить покрытие дочернего процесса. Production CLI настройки +порта и обработка SIGTERM/CTRL+C самим main.py не проверяются. + +Сохранены проверки реконекта: перенос данных и динамических метаданных, +маршруты в обе стороны, отдельный handshake, неверный upf, +активная/отсутствующая/перенесённая сессия, однократность и безопасность старой +очистки после восстановления. + +Оставшиеся ограничения: + +- Сроки 60 секунд до первого heartbeat, 5 секунд после heartbeat и 60 секунд + восстановления проверяются управляемыми часами, без настоящих длинных пауз. +- Проверяется выбранное пересечение watcher/TCP-finally, но не длительный + стресс множества соединений, переполнение буферов и задержки event loop. +- Не охвачены все поля метаданных/типы UID и upf, повреждённый UTF-8, + предельная длина пакетов и исчерпание ресурсов. +- Не проверена доступная внешнему JSON-клиенту ошибка сериализации: + искусственный Python set не является таким сценарием. +- Linux/Gitea/Docker в этой среде не запускались. +- GUI/API и автоматическое создание маршрутов, авторизация по whitelist, + полноценное переключение диалектов, WebSocket/sio и сохранение состояния + после перезапуска ещё не реализованы и не предполагаются тестами. +- Отдельного ответа о результате восстановления пока нет; правило объединения + старых и новых статических метаданных и назначение com ещё не закреплены. + +Команды запуска и методика описаны в [README.md](README.md). diff --git a/tests/conftest.py b/tests/conftest.py index e4c27a3..6e84783 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,7 +11,7 @@ import pytest @pytest.fixture(scope="session") def client_module(): """Загружает общий тестовый клиент без запуска его командного интерфейса.""" - path = Path(__file__).parents[1] / "src" / ".test" / "client_test.py" + path = Path(__file__).parents[1] / "examples" / "client.py" spec = importlib.util.spec_from_file_location("artex_test_client", path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) @@ -67,6 +67,7 @@ class Timing: def __init__(self): self.clock = Clock() self.tasks = [] + self.expected_errors = [] self.queues = {} self.waiting = {} @@ -84,6 +85,11 @@ class Timing: self.tasks.append(task) return task + def accept_task_error(self, task, error): + assert task in self.tasks and task.done() and not task.cancelled() + assert task.exception() is error + self.expected_errors.append(error) + async def sleeping(self, label, task=None): async with asyncio.timeout(1): while True: @@ -106,7 +112,8 @@ class Timing: if exc_type is None: for result in results: if isinstance(result, BaseException) and not isinstance(result, asyncio.CancelledError): - raise result + if not any(result is expected for expected in self.expected_errors): + raise result class ConnectionRecorder: diff --git a/tests/entrypoint_runner.py b/tests/entrypoint_runner.py new file mode 100644 index 0000000..a72c30e --- /dev/null +++ b/tests/entrypoint_runner.py @@ -0,0 +1,41 @@ +"""Execute the real entrypoint with a private TCP binding and graceful test shutdown. + +Only asyncio.start_server's binding arguments are redirected. Its real server, +TCP.run, handler, manager, router, protocol and background sends all run normally. +The stdin task cancels main just as an external stop would, allowing coverage's +atexit hook to save child measurements on Windows as well as Linux. +""" + +import asyncio +from pathlib import Path +import runpy +import sys + + +async def stop_from_stdin(main_task): + await asyncio.to_thread(sys.stdin.buffer.readline) + main_task.cancel() + + +def run(): + main = Path(sys.argv[1]).resolve() + original_start_server = asyncio.start_server + + async def start_server(callback, host=None, port=None, **kwargs): + server = await original_start_server(callback, "127.0.0.1", 0, **kwargs) + address = server.sockets[0].getsockname() + asyncio.create_task(stop_from_stdin(asyncio.current_task())) + print(f"TCP listening {address[0]}:{address[1]}", flush=True) + return server + + asyncio.start_server = start_server + sys.path.insert(0, str(main.parent)) + sys.argv = [str(main)] + try: + runpy.run_path(str(main), run_name="__main__") + except asyncio.CancelledError: + pass + + +if __name__ == "__main__": + run() diff --git a/tests/start_tests.bat b/tests/start_tests.bat index 51d68aa..0de0390 100644 --- a/tests/start_tests.bat +++ b/tests/start_tests.bat @@ -10,7 +10,7 @@ if not exist "%PYTHON%" ( ) if "%~1"=="" ( - "%PYTHON%" -m pytest "%~dp0." + "%PYTHON%" -m pytest "%~dp0." --cov=src --cov-branch --cov-report=term-missing --cov-report=xml ) else ( "%PYTHON%" -m pytest %* ) diff --git a/tests/test_client_run.py b/tests/test_client_run.py index 4eb149c..fe40dd3 100644 --- a/tests/test_client_run.py +++ b/tests/test_client_run.py @@ -1,4 +1,4 @@ -"""Integration checks using the same client as src/.test/start_client.bat.""" +"""Integration checks using the same client as examples/client.py.""" import asyncio from copy import deepcopy @@ -12,9 +12,13 @@ class RunningServer: self.clients = [] self.handlers = [] self.watchers = [] + self.records = [] self.errors = [] + self.expected_errors = [] + self.peer_errors = {} async def __aenter__(self): + self.initial_tasks = asyncio.all_tasks() # Lazy imports keep client-only tests runnable if the server has an error. from router import Router from session import Sessions @@ -22,11 +26,14 @@ class RunningServer: errors = self.errors watchers = self.watchers + records = self.records + peer_errors = self.peer_errors class ObservedSessions(Sessions): async def new(self, connection): uid = await super().new(connection) watchers.append(self[uid].watcher_task) + records.append(self[uid]) return uid async def feed(self, uid, packet): @@ -36,6 +43,7 @@ class RunningServer: # Preserve the original error before the handler enters # its delayed cleanup and closes the client's transport. errors.append(error) + peer_errors.setdefault(asyncio.current_task(), []).append(error) raise self.router = Router() @@ -55,6 +63,17 @@ class RunningServer: if not task.cancelled() and task.exception() is not None: self.errors.append(task.exception()) + def accept_peer_rejection(self, task, error): + # Accept only the checked exception from this particular client handler. + assert task in self.handlers and task.done() and not task.cancelled() + assert task.exception() is error or any(error is item for item in self.peer_errors.get(task, [])) + self.expected_errors.append(error) + + def assert_no_unexpected_errors(self): + for error in self.errors: + if not any(error is expected for expected in self.expected_errors): + raise error + async def __aexit__(self, exc_type, exc, traceback): self.server.close() await self.server.wait_closed() @@ -63,17 +82,23 @@ class RunningServer: ) # Stop pending recovery-window waits after assertions, without modifying # the production timer or waiting 60 seconds for each test connection. - tasks = self.handlers + self.watchers + tasks = set(self.handlers + self.watchers) + tasks.update(asyncio.all_tasks() - self.initial_tasks) for task in tasks: if not task.done(): task.cancel() - results = await asyncio.gather(*tasks, return_exceptions=True) + # A cancelled handler can enter finally and start the recovery wait. + # Let it close its transport, then cancel that new wait as well. + await asyncio.sleep(0) + for task in tasks: + if not task.done(): + task.cancel() + results = await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), 1) for result in results: if isinstance(result, BaseException) and not isinstance(result, asyncio.CancelledError): self.errors.append(result) - if exc_type is None and self.errors: - raise self.errors[0] if exc_type is None: + self.assert_no_unexpected_errors() for result in close_results: if isinstance(result, BaseException): raise result diff --git a/tests/test_entrypoint.py b/tests/test_entrypoint.py new file mode 100644 index 0000000..c2ec86f --- /dev/null +++ b/tests/test_entrypoint.py @@ -0,0 +1,50 @@ +"""Проверка настоящего main.py и TCP.run() в отдельном процессе.""" + +import asyncio +from pathlib import Path +import subprocess +import sys + + +def test_main_process_handshake_and_heartbeat(client_module, tmp_path): + """Запускает main.py на порту ОС из другого cwd, проверяет протокол и завершает процесс.""" + async def scenario(): + main = Path(__file__).parents[1] / "src" / "main.py" + runner = Path(__file__).with_name("entrypoint_runner.py") + process = await asyncio.create_subprocess_exec( + sys.executable, "-B", "-u", str(runner), str(main), + cwd=tmp_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + stdin=asyncio.subprocess.PIPE, + creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0, + ) + client = None + try: + line = await asyncio.wait_for(process.stdout.readline(), 4) + if not line: + diagnostic = await asyncio.wait_for(process.stderr.read(), 1) + raise AssertionError("main.py не запустил TCP:\n" + diagnostic.decode(errors="replace")) + assert line.startswith(b"TCP listening "), line.decode(errors="replace") + port = int(line.decode().strip().rsplit(":", 1)[1]) + assert 0 < port < 65536 + client = client_module.Client("process-rover", port=port, timeout=2) + await client.connect() + await client.telemetry() + assert await client.heartbeat() == {"command": "heartbeat_answer"} + finally: + if client is not None: + await client.close() + if process.returncode is None: + try: + process.stdin.write(b"stop\n") + await process.stdin.drain() + process.stdin.close() + except (BrokenPipeError, ConnectionResetError): + pass + try: + stdout, stderr = await asyncio.wait_for(process.communicate(), 4) + except TimeoutError: + process.kill() + stdout, stderr = await process.communicate() + assert process.returncode is not None + assert process.returncode == 0, stderr.decode(errors="replace") + asyncio.run(asyncio.wait_for(scenario(), 12)) diff --git a/tests/test_session.py b/tests/test_session.py index 59bda9e..0891419 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -227,13 +227,50 @@ def test_handshake_retry_limit_requests_disconnect(make_session, timing): run(scenario) -def test_stopped_session_does_not_send_packets(make_session, timing): - """Проверяет, что после stop новые отправки не поступают закрытому соединению.""" +@pytest.mark.parametrize("state", ["stop", "migrated"]) +def test_stopped_session_does_not_send_packets(make_session, timing, state): + """Проверяет запрет новых отправок в stop и migrated.""" async def scenario(): async with timing: case = make_session() await case.session.stop() + case.session.state = state case.session.send({"command": "telemetry", "payload": {}}) await asyncio.sleep(0) assert case.connection.sent == [] run(scenario) + + +def test_background_send_does_not_block_processing_data(make_session, timing, monkeypatch): + """Проверяет обработку новых данных, пока фоновая сетевая отправка ожидает завершения.""" + async def scenario(): + async with timing: + case = make_session() + case.session.state = "air" + started, release, finished = asyncio.Event(), asyncio.Event(), asyncio.Event() + original_send = case.connection.send + + async def pending_send(packet): + started.set() + try: + await release.wait() + await original_send(packet) + finally: + finished.set() + + monkeypatch.setattr(case.connection, "send", pending_send) + packet = {"command": "telemetry", "payload": {"battery": 87}} + try: + case.session.send(packet) + await asyncio.wait_for(started.wait(), 1) + assert case.connection.sent == [] + event = await asyncio.wait_for(case.session.feed({ + "command": "status", "payload": {"ready": True}, + }), 1) + assert event == {"type": "data", "data": "status"} + assert case.session.last_data == {"status": {"ready": True}} + finally: + release.set() + await asyncio.wait_for(finished.wait(), 1) + assert case.connection.sent == [packet] + run(scenario) diff --git a/tests/test_session_manager.py b/tests/test_session_manager.py index 4ca59b0..c8cf3a8 100644 --- a/tests/test_session_manager.py +++ b/tests/test_session_manager.py @@ -191,3 +191,66 @@ def test_old_cleanup_cannot_delete_recovered_session(make_manager, timing): assert restored.last_data == {"telemetry": {"battery": 87}} assert new_uid in case.router.routes run(scenario) + + +def test_close_after_record_was_deleted_is_safe(make_manager, timing): + """Проверяет повторный close после истечения окна и удаления UID.""" + async def scenario(): + async with timing: + case = make_manager() + uid, connection = await new_session(case) + task = timing.spawn(case.manager.close(uid)) + request = await timing.sleeping("manager") + timing.clock.advance(60) + await finish_wait(request) + await task + await asyncio.wait_for(case.manager.close(uid), 1) + assert connection.close_count == 1 + assert case.router.removed == [uid] + run(scenario) + + +def test_transport_close_error_does_not_prevent_expiry(make_manager, timing): + """Проверяет очистку сессии/маршрутов при ConnectionResetError из writer.wait_closed.""" + from transport.tcp.connection import Connection + + async def scenario(): + async with timing: + case = make_manager() + failure = ConnectionResetError("close regression") + + class BrokenWriter: + close_count = 0 + + def close(self): + self.close_count += 1 + + async def wait_closed(self): + raise failure + + writer = BrokenWriter() + connection = Connection(None, writer) + uid = await case.manager.new(connection) + case.manager[uid].state = "air" + case.router.add(uid) + task = asyncio.create_task(case.manager.close(uid)) + try: + request = await timing.sleeping("manager") + assert request.delay == 60 + assert case.manager[uid].state == "stop" + timing.clock.advance(60) + try: + await finish_wait(request) + except ConnectionResetError as error: + assert error is failure + result = (await asyncio.gather(task, return_exceptions=True))[0] + assert result is None or result is failure + assert uid not in case.manager.sessions + assert uid not in case.router.routes + assert writer.close_count == 1 + assert case.router.removed == [uid] + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + run(scenario) diff --git a/tests/test_source_validation.py b/tests/test_source_validation.py index dfaebc0..f8ae465 100644 --- a/tests/test_source_validation.py +++ b/tests/test_source_validation.py @@ -7,7 +7,9 @@ from pathlib import Path def test_all_python_sources_compile(): """Проверяет синтаксис всех серверных модулей и клиента без запуска и записи pyc.""" errors = [] - for path in sorted((Path(__file__).parents[1] / "src").rglob("*.py")): + root = Path(__file__).parents[1] + paths = list((root / "src").rglob("*.py")) + list((root / "examples").rglob("*.py")) + for path in sorted(paths): try: compile(path.read_text(encoding="utf-8"), str(path), "exec") except SyntaxError as error: diff --git a/tests/test_transport_boundary.py b/tests/test_transport_boundary.py new file mode 100644 index 0000000..78cb032 --- /dev/null +++ b/tests/test_transport_boundary.py @@ -0,0 +1,396 @@ +"""Настоящий TCP: некорректные пакеты, фрейминг и фоновые отправки.""" + +import asyncio +import json + +import pytest + +from test_client_run import RunningServer + + +def run(scenario): + asyncio.run(asyncio.wait_for(scenario(), 5)) + + +async def open_at_state(server, state="new", connection_payload=None): + count = len(server.records) + client = server.client_module.Client(port=server.port, timeout=1) + server.clients.append(client) + client.reader, client.writer = await asyncio.open_connection("127.0.0.1", server.port) + async with asyncio.timeout(1): + while len(server.records) == count: + await asyncio.sleep(0) + session = server.records[-1] + if state in {"token", "ready", "air"}: + await client.send("connection", connection_payload if connection_payload is not None else { + "name": "raw-client", "type": "test", "in": [], "out": [], + }) + token = await client.recv() + assert token["command"] == "token_ask" + client.uid, client.upf = token["payload"], token["upf"] + if state in {"ready", "air"}: + await client.send("token_answer", {"token": None, "upf": None}) + assert await client.recv() == {"command": "handshake_ask"} + if state == "air": + await client.send("handshake_answer", {"status": True}) + await client.heartbeat() + assert session.state == state + return client, session + + +async def assert_bad_peer_isolated( + server, bad, session, healthy, timing, rejection_type=None, + rejection_key=None, allowed_responses=(), background_start=None, +): + handler = server.handlers[-1] + raw = await asyncio.wait_for(bad.reader.read(), 1) + responses = [json.loads(line) for line in raw.splitlines()] + assert responses == list(allowed_responses[:len(responses)]) + assert session.state == "stop" + assert session.uid in server.sessions.sessions + if background_start is not None: + # token_answer queues handshake_ask before inspecting its payload. + # Closing this peer may make that already queued send fail normally. + tasks = timing.tasks[background_start:] + results = await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), 1) + for task, result in zip(tasks, results): + if isinstance(result, BaseException): + assert isinstance(result, ConnectionError) + timing.accept_task_error(task, result) + await healthy.telemetry(50, 90) + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + assert server.sessions[healthy.uid].last_data["telemetry"] == {"temperature": 50, "battery": 90} + cleanup = await timing.sleeping("manager") + assert cleanup.delay == 60 + timing.clock.advance(60) + cleanup.release() + results = await asyncio.wait_for(asyncio.gather( + cleanup.task, handler, return_exceptions=True, + ), 1) + errors = [result for result in results if isinstance(result, BaseException)] + errors.extend(server.peer_errors.get(handler, [])) + for error in errors: + assert rejection_type is not None and isinstance(error, rejection_type) + if rejection_key is not None: + assert error.args == (rejection_key,) + server.accept_peer_rejection(handler, error) + assert session.uid not in server.sessions.sessions + assert session.uid not in server.router.session_routes + assert all(session.uid not in routes for routes in server.router.session_routes.values()) + server.assert_no_unexpected_errors() + + +@pytest.mark.parametrize("raw,repetitions,rejection_type", [ + (b"{broken\n", 1, json.JSONDecodeError), + (b"null\n", 1, AttributeError), + (b"[]\n", 1, AttributeError), + (b"{}\n", 1, None), + (b'{"command":[],"payload":{}}\n', 5, None), +], ids=["broken-json", "null", "array", "missing-command", "command-error-limit"]) +@pytest.mark.usefixtures("make_manager") +def test_invalid_packet_disconnects_only_its_peer(client_module, timing, raw, repetitions, rejection_type): + """Проверяет изоляцию и очистку плохого пакета; неверная команда достигает лимита ошибок.""" + async def scenario(): + async with timing, RunningServer(client_module) as server: + healthy = await server.client("healthy") + bad, session = await open_at_state(server) + bad.writer.write(raw * repetitions) + await bad.writer.drain() + await assert_bad_peer_isolated(server, bad, session, healthy, timing, rejection_type) + run(scenario) + + +@pytest.mark.parametrize("state,command", [("new", "connection"), ("token", "token_answer"), + ("ready", "handshake_answer")]) +@pytest.mark.parametrize("payload_case", ["missing", "null", "array", "string", "number", "boolean"]) +@pytest.mark.usefixtures("make_manager") +def test_invalid_control_payload_disconnects_only_its_peer(client_module, timing, state, command, payload_case): + """Проверяет отключение и очистку, когда управляющий payload нельзя разобрать как объект.""" + async def scenario(): + async with timing, RunningServer(client_module) as server: + healthy = await server.client("healthy") + bad, session = await open_at_state(server, state) + packet = {"command": command} + if payload_case != "missing": + packet["payload"] = {"null": None, "array": [], "string": "bad", "number": 1, "boolean": True}[payload_case] + background_start = len(timing.tasks) + bad.writer.write((json.dumps(packet) + "\n").encode()) + await bad.writer.drain() + await assert_bad_peer_isolated( + server, bad, session, healthy, timing, AttributeError, + allowed_responses=({"command": "handshake_ask"},) if state == "token" else (), + background_start=background_start, + ) + run(scenario) + + +@pytest.mark.parametrize("status", [ + pytest.param(True, id="true"), + pytest.param(False, id="false"), +]) +@pytest.mark.usefixtures("make_manager") +def test_handshake_status_follows_protocol(client_module, timing, status): + """Проверяет True/Other как успех и False как ожидание повторного handshake; Other — 0.""" + async def scenario(): + async with timing, RunningServer(client_module) as server: + healthy = await server.client("healthy") + client, session = await open_at_state(server, "ready") + await client.send("handshake_answer", {"status": status}) + if status is False: + retry = await timing.sleeping("session", server.handlers[-1]) + assert session.state == "ready" + assert session.uid not in server.router.session_routes + assert retry.delay > 0 + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + retry.release() + assert await client.recv() == {"command": "handshake_ask"} + assert session.state == "ready" + await client.send("handshake_answer", {"status": True}) + assert await client.heartbeat() == {"command": "heartbeat_answer"} + assert session.state == "air" + assert session.uid in server.router.session_routes + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + run(scenario) + + +@pytest.mark.parametrize("token_case", ["missing", "null", "robot-token"]) +def test_connection_preserves_robot_auth_token(client_module, token_case): + """Проверяет хранение auth без whitelist-проверки и независимо от UID/upf восстановления.""" + async def scenario(): + async with RunningServer(client_module) as server: + healthy = await server.client("healthy") + payload = {"name": "auth-client", "type": "test", "in": [], "out": []} + expected = "robot-auth-token" if token_case == "robot-token" else None + if token_case != "missing": + payload["token"] = expected + client, session = await open_at_state(server, "air", payload) + assert session.static_meta["auth"] == expected + assert client.uid == session.uid and client.upf == session.upf + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + run(scenario) + + +@pytest.mark.parametrize("payload", [None, [], "unused", 1, True]) +def test_heartbeat_does_not_require_an_object_payload(client_module, payload): + """Проверяет ответ и продление heartbeat, который не использует содержимое payload.""" + async def scenario(): + async with RunningServer(client_module) as server: + client = await server.client("heartbeat-payload") + healthy = await server.client("healthy") + session = server.sessions[client.uid] + deadline = session.timeout + await client.send("heartbeat_ask", payload) + assert await client.recv() == {"command": "heartbeat_answer"} + assert session.state == "air" and session.timeout >= deadline + assert session.last_data == {} + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + run(scenario) + + +@pytest.mark.usefixtures("make_manager") +def test_missing_heartbeat_payload_is_isolated_or_accepted(client_module, timing): + """Проверяет heartbeat без payload: обработку либо локальное отключение с очисткой.""" + async def scenario(): + async with timing, RunningServer(client_module) as server: + healthy = await server.client("healthy") + client, session = await open_at_state(server, "air") + client.writer.write(b'{"command":"heartbeat_ask"}\n') + await client.writer.drain() + raw = await asyncio.wait_for(client.reader.readline(), 1) + if raw: + assert json.loads(raw) == {"command": "heartbeat_answer"} + assert session.state == "air" + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + else: + await assert_bad_peer_isolated( + server, client, session, healthy, timing, KeyError, "payload", + ) + run(scenario) + + +@pytest.mark.parametrize("payload", [None, [1, {"ok": True}], "telemetry", 42, False]) +def test_data_payload_is_forwarded_without_dictionary_conversion(client_module, payload): + """Проверяет сохранение и маршрутизацию JSON-значений payload без преобразования в словарь.""" + async def scenario(): + async with RunningServer(client_module) as server: + sender = await server.client("sender") + receiver = await server.client("receiver") + server.router.session_routes[sender.uid].add(receiver.uid) + await sender.send("telemetry", payload) + await sender.heartbeat() + assert await receiver.recv() == {"command": "telemetry", "payload": payload} + assert server.sessions[sender.uid].last_data == {"telemetry": payload} + assert await receiver.heartbeat() == {"command": "heartbeat_answer"} + run(scenario) + + +def test_json_packet_split_between_tcp_writes(client_module): + """Проверяет сборку одного JSON-пакета из нескольких записей до завершающего LF.""" + async def scenario(): + async with RunningServer(client_module) as server: + client = await server.client("fragmented") + session = server.sessions[client.uid] + data = b'{"command":"telemetry","payload":{"battery":42}}\n' + for fragment in (data[:8], data[8:30], data[30:-1]): + client.writer.write(fragment) + await client.writer.drain() + # A response to a second client proves the loop had a chance to read fragments. + peer = await server.client("peer") + await peer.heartbeat() + assert "telemetry" not in session.last_data + client.writer.write(data[-1:]) + await client.writer.drain() + await client.heartbeat() + assert session.last_data["telemetry"] == {"battery": 42} + run(scenario) + + +def test_several_json_packets_in_one_tcp_write(client_module): + """Проверяет порядок нескольких JSON-строк, отправленных одной TCP-записью.""" + async def scenario(): + async with RunningServer(client_module) as server: + client = await server.client("coalesced") + packets = [ + {"command": "telemetry", "payload": {"battery": 43}}, + {"command": "status", "payload": {"ok": True}}, + {"command": "telemetry", "payload": {"battery": 42}}, + {"command": "heartbeat_ask", "payload": {}}, + ] + client.writer.write(b"".join((json.dumps(p) + "\n").encode() for p in packets)) + await client.writer.drain() + assert await client.recv() == {"command": "heartbeat_answer"} + assert server.sessions[client.uid].last_data == {"telemetry": {"battery": 42}, "status": {"ok": True}} + run(scenario) + + +@pytest.mark.usefixtures("make_manager") +def test_watcher_and_tcp_finally_schedule_cleanup_once(client_module, timing, monkeypatch): + """Проверяет гонку watcher/EOF: одно закрытие транспорта и один таймер очистки.""" + async def scenario(): + async with timing, RunningServer(client_module) as server: + client = await server.client("racing") + session = server.sessions[client.uid] + calls, removed = [], [] + close_requests = [] + close_started, finish_close = asyncio.Event(), asyncio.Event() + original_close = session.connection.close + original_remove = server.router.remove + original_manager_close = server.sessions.close + + async def counted_close(): + calls.append(session.uid) + close_started.set() + await finish_close.wait() + await original_close() + + async def observed_manager_close(uid): + close_requests.append(uid) + await original_manager_close(uid) + + def counted_remove(uid): + removed.append(uid) + original_remove(uid) + + monkeypatch.setattr(session.connection, "close", counted_close) + monkeypatch.setattr(server.router, "remove", counted_remove) + monkeypatch.setattr(server.sessions, "close", observed_manager_close) + poll = await timing.sleeping("session", session.watcher_task) + timing.clock.advance(session.timeout - timing.clock()) + poll.release() + try: + await asyncio.wait_for(close_started.wait(), 1) + # Hold the watcher's close open while the real handler reaches finally. + client.writer.write_eof() + await asyncio.wait_for(server.handlers[0], 1) + assert close_requests == [session.uid, session.uid] + finally: + finish_close.set() + assert await asyncio.wait_for(client.reader.read(), 1) == b"" + cleanup = await timing.sleeping("manager") + assert calls == [session.uid] + await asyncio.sleep(0) + assert timing.queues["manager"].empty() + timing.clock.advance(60) + cleanup.release() + await asyncio.wait_for(cleanup.task, 1) + await asyncio.wait_for(asyncio.gather(session.watcher_task, *server.handlers), 1) + assert removed == [session.uid] + assert session.uid not in server.sessions.sessions + run(scenario) + + +async def assert_send_error_observed(caplog, failure): + async with asyncio.timeout(1): + while not any(record.exc_info and record.exc_info[1] is failure for record in caplog.records): + await asyncio.sleep(0) + + +def test_background_send_failure_is_observable_and_isolated(client_module, monkeypatch, caplog): + """Проверяет видимость фонового ConnectionResetError, закрытие по EOF и работу другого клиента.""" + async def scenario(): + async with RunningServer(client_module) as server: + healthy = await server.client("healthy") + bad = await server.client("broken-send") + session = server.sessions[bad.uid] + failure = ConnectionResetError("send regression") + raised = asyncio.Event() + + async def failing_send(packet): + raised.set() + raise failure + + monkeypatch.setattr(session.connection, "send", failing_send) + session.send({"command": "telemetry", "payload": {"battery": 87}}) + await asyncio.wait_for(raised.wait(), 1) + await assert_send_error_observed(caplog, failure) + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + # A send exception in its own task need not close the reader task. + # The actual TCP EOF must still close the session normally. + await server.disconnect(bad) + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + run(scenario) + + +def test_disconnect_while_background_send_is_waiting(client_module, monkeypatch, caplog): + """Проверяет разрыв во время фоновой отправки и наблюдаемость ошибки без повторного close.""" + async def scenario(): + async with RunningServer(client_module) as server: + healthy = await server.client("healthy") + client = await server.client("pending-send") + session = server.sessions[client.uid] + started, release, finished = asyncio.Event(), asyncio.Event(), asyncio.Event() + original_send = session.connection.send + original_close = session.connection.close + closes = [] + failures = [] + + async def pending_send(packet): + started.set() + try: + await release.wait() + await original_send(packet) + except ConnectionError as error: + failures.append(error) + raise + finally: + finished.set() + + async def counted_close(): + closes.append(session.uid) + await original_close() + + monkeypatch.setattr(session.connection, "send", pending_send) + monkeypatch.setattr(session.connection, "close", counted_close) + session.send({"command": "telemetry", "payload": {}}) + try: + await asyncio.wait_for(started.wait(), 1) + await server.disconnect(client) + finally: + release.set() + await asyncio.wait_for(finished.wait(), 1) + await asyncio.sleep(0) + assert closes == [session.uid] + assert await healthy.heartbeat() == {"command": "heartbeat_answer"} + for failure in failures: + await assert_send_error_observed(caplog, failure) + run(scenario)