From fac12b2ea89bf49135a24a1d55e79ef39cb65fb0 Mon Sep 17 00:00:00 2001 From: FoxVoland Date: Wed, 7 Oct 2026 15:12:38 +0300 Subject: [PATCH] [+] mvp1 basic working mvp --- .gitignore | 2 +- pytest.ini | 3 + src/main.py | 32 +++ src/router/__init__.py | 43 ++++ src/session/__init__.py | 1 + src/session/manager.py | 74 ++++++ src/session/session.py | 144 +++++++++++ src/session/versions.py | 28 +++ src/start_server.bat | 15 ++ src/transport/__init__.py | 1 + src/transport/tcp/__init__.py | 19 ++ src/transport/tcp/connection.py | 15 ++ src/transport/tcp/handler.py | 19 ++ tests/.gitkeep | 0 tests/README.md | 70 ++++++ tests/conftest.py | 190 ++++++++++++++ tests/fixtures/client_run_expected.json | 36 +++ tests/start_tests.bat | 19 ++ tests/test_client_protocol.py | 114 +++++++++ tests/test_client_run.py | 321 ++++++++++++++++++++++++ tests/test_router.py | 92 +++++++ tests/test_session.py | 239 ++++++++++++++++++ tests/test_session_manager.py | 193 ++++++++++++++ tests/test_source_validation.py | 27 ++ 24 files changed, 1696 insertions(+), 1 deletion(-) create mode 100644 pytest.ini create mode 100644 src/router/__init__.py create mode 100644 src/session/__init__.py create mode 100644 src/session/manager.py create mode 100644 src/session/session.py create mode 100644 src/session/versions.py create mode 100644 src/start_server.bat create mode 100644 src/transport/__init__.py create mode 100644 src/transport/tcp/__init__.py create mode 100644 src/transport/tcp/connection.py create mode 100644 src/transport/tcp/handler.py delete mode 100644 tests/.gitkeep create mode 100644 tests/README.md create mode 100644 tests/conftest.py create mode 100644 tests/fixtures/client_run_expected.json create mode 100644 tests/start_tests.bat create mode 100644 tests/test_client_protocol.py create mode 100644 tests/test_client_run.py create mode 100644 tests/test_router.py create mode 100644 tests/test_session.py create mode 100644 tests/test_session_manager.py create mode 100644 tests/test_source_validation.py diff --git a/.gitignore b/.gitignore index 4411758..68cf60b 100644 --- a/.gitignore +++ b/.gitignore @@ -43,7 +43,7 @@ ENV/ # ========================= nogit -*.test.* +*.test* diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..80432c2 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +testpaths = tests +pythonpath = src diff --git a/src/main.py b/src/main.py index e69de29..e1d16e8 100644 --- a/src/main.py +++ b/src/main.py @@ -0,0 +1,32 @@ +# -*- coding: utf-8 -*- +""" +Created on Mon Oct 5 11:47:32 2026 + +@author: fox +""" +import asyncio +import json + +from transport import TCP +from session import Sessions + +from router import Router + + + +router = Router() +sessions = Sessions(router) +router.sessions = sessions + + +async def main(): + tcp = TCP( + "0.0.0.0", + 9000, + session=sessions, + ) + + await tcp.run() + + +asyncio.run(main()) \ No newline at end of file diff --git a/src/router/__init__.py b/src/router/__init__.py new file mode 100644 index 0000000..4f4a07e --- /dev/null +++ b/src/router/__init__.py @@ -0,0 +1,43 @@ +import asyncio + +class Router: + def __init__(self): + self.sessions = None + self.session_routes = {} + + def add(self, uid): + if self.session_routes.get(uid) is None: + self.session_routes[uid] = set() + print(f"[ROUTER] connected: {uid}") + + def move(self, old_uid, new_uid): + #TODO redo this hardcode + if old_uid in self.session_routes.keys(): + self.session_routes[new_uid] = self.session_routes.pop(old_uid) + + for target in self.session_routes.values(): + if old_uid in target: + target.remove(old_uid) + target.add(new_uid) + + + + + + + def remove(self, uid): + if uid in self.session_routes.keys(): + self.session_routes.pop(uid) + + for target in self.session_routes.values(): + if uid in target : target.remove(uid) + + async def process(self, source, command): + print(f"[ROUTER] {source} -> others: {command}") + + for uid in self.session_routes[source]: + self.sessions[uid].send( + { + "command" : command, + "payload" : self.sessions[source].last_data[command] + }) \ No newline at end of file diff --git a/src/session/__init__.py b/src/session/__init__.py new file mode 100644 index 0000000..9961336 --- /dev/null +++ b/src/session/__init__.py @@ -0,0 +1 @@ +from .manager import SessionManager as Sessions \ No newline at end of file diff --git a/src/session/manager.py b/src/session/manager.py new file mode 100644 index 0000000..15b0089 --- /dev/null +++ b/src/session/manager.py @@ -0,0 +1,74 @@ +import asyncio + +from .session import Session + +class SessionManager: + + def __init__(self, router): + self.router = router + self.sessions = {} + + async def new(self, connection): + uid = id(connection) + + self.sessions[uid] = Session( + uid, + connection, + on_timeout = self.process + ) + + return uid + + async def feed(self, uid, package): + + session = self.sessions[uid] + event = await session.feed(package) + if event is None : return + + await self.process(uid, event) + + + async def process(self, uid, event): + ''' + event = {type : , data : } + event can be only: + - connect - ping the router (or skip if exist) + - reconnect - renamr the key + - data - ping the router + - disconnect - update our state + ''' + match event["type"]: + case "connect": + self.router.add(uid) + case "reconnect": + if event["uid"] in self.sessions.keys() and self.sessions[event["uid"]].upf == event["upf"]: + if self.sessions[event["uid"]].state == "stop": + self.sessions[event["uid"]].state = "migrated" + + self.router.move(event["uid"], uid) + + self.sessions[uid].dynamic_meta = self.sessions[event["uid"]].dynamic_meta + self.sessions[uid].last_data = self.sessions[event["uid"]].last_data + + + case "data": + await self.router.process(uid, event["data"]) + case "disconnect" : + ...#TODO + await self.close(uid) + + + def __getitem__(self, key): + if key not in self.sessions.keys(): + raise KeyError + return self.sessions[key] + + + async def close(self, uid): + #if not already stopping + if self.sessions[uid].state in ["stop", "migrated"] : return + await self.sessions[uid].stop() + #wait if there ll be reconect + await asyncio.sleep(60) + self.sessions.pop(uid, None) + self.router.remove(uid) \ No newline at end of file diff --git a/src/session/session.py b/src/session/session.py new file mode 100644 index 0000000..985e1fe --- /dev/null +++ b/src/session/session.py @@ -0,0 +1,144 @@ +from time import monotonic as now +import asyncio + +from .versions import VERSION + + +class Session: + + def __init__(self, uid, connection, on_timeout): + self.uid = uid + self.connection = connection + + self.timeout = now() + 60 #TODO move timeout to env + self.watcher_task = asyncio.create_task(self.watch(on_timeout)) + + + self.state = "new" + self.err_count = 0 + self.timeout_attempt = 0 + self.upf = hash(now()) + + self.pv = "v1" #default protocol version + self.static_meta = { + # "name" : "unnamed", + # "type" : "unknown", + # "in" : set(), + # "out" : set(), + } + + self.dynamic_meta = { + # "active_in" : set() + # "active_out" : set() + } + + self.last_data = {} + + async def feed (self, package): + ''' + package here is hust a dict with dialect + we meet here to solve the dialect + ''' + com = package.get(VERSION[self.pv]["COMMAND"]) + data = package.get(VERSION[self.pv]["PAYLOAD"]) + + 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 + + + self.static_meta["name"] = data.get("name", "unnamed") + self.static_meta["type"] = data.get("type", "unknown") + + self.static_meta["in"] = set(data.get("in", [])) + self.dynamic_meta["active_in"] = set() + + self.static_meta["out"] = set(data.get("out", [])) + self.dynamic_meta["active_out"] = set() + + + self.static_meta["com"] = set(data.get("com", [])) + self.static_meta["auth"] = set(data.get("token", [])) + + self.state = "token" + + self.send({VERSION[self.pv]["COMMAND"] : VERSION[self.pv]["TOKEN_ASK"], + VERSION[self.pv]["PAYLOAD"] : self.uid, + VERSION[self.pv]["UPF"] : self.upf}) + return + + if self.state == "token" and com == VERSION[self.pv]["TOKEN_ANS"]: + self.state = "ready" + self.send({VERSION[self.pv]["COMMAND"] : VERSION[self.pv]["HANDSHAKE_ASK"]}) + + + old_token = data.get(VERSION[self.pv]["TOKEN"]) + old_upf = data.get(VERSION[self.pv]["UPF"]) + if old_token and old_upf: + return {"type" : "reconnect", + "uid" : old_token, + "upf" : old_upf} + + return + + if self.state == "ready" and com == VERSION[self.pv]["HANDSHAKE_ANS"]: + + if not data.get(VERSION[self.pv]["STATUS"]): + if self.timeout_attempt == 5 : return {"type" : "disconnect"} + self.timeout_attempt += 1 + await asyncio.sleep(self.timeout_attempt) + self.send({VERSION[self.pv]["COMMAND"] : VERSION[self.pv]["HANDSHAKE_ASK"]}) + return {"type" : "await handshake", "state" : data.get(VERSION[self.pv]["STATE"])} + + self.state = "air" + return {"type" : "connect", "uid" : self.uid} + + if self.state == "air": + data = package[VERSION[self.pv]["PAYLOAD"]] + + if com == VERSION[self.pv]["HEARTBEAT_ASK"]: + #TODO update deadline + self.timeout = now() + 5 #TODO move to env + self.send({VERSION[self.pv]["COMMAND"] : VERSION[self.pv]["HEARTBEAT_ANS"]}) + return + + self.update(com, data) + return {"type" : "data", "data" : com} + + self.err_count += 1 + if self.err_count == 5 : return {"type" : "disconnect"} #BAN + + + + + async def watch(self, on_timeout): + while self.state not in {"stop", "migrated"}: + await asyncio.sleep(1) + + if self.state in {"stop", "migrated"}: + return + + if now() >= self.timeout: + await on_timeout( + self.uid, + {"type": "disconnect"}, + ) + return + + def update(self, command, data): + self.last_data[command] = data + + 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) ) + + async def stop(self): + self.state = "stop" + await self.connection.close() + + \ No newline at end of file diff --git a/src/session/versions.py b/src/session/versions.py new file mode 100644 index 0000000..b996c7b --- /dev/null +++ b/src/session/versions.py @@ -0,0 +1,28 @@ +VERSION = { + "v1": { + "COMMAND": "command", + "PAYLOAD": "payload", + + "CONNECT": "connection", + "TOKEN_ASK": "token_ask", + "TOKEN_ANS": "token_answer", + "TOKEN": "token", + "UPF": "upf", + "HANDSHAKE_ASK": "handshake_ask", + "HANDSHAKE_ANS": "handshake_answer", + "STATUS": "status", + "STATE": "state", + "HEARTBEAT_ASK": "heartbeat_ask", + "HEARTBEAT_ANS": "heartbeat_answer", + }, + + "v2": { + "COMMAND": "com", + "PAYLOAD": "arg", + }, + + "short": { + "COMMAND": "c", + "PAYLOAD": "p", + }, +} \ No newline at end of file diff --git a/src/start_server.bat b/src/start_server.bat new file mode 100644 index 0000000..7fc7b10 --- /dev/null +++ b/src/start_server.bat @@ -0,0 +1,15 @@ +@echo off +setlocal +cd /d "%~dp0" + +set "PYTHON=D:\_Programs\anaconda\envs\sites\python.exe" +if not exist "%PYTHON%" ( + echo Python not found: %PYTHON% + pause + exit /b 1 +) + +"%PYTHON%" -u "%~dp0main.py" %* +set "EXIT_CODE=%ERRORLEVEL%" +pause +exit /b %EXIT_CODE% diff --git a/src/transport/__init__.py b/src/transport/__init__.py new file mode 100644 index 0000000..bb26864 --- /dev/null +++ b/src/transport/__init__.py @@ -0,0 +1 @@ +from .tcp import TCP \ No newline at end of file diff --git a/src/transport/tcp/__init__.py b/src/transport/tcp/__init__.py new file mode 100644 index 0000000..d5be812 --- /dev/null +++ b/src/transport/tcp/__init__.py @@ -0,0 +1,19 @@ +import asyncio +from .handler import handle + + +class TCP: + def __init__(self, addr, port, session): + self.addr = addr + self.port = port + self.session = session + + async def run(self): + server = await asyncio.start_server( + lambda r, w: handle(r, w, self.session), + self.addr, + self.port, + ) + + async with server: + await server.serve_forever() \ No newline at end of file diff --git a/src/transport/tcp/connection.py b/src/transport/tcp/connection.py new file mode 100644 index 0000000..66c6924 --- /dev/null +++ b/src/transport/tcp/connection.py @@ -0,0 +1,15 @@ +import json + +class Connection: + def __init__(self, reader, writer): + self.reader = reader + self.writer = writer + + async def send(self, packet): + data = (json.dumps(packet) + "\n").encode() + self.writer.write(data) + await self.writer.drain() + + async def close(self): + self.writer.close() + await self.writer.wait_closed() \ No newline at end of file diff --git a/src/transport/tcp/handler.py b/src/transport/tcp/handler.py new file mode 100644 index 0000000..00368be --- /dev/null +++ b/src/transport/tcp/handler.py @@ -0,0 +1,19 @@ +from .connection import Connection +import json + +async def handle(reader, writer, session): + connection = Connection(reader, writer) + + uid = await session.new(connection) + + try: + while True: + raw = await reader.readline() + if not raw : break + + packet = json.loads(raw) + await session.feed(uid, packet) + + finally: + await session.close(uid) + \ No newline at end of file diff --git a/tests/.gitkeep b/tests/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..baaabd2 --- /dev/null +++ b/tests/README.md @@ -0,0 +1,70 @@ +# Клиент и тесты сервера + +`src/.test/client_test.py` используется и для ручного запуска, и в интеграционных +тестах. Он ожидает `token_ask`, отвечает `token_answer`, ожидает `handshake_ask`, +отвечает `handshake_answer` и подтверждает состояние `air` ответом на heartbeat. +Затем ручной запуск отправляет десять пакетов телеметрии с heartbeat между ними. +Каждое ожидание ответа имеет таймаут; импорт файла не запускает клиент. + +После запуска сервера: + +```bat +src\.test\start_client.bat rover --count 10 --interval 1 +``` + +Для запроса восстановления укажите старые значения из вывода клиента: + +```bat +src\.test\start_client.bat rover --old-token 123 --old-upf 456 +``` + +Сейчас сервер не сообщает результат восстановления отдельным ответом. +Клиент сообщает только факт отправки запроса; heartbeat подтверждает новый +handshake, но сам по себе не доказывает успешное восстановление. + +Запуск всех тестов из корня `server`: + +```bat +D:\_Programs\anaconda\envs\sites\python.exe -m pytest -q +``` + +Также можно запустить `tests/start_tests.bat`. Только проверки самого клиента: + +```bat +tests\start_tests.bat tests/test_client_protocol.py -q +``` + +Интеграционные тесты проверяют подключение, метаданные портов, телеметрию, +heartbeat, немедленное закрытие транспорта и сохранение отключённой сессии, +отключение после пяти команд в неправильном состоянии протокола, +перенос данных и маршрутов в обе стороны, отказ для активной/отсутствующей +сессии и неверного `upf`, а также однократность восстановления. + +Дополнительные проверки: + +- `test_session.py`: все этапы подключения, истечение и продление дедлайна, + остановка воркера, ошибки последовательности и повторный handshake. +- `test_session_manager.py`: обработка событий, истечение окна восстановления, + однократность переноса и безопасность старой отложенной очистки. +- `test_router.py`: перенос ссылок в обе стороны, несвязанные маршруты, + окончательное удаление связей и данные отправителя при пересылке. +- `test_source_validation.py`: синтаксис исходников и docstring каждого теста. + +В `test_client_run.py` также проверяются настоящие TCP-таймауты в состояниях +`new/token/ready/air` и сохранение соединения после продления heartbeat. +Краткое описание каждой проверки находится в docstring её функции. + +Интеграционные тесты запускают настоящий TCP-обработчик, сессии и роутер на +свободном локальном порту. Маршруты задаются явно внутри теста: правила API/GUI +не предполагаются. Изолированные проверки менеджера используют наблюдатель +вызовов роутера; настоящий роутер дополнительно проверяется отдельно. + +В сценариях с таймерами подменяются только часы и ожидания времени в памяти. +Тест явно продвигает время и освобождает нужное ожидание, поэтому реальные +паузы в 5/60 секунд не нужны. Методы сессии, её воркер, обработчики менеджера +и маршрутизация не подменяются. Неопределённые имена часов не добавляются +тестами: такие ошибки остаются видимыми. Фоновые задачи завершаются или +отменяются при выходе из каждого сценария; ошибки воркеров не скрываются. + +`fixtures/client_run_expected.json` оставлен как исторический эталон старого +сценария. Новые тесты проверяют ответы протокола и состояние сервера, а не логи. diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e4c27a3 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,190 @@ +import asyncio +from copy import deepcopy +import importlib +import importlib.util +from pathlib import Path +from types import SimpleNamespace + +import pytest + + +@pytest.fixture(scope="session") +def client_module(): + """Загружает общий тестовый клиент без запуска его командного интерфейса.""" + path = Path(__file__).parents[1] / "src" / ".test" / "client_test.py" + spec = importlib.util.spec_from_file_location("artex_test_client", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class Clock: + """Монотонные часы, время которых продвигает сам тест.""" + + def __init__(self): + self.value = 1000.0 + + def __call__(self): + return self.value + + def advance(self, seconds): + assert seconds >= 0 + self.value += seconds + + +class SleepRequest: + def __init__(self, delay): + self.delay = delay + self.task = asyncio.current_task() + self.future = asyncio.get_running_loop().create_future() + + def release(self): + if not self.future.done(): + self.future.set_result(None) + + +class AsyncioProxy: + """Подменяет только ожидание времени в выбранном серверном модуле.""" + + def __init__(self, timing, label): + self.timing = timing + self.label = label + + def __getattr__(self, name): + return getattr(asyncio, name) + + def create_task(self, coroutine, **kwargs): + return self.timing.spawn(coroutine, **kwargs) + + async def sleep(self, delay, result=None): + request = SleepRequest(delay) + self.timing.queues[self.label].put_nowait(request) + await request.future + return result + + +class Timing: + def __init__(self): + self.clock = Clock() + self.tasks = [] + self.queues = {} + self.waiting = {} + + def install(self, module, monkeypatch, label): + self.queues[label] = asyncio.Queue() + self.waiting[label] = [] + monkeypatch.setattr(module, "asyncio", AsyncioProxy(self, label)) + # Do not invent a missing clock binding: that must remain a real error. + for name in ("now", "monotonic"): + if hasattr(module, name): + monkeypatch.setattr(module, name, self.clock) + + def spawn(self, coroutine, **kwargs): + task = asyncio.create_task(coroutine, **kwargs) + self.tasks.append(task) + return task + + async def sleeping(self, label, task=None): + async with asyncio.timeout(1): + while True: + for request in list(self.waiting[label]): + if request.future.done(): + self.waiting[label].remove(request) + elif task is None or request.task is task: + self.waiting[label].remove(request) + return request + self.waiting[label].append(await self.queues[label].get()) + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback): + for task in self.tasks: + if not task.done(): + task.cancel() + results = await asyncio.gather(*self.tasks, return_exceptions=True) + if exc_type is None: + for result in results: + if isinstance(result, BaseException) and not isinstance(result, asyncio.CancelledError): + raise result + + +class ConnectionRecorder: + def __init__(self): + self.sent = [] + self.close_count = 0 + + async def send(self, packet): + self.sent.append(deepcopy(packet)) + + async def close(self): + self.close_count += 1 + + +class RouterRecorder: + """Наблюдатель вызовов менеджера; сам Router проверяется отдельно.""" + + def __init__(self): + self.routes = {} + self.moves = [] + self.removed = [] + self.data = [] + + def add(self, uid): + self.routes.setdefault(uid, set()) + + def move(self, old_uid, new_uid): + self.moves.append((old_uid, new_uid)) + if old_uid in self.routes: + self.routes[new_uid] = self.routes.pop(old_uid) + + def remove(self, uid): + self.removed.append(uid) + self.routes.pop(uid, None) + + async def process(self, uid, command): + self.data.append((uid, command)) + + +@pytest.fixture +def timing(): + """Даёт управляемые часы и ожидания без реальных пауз в 5/60 секунд.""" + return Timing() + + +@pytest.fixture +def session_code(timing, monkeypatch): + """Загружает настоящую Session, подменяя только часы и ожидания.""" + module = importlib.import_module("session.session") + timing.install(module, monkeypatch, "session") + return module + + +@pytest.fixture +def make_session(session_code): + """Создаёт настоящую сессию с записью отправок и timeout-событий.""" + def create(uid=11): + connection = ConnectionRecorder() + events = [] + + async def on_timeout(source, event): + events.append((source, deepcopy(event))) + + session = session_code.Session(uid, connection, on_timeout) + return SimpleNamespace(session=session, connection=connection, events=events) + + return create + + +@pytest.fixture +def make_manager(session_code, timing, monkeypatch): + """Создаёт настоящий менеджер с наблюдателем маршрутов и управляемой очисткой.""" + module = importlib.import_module("session.manager") + timing.install(module, monkeypatch, "manager") + + def create(): + router = RouterRecorder() + manager = module.SessionManager(router) + return SimpleNamespace(manager=manager, router=router, connection=ConnectionRecorder) + + return create diff --git a/tests/fixtures/client_run_expected.json b/tests/fixtures/client_run_expected.json new file mode 100644 index 0000000..cd61817 --- /dev/null +++ b/tests/fixtures/client_run_expected.json @@ -0,0 +1,36 @@ +{ + "steps": [ + {"command": "connection", "state": "token"}, + {"command": "token_answer", "state": "ready"}, + {"command": "handshake_answer", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"}, + {"command": "telemetry", "state": "air"} + ], + "last_data": { + "telemetry": {"temperature": 42.0, "battery": 87} + }, + "server_responses": [], + "sessions_after_disconnect": 0, + "routes_after_disconnect": 0, + "server_output": [ + "[ROUTER] connected: ", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry", + "[ROUTER] -> others: telemetry" + ] +} diff --git a/tests/start_tests.bat b/tests/start_tests.bat new file mode 100644 index 0000000..51d68aa --- /dev/null +++ b/tests/start_tests.bat @@ -0,0 +1,19 @@ +@echo off +setlocal +cd /d "%~dp0.." + +set "PYTHON=D:\_Programs\anaconda\envs\sites\python.exe" +if not exist "%PYTHON%" ( + echo Python not found: %PYTHON% + pause + exit /b 1 +) + +if "%~1"=="" ( + "%PYTHON%" -m pytest "%~dp0." +) else ( + "%PYTHON%" -m pytest %* +) +set "EXIT_CODE=%ERRORLEVEL%" +pause +exit /b %EXIT_CODE% diff --git a/tests/test_client_protocol.py b/tests/test_client_protocol.py new file mode 100644 index 0000000..6398c49 --- /dev/null +++ b/tests/test_client_protocol.py @@ -0,0 +1,114 @@ +"""Check the real test client independently of the server implementation.""" + +import asyncio +import json + +import pytest + + +async def send_packet(writer, packet): + writer.write((json.dumps(packet) + "\n").encode()) + await writer.drain() + + +def test_client_waits_for_handshake_and_checks_heartbeat(client_module): + """Проверяет порядок handshake и сохранение данных, пришедших до heartbeat-ответа.""" + async def scenario(): + finished = asyncio.get_running_loop().create_future() + + async def peer(reader, writer): + try: + packet = json.loads(await reader.readline()) + assert packet == { + "command": "connection", + "payload": {"name": "rover", "type": "robot", "in": ["control"], "out": ["telemetry"]}, + } + await send_packet(writer, {"command": "token_ask", "payload": 123, "upf": 456}) + assert json.loads(await reader.readline()) == { + "command": "token_answer", "payload": {"token": None, "upf": None}, + } + await send_packet(writer, {"command": "handshake_ask"}) + assert json.loads(await reader.readline()) == { + "command": "handshake_answer", "payload": {"status": True, "state": None}, + } + assert json.loads(await reader.readline()) == {"command": "heartbeat_ask", "payload": {}} + await send_packet(writer, {"command": "control", "payload": {"turn": 1}}) + await send_packet(writer, {"command": "heartbeat_answer"}) + assert json.loads(await reader.readline()) == { + "command": "telemetry", "payload": {"temperature": 42.0, "battery": 87}, + } + assert await reader.read() == b"" + except BaseException as error: + finished.set_exception(error) + else: + finished.set_result(None) + finally: + writer.close() + await writer.wait_closed() + + server = await asyncio.start_server(peer, "127.0.0.1", 0) + async with server: + client = client_module.Client( + "rover", port=server.sockets[0].getsockname()[1], + node_type="robot", inputs=["control"], + ) + try: + await client.connect() + assert (client.uid, client.upf) == (123, 456) + assert await client.recv() == {"command": "control", "payload": {"turn": 1}} + await client.telemetry() + finally: + await client.close() + await finished + + asyncio.run(asyncio.wait_for(scenario(), 5)) + + +def test_client_times_out_when_server_does_not_answer(client_module): + """Проверяет таймаут молчащего сервера и освобождение клиентского соединения.""" + async def scenario(): + finished = asyncio.Event() + + async def peer(reader, writer): + try: + await reader.read() + finally: + writer.close() + await writer.wait_closed() + finished.set() + + server = await asyncio.start_server(peer, "127.0.0.1", 0) + async with server: + client = client_module.Client(port=server.sockets[0].getsockname()[1], timeout=0.1) + with pytest.raises(TimeoutError): + await client.connect() + assert client.writer is None + await finished.wait() + + asyncio.run(asyncio.wait_for(scenario(), 5)) + + +def test_client_reports_wrong_handshake_packet(client_module): + """Проверяет понятную ошибку и закрытие клиента при неверном ответе handshake.""" + async def scenario(): + finished = asyncio.Event() + + async def peer(reader, writer): + try: + await reader.readline() + await send_packet(writer, {"command": "handshake_ask"}) + await reader.read() + finally: + writer.close() + await writer.wait_closed() + finished.set() + + server = await asyncio.start_server(peer, "127.0.0.1", 0) + async with server: + client = client_module.Client(port=server.sockets[0].getsockname()[1]) + with pytest.raises(client_module.ProtocolError, match="Expected token_ask"): + await client.connect() + assert client.writer is None + await finished.wait() + + asyncio.run(asyncio.wait_for(scenario(), 5)) diff --git a/tests/test_client_run.py b/tests/test_client_run.py new file mode 100644 index 0000000..4eb149c --- /dev/null +++ b/tests/test_client_run.py @@ -0,0 +1,321 @@ +"""Integration checks using the same client as src/.test/start_client.bat.""" + +import asyncio +from copy import deepcopy + +import pytest + + +class RunningServer: + def __init__(self, client_module): + self.client_module = client_module + self.clients = [] + self.handlers = [] + self.watchers = [] + self.errors = [] + + async def __aenter__(self): + # Lazy imports keep client-only tests runnable if the server has an error. + from router import Router + from session import Sessions + from transport.tcp.handler import handle + + errors = self.errors + watchers = self.watchers + + class ObservedSessions(Sessions): + async def new(self, connection): + uid = await super().new(connection) + watchers.append(self[uid].watcher_task) + return uid + + async def feed(self, uid, packet): + try: + await super().feed(uid, packet) + except Exception as error: + # Preserve the original error before the handler enters + # its delayed cleanup and closes the client's transport. + errors.append(error) + raise + + self.router = Router() + self.sessions = ObservedSessions(self.router) + self.router.sessions = self.sessions + + def connected(reader, writer): + task = asyncio.create_task(handle(reader, writer, self.sessions)) + self.handlers.append(task) + task.add_done_callback(self.handler_finished) + + self.server = await asyncio.start_server(connected, "127.0.0.1", 0) + self.port = self.server.sockets[0].getsockname()[1] + return self + + def handler_finished(self, task): + if not task.cancelled() and task.exception() is not None: + self.errors.append(task.exception()) + + async def __aexit__(self, exc_type, exc, traceback): + self.server.close() + await self.server.wait_closed() + close_results = await asyncio.gather( + *(client.close() for client in self.clients), return_exceptions=True, + ) + # 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 + for task in tasks: + if not task.done(): + task.cancel() + results = await asyncio.gather(*tasks, return_exceptions=True) + 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: + for result in close_results: + if isinstance(result, BaseException): + raise result + + async def client( + self, name, inputs=("control",), outputs=("telemetry",), + node_type="test", **recovery, + ): + client = self.client_module.Client( + name, port=self.port, timeout=1, + inputs=inputs, outputs=outputs, node_type=node_type, + ) + self.clients.append(client) + try: + await client.connect(**recovery) + except (TimeoutError, self.client_module.ProtocolError) as error: + if self.errors: + raise self.errors[0] from error + raise + return client + + async def disconnect(self, client): + # A half-close proves that the SERVER closes the transport immediately. + client.writer.write_eof() + assert await asyncio.wait_for(client.reader.read(), 1) == b"" + await client.close() + assert self.sessions[client.uid].state == "stop" + assert client.uid in self.sessions.sessions + + +def test_server_connection_telemetry_heartbeat_and_disconnect(client_module): + """Проверяет handshake, метаданные, 10 пакетов, heartbeat и сохранение после EOF.""" + async def scenario(): + async with RunningServer(client_module) as server: + client = await server.client("rover") + session = server.sessions[client.uid] + assert session.state == "air" + assert session.upf == client.upf + assert session.static_meta["name"] == "rover" + assert session.static_meta["type"] == "test" + assert session.static_meta["in"] == {"control"} + assert session.static_meta["out"] == {"telemetry"} + assert session.dynamic_meta["active_in"] == set() + assert session.dynamic_meta["active_out"] == set() + assert client.uid in server.router.session_routes + + deadline = session.timeout + for index in range(10): + await client.telemetry(temperature=42.0 + index, battery=87 - index) + assert await client.heartbeat() == {"command": "heartbeat_answer"} + assert session.last_data == { + "telemetry": {"temperature": 42.0 + index, "battery": 87 - index}, + } + assert session.timeout >= deadline + deadline = session.timeout + # Rapid heartbeats may share one monotonic clock tick. Exact renewal + # is checked separately with a controlled clock. + + await server.disconnect(client) + assert session.last_data == {"telemetry": {"temperature": 51.0, "battery": 78}} + assert client.uid in server.router.session_routes + + asyncio.run(asyncio.wait_for(scenario(), 10)) + + +def test_logical_errors_disconnect_before_handshake(client_module): + """Проверяет закрытие TCP после пяти команд в неправильном состоянии протокола.""" + async def scenario(): + async with RunningServer(client_module) as server: + client = 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) + # Deliberately answer a handshake that the server has not requested. + for _ in range(5): + await client.send("handshake_answer", {"status": True, "state": None}) + assert await asyncio.wait_for(client.reader.read(), 1) == b"" + assert len(server.sessions.sessions) == 1 + session = next(iter(server.sessions.sessions.values())) + assert session.state == "stop" + + asyncio.run(asyncio.wait_for(scenario(), 5)) + + +def test_recovery_preserves_data_and_moves_both_route_directions(client_module): + """Проверяет восстановление данных и фактическую передачу по маршрутам в обе стороны.""" + async def scenario(): + async with RunningServer(client_module) as server: + old = await server.client("old-rover") + peer = await server.client( + "peer", inputs=["telemetry"], outputs=["control"], node_type="operator", + ) + # Route creation has no API yet; explicitly seed both directions. + server.router.session_routes[old.uid].add(peer.uid) + server.router.session_routes[peer.uid].add(old.uid) + await old.telemetry() + await old.heartbeat() + assert await peer.recv() == { + "command": "telemetry", "payload": {"temperature": 42.0, "battery": 87}, + } + server.sessions[old.uid].dynamic_meta = { + "active_in": {"control"}, "active_out": {"telemetry"}, + } + last_data = deepcopy(server.sessions[old.uid].last_data) + dynamic_meta = deepcopy(server.sessions[old.uid].dynamic_meta) + await server.disconnect(old) + + restored = await server.client("restored-rover", old_token=old.uid, old_upf=old.upf) + assert restored.uid != old.uid + assert server.sessions[restored.uid].state == "air" + assert server.sessions[restored.uid].last_data == last_data + assert server.sessions[restored.uid].dynamic_meta == dynamic_meta + assert old.uid not in server.router.session_routes + assert set(server.router.session_routes[restored.uid]) == {peer.uid} + assert set(server.router.session_routes[peer.uid]) == {restored.uid} + + await peer.send("control", {"turn": 1}) + await peer.heartbeat() + assert await restored.recv() == {"command": "control", "payload": {"turn": 1}} + await restored.telemetry(43.0, 86) + await restored.heartbeat() + assert await peer.recv() == { + "command": "telemetry", "payload": {"temperature": 43.0, "battery": 86}, + } + + asyncio.run(asyncio.wait_for(scenario(), 10)) + + +@pytest.mark.parametrize("reason", ["active", "wrong_upf", "missing"]) +def test_recovery_does_not_take_an_ineligible_session(client_module, reason): + """Проверяет отказ переноса для активной/отсутствующей сессии и неверного upf.""" + async def scenario(): + async with RunningServer(client_module) as server: + old = await server.client("old-rover") + await old.telemetry() + await old.heartbeat() + if reason != "active": + await server.disconnect(old) + uid = -1 if reason == "missing" else old.uid + upf = old.upf + 1 if reason == "wrong_upf" else old.upf + fresh = await server.client("fresh-rover", old_token=uid, old_upf=upf) + assert server.sessions[fresh.uid].last_data == {} + assert old.uid in server.router.session_routes + assert server.sessions[old.uid].state == ("air" if reason == "active" else "stop") + + asyncio.run(asyncio.wait_for(scenario(), 10)) + + +def test_old_recovery_proof_can_only_be_used_once(client_module): + """Проверяет, что повторный старый UID/upf не забирает уже восстановленные данные.""" + async def scenario(): + async with RunningServer(client_module) as server: + old = await server.client("old-rover") + await old.telemetry() + await old.heartbeat() + await server.disconnect(old) + restored = await server.client("restored-rover", old_token=old.uid, old_upf=old.upf) + expected = {"telemetry": {"temperature": 42.0, "battery": 87}} + assert server.sessions[restored.uid].last_data == expected + repeated = await server.client("repeated-rover", old_token=old.uid, old_upf=old.upf) + assert server.sessions[repeated.uid].last_data == {} + assert server.sessions[restored.uid].last_data == expected + assert restored.uid in server.router.session_routes + assert old.uid not in server.router.session_routes + + asyncio.run(asyncio.wait_for(scenario(), 10)) + + +async def expect_timeout_close(client, session): + read = asyncio.create_task(client.reader.read()) + try: + done, _ = await asyncio.wait( + {read, session.watcher_task}, timeout=1, + return_when=asyncio.FIRST_COMPLETED, + ) + assert done, "Воркер не обработал истечение дедлайна" + if session.watcher_task in done: + # Expose a crashed watcher rather than just a later socket timeout. + await session.watcher_task + assert await asyncio.wait_for(read, 1) == b"" + assert session.state == "stop" + finally: + if not read.done(): + read.cancel() + await asyncio.gather(read, return_exceptions=True) + + +@pytest.mark.parametrize("state", ["new", "token", "ready", "air"]) +@pytest.mark.usefixtures("make_manager") +def test_server_timeout_closes_transport_in_each_phase(client_module, timing, state): + """Проверяет закрытие настоящего TCP по таймауту на каждом этапе подключения.""" + async def scenario(): + async with timing, RunningServer(client_module) as server: + client = 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 not server.sessions.sessions: + await asyncio.sleep(0) + session = next(iter(server.sessions.sessions.values())) + if state in {"token", "ready", "air"}: + await client.send("connection", {"name": "idle-rover", "type": "test", "in": [], "out": ["telemetry"]}) + token = await client.recv() + assert token["command"] == "token_ask" + 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, "state": None}) + await client.heartbeat() + assert session.state == state + request = await timing.sleeping("session", session.watcher_task) + timing.clock.advance(session.timeout - timing.clock()) + request.release() + await expect_timeout_close(client, session) + assert server.sessions[session.uid] is session + + asyncio.run(asyncio.wait_for(scenario(), 5)) + + +@pytest.mark.usefixtures("make_manager") +def test_server_heartbeat_prevents_stale_deadline_disconnect(client_module, timing): + """Проверяет по TCP продление heartbeat и последующее отключение при тишине.""" + async def scenario(): + async with timing, RunningServer(client_module) as server: + client = await server.client("rover") + session = server.sessions[client.uid] + request = await timing.sleeping("session", session.watcher_task) + old_deadline = session.timeout + timing.clock.advance(4) + await client.heartbeat() + assert session.timeout > old_deadline + timing.clock.advance(old_deadline - timing.clock()) + request.release() + await asyncio.sleep(0) + if session.watcher_task.done(): + await session.watcher_task + assert session.state == "air" + await client.heartbeat() + request = await timing.sleeping("session", session.watcher_task) + timing.clock.advance(session.timeout - timing.clock()) + request.release() + await expect_timeout_close(client, session) + assert client.uid in server.sessions.sessions + + asyncio.run(asyncio.wait_for(scenario(), 5)) diff --git a/tests/test_router.py b/tests/test_router.py new file mode 100644 index 0000000..520e090 --- /dev/null +++ b/tests/test_router.py @@ -0,0 +1,92 @@ +"""Инварианты графа маршрутов и выбор полезной нагрузки при пересылке.""" + +import asyncio +from copy import deepcopy + +import pytest + + +@pytest.fixture +def router(): + """Создаёт настоящий Router независимо от менеджера и сессий.""" + from router import Router + return Router() + + +class Node: + def __init__(self, last_data=None): + self.last_data = last_data or {} + self.sent = [] + + def send(self, packet): + self.sent.append(deepcopy(packet)) + + +def test_add_creates_empty_routes_and_preserves_existing_edges(router): + """Проверяет создание множества маршрутов и сохранение связей при повторном add.""" + router.add(1) + assert router.session_routes[1] == set() + router.session_routes[1].add(2) + router.add(1) + assert router.session_routes[1] == {2} + + +def test_move_rewrites_only_edges_pointing_to_old_uid(router): + """Проверяет перенос исходящих/входящих связей без изменения несвязанных маршрутов.""" + router.session_routes = {1: {2}, 2: {1, 4}, 3: set(), 4: {2}, 5: set()} + router.move(1, 5) + assert router.session_routes == {2: {5, 4}, 3: set(), 4: {2}, 5: {2}} + + +def test_move_handles_self_route(router): + """Проверяет замену UID в маршруте, который указывал на собственную сессию.""" + router.session_routes = {1: {1, 2}, 2: {1}} + router.move(1, 3) + assert router.session_routes == {2: {3}, 3: {3, 2}} + + +def test_move_missing_uid_does_not_change_graph(router): + """Проверяет безопасный перенос отсутствующего UID без добавления новых связей.""" + router.session_routes = {1: {2}, 2: set()} + before = deepcopy(router.session_routes) + router.move(99, 100) + assert router.session_routes == before + + +def test_remove_clears_outgoing_and_incoming_edges(router): + """Проверяет удаление узла и всех ссылок на него после окончательной очистки.""" + router.session_routes = {1: {2}, 2: {1, 3}, 3: {1}, 4: set()} + router.remove(1) + assert router.session_routes == {2: {3}, 3: set(), 4: set()} + + +def test_remove_missing_uid_is_safe(router): + """Проверяет повторное удаление уже отсутствующего UID.""" + router.session_routes = {1: set()} + router.remove(99) + assert router.session_routes == {1: set()} + + +def test_process_delivers_source_payload_to_each_target(router): + """Проверяет fan-out, данные отправителя и отсутствие доставки несвязанным узлам.""" + nodes = { + 1: Node({"telemetry": {"battery": 87}}), + 2: Node({"telemetry": {"battery": 1}}), + 3: Node(), + 4: Node(), + } + router.sessions = nodes + router.session_routes = {1: {2, 3}, 2: set(), 3: set(), 4: set()} + asyncio.run(router.process(1, "telemetry")) + expected = [{"command": "telemetry", "payload": {"battery": 87}}] + assert nodes[2].sent == expected + assert nodes[3].sent == expected + assert nodes[1].sent == nodes[4].sent == [] + + +def test_process_with_no_routes_sends_nothing(router): + """Проверяет отсутствие доставки при пустом множестве получателей.""" + router.sessions = {1: Node({"telemetry": {"battery": 87}}), 2: Node()} + router.session_routes = {1: set(), 2: set()} + asyncio.run(router.process(1, "telemetry")) + assert router.sessions[1].sent == router.sessions[2].sent == [] diff --git a/tests/test_session.py b/tests/test_session.py new file mode 100644 index 0000000..59bda9e --- /dev/null +++ b/tests/test_session.py @@ -0,0 +1,239 @@ +"""Переходы протокола и жизненный цикл таймера отдельной сессии.""" + +import asyncio + +import pytest + + +def run(scenario): + asyncio.run(asyncio.wait_for(scenario(), 3)) + + +async def tick(request, session): + request.release() + await asyncio.sleep(0) + if session.watcher_task.done() and not session.watcher_task.cancelled(): + error = session.watcher_task.exception() + if error is not None: + raise error + + +def test_initial_deadline_and_watcher(make_session, timing): + """Проверяет начальное окно в 60 секунд и запуск собственного воркера сессии.""" + async def scenario(): + async with timing: + case = make_session() + assert case.session.state == "new" + assert case.session.timeout == timing.clock() + 60 + request = await timing.sleeping("session", case.session.watcher_task) + assert request.delay > 0 + assert case.events == [] + run(scenario) + + +@pytest.mark.parametrize("state", ["new", "token", "ready", "air"]) +def test_expired_deadline_emits_one_disconnect(make_session, timing, state): + """Проверяет однократное timeout-событие во всех незавершённых и активном состояниях.""" + async def scenario(): + async with timing: + case = make_session() + case.session.state = state + request = await timing.sleeping("session", case.session.watcher_task) + timing.clock.advance(60) + request.release() + await asyncio.wait_for(case.session.watcher_task, 1) + assert case.events == [(case.session.uid, {"type": "disconnect"})] + run(scenario) + + +def test_watcher_does_not_disconnect_before_deadline(make_session, timing): + """Проверяет, что воркер не отключает сессию раньше её дедлайна.""" + async def scenario(): + async with timing: + case = make_session() + request = await timing.sleeping("session", case.session.watcher_task) + timing.clock.advance(59) + await tick(request, case.session) + assert case.events == [] + assert not case.session.watcher_task.done() + await timing.sleeping("session", case.session.watcher_task) + run(scenario) + + +def test_heartbeat_extends_current_deadline(make_session, timing): + """Проверяет перенос дедлайна heartbeat и отсутствие отключения по старому сроку.""" + async def scenario(): + async with timing: + case = make_session() + case.session.state = "air" + request = await timing.sleeping("session", case.session.watcher_task) + timing.clock.advance(10) + await case.session.feed({"command": "heartbeat_ask", "payload": {}}) + await asyncio.sleep(0) + assert case.session.timeout == timing.clock() + 5 + assert case.connection.sent == [{"command": "heartbeat_answer"}] + assert case.session.last_data == {} + + timing.clock.advance(4) + await case.session.feed({"command": "heartbeat_ask", "payload": {}}) + new_deadline = case.session.timeout + timing.clock.advance(1) + await tick(request, case.session) + assert case.events == [] + request = await timing.sleeping("session", case.session.watcher_task) + timing.clock.advance(new_deadline - timing.clock()) + request.release() + await asyncio.wait_for(case.session.watcher_task, 1) + assert case.events == [(case.session.uid, {"type": "disconnect"})] + run(scenario) + + +@pytest.mark.parametrize("state", ["stop", "migrated"]) +def test_terminal_session_does_not_emit_timeout(make_session, timing, state): + """Проверяет завершение воркера без timeout для отключённой или перенесённой сессии.""" + async def scenario(): + async with timing: + case = make_session() + request = await timing.sleeping("session", case.session.watcher_task) + case.session.state = state + timing.clock.advance(100) + request.release() + await asyncio.wait_for(case.session.watcher_task, 1) + assert case.events == [] + run(scenario) + + +def test_stop_closes_transport_and_finishes_watcher(make_session, timing): + """Проверяет закрытие соединения, состояние stop и завершение таймерного воркера.""" + async def scenario(): + async with timing: + case = make_session() + request = await timing.sleeping("session", case.session.watcher_task) + await case.session.stop() + assert case.session.state == "stop" + assert case.connection.close_count == 1 + request.release() + await asyncio.wait_for(case.session.watcher_task, 1) + assert case.events == [] + run(scenario) + + +def test_connection_and_successful_handshake(make_session, timing): + """Проверяет new→token→ready→air, ответы сервера и метаданные портов.""" + async def scenario(): + async with timing: + case = make_session() + session = case.session + assert await session.feed({"command": "connection", "payload": { + "name": "rover", "type": "robot", "in": ["control"], "out": ["telemetry"], + }}) is None + assert session.state == "token" + await asyncio.sleep(0) + assert case.connection.sent == [{"command": "token_ask", "payload": session.uid, "upf": session.upf}] + assert session.static_meta["name"] == "rover" + assert session.static_meta["type"] == "robot" + assert session.static_meta["in"] == {"control"} + assert session.static_meta["out"] == {"telemetry"} + assert session.dynamic_meta == {"active_in": set(), "active_out": set()} + + assert await session.feed({"command": "token_answer", "payload": {"token": None, "upf": None}}) is None + assert session.state == "ready" + await asyncio.sleep(0) + assert case.connection.sent[-1] == {"command": "handshake_ask"} + assert await session.feed({"command": "handshake_answer", "payload": {"status": True, "state": None}}) == { + "type": "connect", "uid": session.uid, + } + assert session.state == "air" + run(scenario) + + +def test_recovery_request_still_requires_own_handshake(make_session, timing): + """Проверяет нормализацию старого UID/upf и ожидание отдельного нового handshake.""" + async def scenario(): + async with timing: + case = make_session() + case.session.state = "token" + event = await case.session.feed({"command": "token_answer", "payload": {"token": 123, "upf": 456}}) + assert event == {"type": "reconnect", "uid": 123, "upf": 456} + assert case.session.state == "ready" + event = await case.session.feed({"command": "handshake_answer", "payload": {"status": True}}) + assert event == {"type": "connect", "uid": case.session.uid} + assert case.session.state == "air" + run(scenario) + + +def test_data_updates_latest_value_for_each_command(make_session, timing): + """Проверяет нормализованные data-события и замену последнего значения по команде.""" + async def scenario(): + async with timing: + case = make_session() + case.session.state = "air" + for command, payload in [("telemetry", {"battery": 87}), ("status", {"ok": True}), ("telemetry", {"battery": 86})]: + assert await case.session.feed({"command": command, "payload": payload}) == {"type": "data", "data": command} + assert case.session.last_data == {"telemetry": {"battery": 86}, "status": {"ok": True}} + run(scenario) + + +@pytest.mark.parametrize("state", ["new", "token", "ready"]) +def test_wrong_sequence_disconnects_at_error_limit(make_session, timing, state): + """Проверяет четыре допустимых логических ошибки и disconnect на пятой.""" + async def scenario(): + async with timing: + case = make_session() + case.session.state = state + for _ in range(4): + assert await case.session.feed({"command": "telemetry", "payload": {}}) is None + assert await case.session.feed({"command": "telemetry", "payload": {}}) == {"type": "disconnect"} + run(scenario) + + +def test_missing_command_requests_disconnect(make_session, timing): + """Проверяет немедленное disconnect-событие у пакета без команды.""" + async def scenario(): + async with timing: + case = make_session() + assert await case.session.feed({"payload": {}}) == {"type": "disconnect"} + run(scenario) + + +def test_failed_handshake_waits_before_retry(make_session, timing): + """Проверяет реальное ожидание задержки перед повторным запросом handshake.""" + async def scenario(): + async with timing: + case = make_session() + case.session.state = "ready" + task = timing.spawn(case.session.feed({"command": "handshake_answer", "payload": {"status": False, "state": "booting"}})) + await asyncio.sleep(0) + assert not task.done(), "Повторный handshake выполняется без ожидания задержки" + request = await timing.sleeping("session", task) + assert request.delay == 1 + assert case.connection.sent == [] + request.release() + assert await task == {"type": "await handshake", "state": "booting"} + await asyncio.sleep(0) + assert case.connection.sent == [{"command": "handshake_ask"}] + assert case.session.state == "ready" + run(scenario) + + +def test_handshake_retry_limit_requests_disconnect(make_session, timing): + """Проверяет отказ от дальнейших повторов после пяти неудачных retry handshake.""" + async def scenario(): + async with timing: + case = make_session() + case.session.state = "ready" + case.session.timeout_attempt = 5 + assert await case.session.feed({"command": "handshake_answer", "payload": {"status": False}}) == {"type": "disconnect"} + run(scenario) + + +def test_stopped_session_does_not_send_packets(make_session, timing): + """Проверяет, что после stop новые отправки не поступают закрытому соединению.""" + async def scenario(): + async with timing: + case = make_session() + await case.session.stop() + case.session.send({"command": "telemetry", "payload": {}}) + await asyncio.sleep(0) + assert case.connection.sent == [] + run(scenario) diff --git a/tests/test_session_manager.py b/tests/test_session_manager.py new file mode 100644 index 0000000..4ca59b0 --- /dev/null +++ b/tests/test_session_manager.py @@ -0,0 +1,193 @@ +"""События менеджера, окно восстановления и гонки с отложенной очисткой.""" + +import asyncio +from copy import deepcopy + +import pytest + + +def run(scenario): + asyncio.run(asyncio.wait_for(scenario(), 3)) + + +async def new_session(case, state="air"): + connection = case.connection() + uid = await case.manager.new(connection) + case.manager[uid].state = state + if state == "air": + case.router.add(uid) + return uid, connection + + +async def finish_wait(request, allow_cancel=False): + request.release() + try: + await asyncio.wait_for(request.task, 1) + except asyncio.CancelledError: + if not allow_cancel: + raise + + +def test_new_connections_have_distinct_live_uids(make_manager, timing): + """Проверяет разные UID одновременно существующих соединений.""" + async def scenario(): + async with timing: + case = make_manager() + first, _ = await new_session(case) + second, _ = await new_session(case) + assert first != second + assert len(case.manager.sessions) == 2 + run(scenario) + + +def test_connect_and_data_events_reach_router(make_manager, timing): + """Проверяет регистрацию connect и передачу UID/команды data роутеру.""" + async def scenario(): + async with timing: + case = make_manager() + uid, _ = await new_session(case, "ready") + await case.manager.process(uid, {"type": "connect", "uid": uid}) + assert uid in case.router.routes + await case.manager.process(uid, {"type": "data", "data": "telemetry"}) + assert case.router.data == [(uid, "telemetry")] + run(scenario) + + +def test_disconnect_event_stops_transport_and_retains_record(make_manager, timing): + """Проверяет немедленное закрытие транспорта по событию disconnect и начало окна восстановления.""" + async def scenario(): + async with timing: + case = make_manager() + uid, connection = await new_session(case) + task = timing.spawn(case.manager.process(uid, {"type": "disconnect"})) + await asyncio.sleep(0) + assert connection.close_count == 1, "Менеджер не обработал событие disconnect" + assert case.manager[uid].state == "stop" + assert uid in case.manager.sessions + request = await timing.sleeping("manager") + assert request.delay == 60 + timing.clock.advance(60) + await finish_wait(request) + await asyncio.wait_for(task, 1) + # If cleanup is detached, wait for its observable effect, not an arbitrary pause. + async with asyncio.timeout(1): + while uid in case.manager.sessions: + await asyncio.sleep(0) + assert uid not in case.router.routes + run(scenario) + + +def test_close_preserves_session_until_recovery_window_expires(make_manager, timing): + """Проверяет сохранение сессии на 60 секунд и удаление записи/маршрутов после окна.""" + async def scenario(): + async with timing: + case = make_manager() + uid, connection = await new_session(case) + case.manager[uid].last_data = {"telemetry": {"battery": 87}} + task = timing.spawn(case.manager.close(uid)) + request = await timing.sleeping("manager") + assert request.delay == 60 + assert connection.close_count == 1 + assert case.manager[uid].state == "stop" + assert case.manager[uid].last_data == {"telemetry": {"battery": 87}} + assert uid in case.router.routes + timing.clock.advance(60) + await finish_wait(request) + await task + assert uid not in case.manager.sessions + assert uid not in case.router.routes + run(scenario) + + +@pytest.mark.parametrize("reason", ["active", "wrong_upf", "missing", "migrated"]) +def test_ineligible_recovery_does_not_move_data(make_manager, timing, reason): + """Проверяет отказ восстановления активной, отсутствующей, перенесённой сессии и неверного upf.""" + async def scenario(): + async with timing: + case = make_manager() + old_uid, _ = await new_session(case) + old = case.manager[old_uid] + old.state = "air" if reason == "active" else ("migrated" if reason == "migrated" else "stop") + old.last_data = {"telemetry": {"battery": 87}} + new_uid, _ = await new_session(case, "ready") + await case.manager.process(new_uid, { + "type": "reconnect", + "uid": -1 if reason == "missing" else old_uid, + "upf": old.upf + 1 if reason == "wrong_upf" else old.upf, + }) + assert case.router.moves == [] + assert case.manager[new_uid].last_data == {} + assert old.last_data == {"telemetry": {"battery": 87}} + assert case.manager[new_uid].state == "ready" + run(scenario) + + +def test_successful_recovery_preserves_data_and_is_consumed_once(make_manager, timing): + """Проверяет перенос динамических данных ровно один раз и сохранение собственного handshake.""" + async def scenario(): + async with timing: + case = make_manager() + old_uid, _ = await new_session(case) + old = case.manager[old_uid] + old.state = "stop" + old.last_data = {"telemetry": {"battery": 87}} + old.dynamic_meta = {"active_in": {"control"}, "active_out": {"telemetry"}} + expected_data = deepcopy(old.last_data) + expected_meta = deepcopy(old.dynamic_meta) + new_uid, _ = await new_session(case, "ready") + event = {"type": "reconnect", "uid": old_uid, "upf": old.upf} + await case.manager.process(new_uid, event) + assert case.manager[new_uid].last_data == expected_data + assert case.manager[new_uid].dynamic_meta == expected_meta + assert case.manager[new_uid].state == "ready" + assert case.router.moves == [(old_uid, new_uid)] + + repeated_uid, _ = await new_session(case, "ready") + await case.manager.process(repeated_uid, event) + assert case.manager[repeated_uid].last_data == {} + assert case.router.moves == [(old_uid, new_uid)] + assert case.manager[new_uid].last_data == expected_data + run(scenario) + + +def test_expired_session_cannot_be_recovered(make_manager, timing): + """Проверяет отказ восстановления после удаления старой сессии по 60-секундному таймеру.""" + async def scenario(): + async with timing: + case = make_manager() + old_uid, _ = await new_session(case) + old_upf = case.manager[old_uid].upf + task = timing.spawn(case.manager.close(old_uid)) + request = await timing.sleeping("manager") + timing.clock.advance(60) + await finish_wait(request) + await task + new_uid, _ = await new_session(case, "ready") + await case.manager.process(new_uid, {"type": "reconnect", "uid": old_uid, "upf": old_upf}) + assert case.router.moves == [] + assert case.manager[new_uid].last_data == {} + run(scenario) + + +def test_old_cleanup_cannot_delete_recovered_session(make_manager, timing): + """Проверяет, что старый таймер после восстановления не удаляет новую запись и её данные.""" + async def scenario(): + async with timing: + case = make_manager() + old_uid, _ = await new_session(case) + old = case.manager[old_uid] + old.last_data = {"telemetry": {"battery": 87}} + cleanup = timing.spawn(case.manager.close(old_uid)) + request = await timing.sleeping("manager") + new_uid, _ = await new_session(case, "ready") + restored = case.manager[new_uid] + await case.manager.process(new_uid, {"type": "reconnect", "uid": old_uid, "upf": old.upf}) + assert restored.last_data == {"telemetry": {"battery": 87}} + timing.clock.advance(60) + await finish_wait(request, allow_cancel=True) + if not cleanup.cancelled(): + await cleanup + assert case.manager[new_uid] is restored + assert restored.last_data == {"telemetry": {"battery": 87}} + assert new_uid in case.router.routes + run(scenario) diff --git a/tests/test_source_validation.py b/tests/test_source_validation.py new file mode 100644 index 0000000..dfaebc0 --- /dev/null +++ b/tests/test_source_validation.py @@ -0,0 +1,27 @@ +"""Проверки исходников, которые работают даже при ошибках импорта сервера.""" + +import ast +from pathlib import Path + + +def test_all_python_sources_compile(): + """Проверяет синтаксис всех серверных модулей и клиента без запуска и записи pyc.""" + errors = [] + for path in sorted((Path(__file__).parents[1] / "src").rglob("*.py")): + try: + compile(path.read_text(encoding="utf-8"), str(path), "exec") + except SyntaxError as error: + errors.append(f"{path.relative_to(Path(__file__).parents[1])}:{error.lineno}: {error.msg}") + assert not errors, "\n".join(errors) + + +def test_every_test_has_a_description(): + """Проверяет наличие краткого docstring у каждого теста набора.""" + missing = [] + for path in Path(__file__).parent.glob("test_*.py"): + tree = ast.parse(path.read_text(encoding="utf-8")) + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test_"): + if not ast.get_docstring(node): + missing.append(f"{path.name}:{node.lineno}: {node.name}") + assert not missing, "\n".join(missing)