core/tests/conftest.py

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] / "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)
    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.expected_errors = []
        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

    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:
                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):
                    if not any(result is expected for expected in self.expected_errors):
                        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