From a0640e2e26ecd5c3a9e133149f8b6be6e8d23995 Mon Sep 17 00:00:00 2001 From: mamahoos Date: Sat, 8 Aug 2026 13:48:13 +0330 Subject: [PATCH] test: add real Redis integration suite and CI service Run live Redis tests against redis:7 in Actions, keep fakeredis for units, and persist storage markers before aborting overwritten waits. --- .github/workflows/redis-integration.yml | 42 ++++++++ .github/workflows/test.yml | 3 +- pyproject.toml | 3 + src/aiogram_input/session.py | 8 +- tests/integration/__init__.py | 0 tests/integration/conftest.py | 34 +++++++ tests/integration/test_redis_live.py | 127 ++++++++++++++++++++++++ 7 files changed, 213 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/redis-integration.yml create mode 100644 tests/integration/__init__.py create mode 100644 tests/integration/conftest.py create mode 100644 tests/integration/test_redis_live.py diff --git a/.github/workflows/redis-integration.yml b/.github/workflows/redis-integration.yml new file mode 100644 index 0000000..9e88c16 --- /dev/null +++ b/.github/workflows/redis-integration.yml @@ -0,0 +1,42 @@ +name: Redis Integration + +on: + push: + branches: [main] + pull_request: + branches: [main] + +concurrency: + group: redis-integration-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + redis-integration: + runs-on: ubuntu-latest + services: + redis: + image: redis:7-alpine + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 3s + --health-retries 10 + + steps: + - uses: actions/checkout@v4 + + - name: Install uv + uses: astral-sh/setup-uv@v6 + with: + python-version: '3.12' + enable-cache: true + + - name: Sync dependencies + run: uv sync --group dev --frozen + + - name: Run Redis integration tests + env: + REDIS_URL: redis://127.0.0.1:6379/15 + run: uv run pytest -q tests/integration -m integration diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 9fea4b0..1eda2f6 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -30,9 +30,10 @@ jobs: - name: Sync dependencies run: uv sync --group dev --frozen - - name: Run tests with coverage + - name: Run unit tests with coverage run: > uv run pytest -q + -m "not integration" --cov=aiogram_input --cov-report=term-missing --cov-fail-under=95 diff --git a/pyproject.toml b/pyproject.toml index cb7ea13..f0ea9b9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,6 +71,9 @@ testpaths = ["tests"] pythonpath = ["src", "."] asyncio_mode = "auto" addopts = ["--strict-markers", "--strict-config"] +markers = [ + "integration: tests that need a real Redis server", +] [tool.coverage.run] source = ["aiogram_input"] diff --git a/src/aiogram_input/session.py b/src/aiogram_input/session.py index 1501c1f..34bd4ec 100644 --- a/src/aiogram_input/session.py +++ b/src/aiogram_input/session.py @@ -135,12 +135,14 @@ async def _register_pending( chat_id, PendingWait(wait_id=wait_id, future=future, filter=filter), ) - if previous is not None: - logger.debug("[SESSION] Overwriting existing pending entry chat=%s", chat_id) - self._registry.cancel_wait(previous, chat_id=chat_id) + # Persist the new marker before aborting the previous wait so a racing + # cleanup cannot delete the replacement key from Redis/Memory. await self._storage.set( chat_id, WaitRecord(wait_id=wait_id, created_at=time.time()) ) + if previous is not None: + logger.debug("[SESSION] Overwriting existing pending entry chat=%s", chat_id) + self._registry.cancel_wait(previous, chat_id=chat_id) async def _cleanup(self, chat_id: int, wait_id: str) -> None: wait = await self._registry.pop_if(chat_id, wait_id) diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 0000000..1f31ebd --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import os +import uuid + +import pytest +from redis.asyncio import Redis + +from aiogram_input import RedisInputStorage + + +@pytest.fixture +def redis_url() -> str: + return os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/15") + + +@pytest.fixture +async def redis(redis_url: str): + client = Redis.from_url(redis_url, decode_responses=True) + await client.ping() + prefix = f"aiogram_input:it:{uuid.uuid4().hex}:" + try: + yield client, prefix + finally: + keys = [key async for key in client.scan_iter(match=f"{prefix}*")] + if keys: + await client.delete(*keys) + await client.aclose() + + +@pytest.fixture +async def storage(redis) -> RedisInputStorage: + client, prefix = redis + return RedisInputStorage(client, key_prefix=prefix) diff --git a/tests/integration/test_redis_live.py b/tests/integration/test_redis_live.py new file mode 100644 index 0000000..3b369d0 --- /dev/null +++ b/tests/integration/test_redis_live.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import asyncio + +import pytest +from aiogram import Dispatcher +from redis.asyncio import Redis + +from aiogram_input import InputWaiter, RedisInputStorage, setup_input +from aiogram_input.registry import WaitRegistry +from aiogram_input.session import SessionManager +from aiogram_input.types import WaitRecord +from tests.helpers import make_message + +pytestmark = pytest.mark.integration + + +@pytest.mark.asyncio +async def test_live_redis_roundtrip(storage: RedisInputStorage) -> None: + record = WaitRecord(wait_id="live-1", created_at=10.0) + await storage.set(1001, record) + assert await storage.contains(1001) is True + assert await storage.get(1001) == record + assert await storage.pop(1001) == record + assert await storage.contains(1001) is False + + +@pytest.mark.asyncio +async def test_live_redis_pop_if_atomic(storage: RedisInputStorage) -> None: + await storage.set(1002, WaitRecord(wait_id="keep", created_at=11.0)) + assert await storage.pop_if(1002, "nope") is None + assert await storage.contains(1002) is True + popped = await storage.pop_if(1002, "keep") + assert popped is not None + assert popped.wait_id == "keep" + assert await storage.contains(1002) is False + + +@pytest.mark.asyncio +async def test_live_redis_ttl(redis) -> None: + client, prefix = redis + storage = RedisInputStorage(client, key_prefix=prefix, ttl=2) + await storage.set(1003, WaitRecord(wait_id="ttl", created_at=1.0)) + ttl = await client.ttl(f"{prefix}1003") + assert 0 < ttl <= 2 + await asyncio.sleep(2.2) + assert await storage.contains(1003) is False + + +@pytest.mark.asyncio +async def test_live_session_wait_feed_with_redis(storage: RedisInputStorage) -> None: + session = SessionManager(storage, WaitRegistry()) + chat_id = 4242 + task = asyncio.create_task(session.start_waiting(chat_id, timeout=2, filter=None)) + await asyncio.sleep(0) + assert await storage.contains(chat_id) is True + + msg = make_message(chat_id, message_id=7) + assert await session.feed(msg) is True + assert await task is msg + assert await storage.contains(chat_id) is False + + +@pytest.mark.asyncio +async def test_live_overwrite_cleans_redis_marker(storage: RedisInputStorage) -> None: + session = SessionManager(storage, WaitRegistry()) + chat_id = 5151 + first = asyncio.create_task(session.start_waiting(chat_id, timeout=3, filter=None)) + for _ in range(50): + if await storage.contains(chat_id): + break + await asyncio.sleep(0.01) + else: + raise AssertionError("first wait never registered in Redis") + + second = asyncio.create_task(session.start_waiting(chat_id, timeout=3, filter=None)) + assert await first is None + + for _ in range(50): + if await storage.contains(chat_id): + break + await asyncio.sleep(0.01) + else: + raise AssertionError("second wait missing from Redis after overwrite") + + msg = make_message(chat_id, message_id=9) + assert await session.feed(msg) is True + assert await second is msg + assert await storage.contains(chat_id) is False + + +@pytest.mark.asyncio +async def test_live_setup_input_and_waiter(redis) -> None: + client, prefix = redis + dp = Dispatcher() + storage = RedisInputStorage(client, key_prefix=prefix) + waiter = setup_input(dp, storage=storage, data_key="aiogram_input") + assert isinstance(waiter, InputWaiter) + + task = asyncio.create_task(waiter.wait(6060, timeout=2)) + await asyncio.sleep(0) + assert await storage.contains(6060) is True + + middleware = dp.message.outer_middleware._middlewares[0] + msg = make_message(6060) + + async def handler(event, data): + return "should-not-run" + + assert await middleware(handler, msg, {}) is None + assert await task is msg + assert await storage.contains(6060) is False + + +@pytest.mark.asyncio +async def test_live_two_connections_share_marker(redis_url: str, redis) -> None: + client, prefix = redis + other = Redis.from_url(redis_url, decode_responses=True) + try: + writer = RedisInputStorage(client, key_prefix=prefix) + reader = RedisInputStorage(other, key_prefix=prefix) + await writer.set(7070, WaitRecord(wait_id="shared", created_at=1.0)) + assert await reader.get(7070) == WaitRecord(wait_id="shared", created_at=1.0) + assert await reader.pop_if(7070, "shared") is not None + assert await writer.contains(7070) is False + finally: + await other.aclose()