From 1e8ffa15d0c74758f3ce095ba7cd86f454159bdc Mon Sep 17 00:00:00 2001 From: 3252a8 <3252a8@proton.me> Date: Sun, 17 May 2026 00:31:12 +0300 Subject: [PATCH] chore: add tests for redis infra, webhook queue and split-arch entrypoints --- tests/test_admin_sync_queue.py | 121 +++++++++++++ tests/test_app_logging.py | 75 ++++++++ tests/test_panel_webhook_routing.py | 141 +++++++++++++++ tests/test_startup_banner.py | 102 ++++++++--- tests/test_webhook_queue.py | 255 ++++++++++++++++++++++++++++ 5 files changed, 675 insertions(+), 19 deletions(-) create mode 100644 tests/test_admin_sync_queue.py create mode 100644 tests/test_app_logging.py create mode 100644 tests/test_panel_webhook_routing.py create mode 100644 tests/test_webhook_queue.py diff --git a/tests/test_admin_sync_queue.py b/tests/test_admin_sync_queue.py new file mode 100644 index 0000000..323f009 --- /dev/null +++ b/tests/test_admin_sync_queue.py @@ -0,0 +1,121 @@ +"""Verifies that the admin sync endpoint hands work to the worker via Redis. + +After the container split, /api/admin/sync no longer runs ``perform_sync`` +in-process; it enqueues a ``panel_sync`` event onto the webhook queue and +returns a fast ack. These tests pin that contract. +""" + +import json +import unittest +from types import SimpleNamespace +from typing import Any, List +from unittest.mock import patch + +from aiohttp import web + +# Importing the facade populates each admin_api_impl submodule's globals with +# helpers like ``_require_admin_user_id`` and ``_ok``/``_error``. Without this +# side effect, ``sync_module._require_admin_user_id`` does not exist yet. +import bot.app.web.subscription_webapp # noqa: F401 +from bot.app.web.admin_api_impl import sync as sync_module + + +class _FakeRequest: + """The shape ``admin_sync_route`` reads from the aiohttp request.""" + + def __init__(self, settings: SimpleNamespace, admin_telegram_id: int = 42) -> None: + self.app = {"settings": settings} + self._store = {"admin_telegram_id": admin_telegram_id} + + def get(self, key: str, default: Any = None) -> Any: + return self._store.get(key, default) + + +def _make_settings() -> SimpleNamespace: + return SimpleNamespace( + ADMIN_IDS=[42], + REDIS_KEY_PREFIX="shop", + WEBHOOK_QUEUE_NAME="webhook-events", + REDIS_URL="redis://r:6379/0", + ) + + +def _parse(response: web.Response) -> dict: + return json.loads(response.body.decode()) + + +def _patch_admin_auth(monkeypatch_target: Any) -> None: + """``_require_admin_user_id`` looks up session state we don't have here.""" + monkeypatch_target.side_effect = lambda request: int(request.get("admin_telegram_id")) + + +class AdminSyncQueueTests(unittest.IsolatedAsyncioTestCase): + async def test_returns_queued_when_redis_accepts_event(self): + recorded: List[dict] = [] + + async def fake_enqueue(settings, provider, payload, *, event_id=None): + recorded.append( + { + "settings": settings, + "provider": provider, + "payload": payload, + "event_id": event_id, + } + ) + return True + + with ( + patch.object(sync_module, "enqueue_webhook_event", fake_enqueue), + patch.object(sync_module, "_require_admin_user_id") as auth, + ): + _patch_admin_auth(auth) + response = await sync_module.admin_sync_route(_FakeRequest(_make_settings())) + + self.assertEqual(response.status, 200) + body = _parse(response) + self.assertTrue(body["ok"]) + self.assertEqual(body["result"], {"status": "queued"}) + + self.assertEqual(len(recorded), 1) + entry = recorded[0] + self.assertEqual(entry["provider"], "panel_sync") + self.assertEqual(entry["payload"], {"requested_by": 42}) + # event_id=None lets Redis enqueue every admin request (no dedupe key). + self.assertIsNone(entry["event_id"]) + + async def test_returns_503_when_queue_is_unavailable(self): + async def fake_enqueue(settings, provider, payload, *, event_id=None): + return False + + with ( + patch.object(sync_module, "enqueue_webhook_event", fake_enqueue), + patch.object(sync_module, "_require_admin_user_id") as auth, + ): + _patch_admin_auth(auth) + response = await sync_module.admin_sync_route(_FakeRequest(_make_settings())) + + self.assertEqual(response.status, 503) + body = _parse(response) + self.assertFalse(body["ok"]) + self.assertEqual(body["error"], "queue_unavailable") + + async def test_non_admin_is_rejected_before_enqueueing(self): + async def fake_enqueue(*args, **kwargs): + raise AssertionError("enqueue must not be called for non-admin requests") + + def deny(_request): + raise web.HTTPForbidden( + text=json.dumps({"ok": False, "error": "forbidden"}), + content_type="application/json", + ) + + with ( + patch.object(sync_module, "enqueue_webhook_event", fake_enqueue), + patch.object(sync_module, "_require_admin_user_id", side_effect=deny), + ): + with self.assertRaises(web.HTTPForbidden): + await sync_module.admin_sync_route(_FakeRequest(_make_settings())) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/test_app_logging.py b/tests/test_app_logging.py new file mode 100644 index 0000000..fc9a1da --- /dev/null +++ b/tests/test_app_logging.py @@ -0,0 +1,75 @@ +import io +import logging +import os +import sys +import unittest +from unittest.mock import patch + +import app_logging + + +def _reset_root_logger() -> None: + root = logging.getLogger() + for handler in list(root.handlers): + root.removeHandler(handler) + handler.close() + root.setLevel(logging.WARNING) + + +class ConfigureLoggingTests(unittest.TestCase): + def setUp(self) -> None: + _reset_root_logger() + self.addCleanup(_reset_root_logger) + + def test_default_level_is_info(self): + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("LOG_LEVEL", None) + app_logging.configure_logging() + self.assertEqual(logging.getLogger().level, logging.INFO) + + def test_log_level_env_is_honored(self): + with patch.dict(os.environ, {"LOG_LEVEL": "DEBUG"}, clear=False): + app_logging.configure_logging() + self.assertEqual(logging.getLogger().level, logging.DEBUG) + + def test_invalid_log_level_falls_back_to_info(self): + with patch.dict(os.environ, {"LOG_LEVEL": "TOTALLY_BOGUS"}, clear=False): + app_logging.configure_logging() + self.assertEqual(logging.getLogger().level, logging.INFO) + + def test_handler_writes_to_stdout_with_expected_format(self): + # basicConfig is a no-op if root already has handlers — start clean. + with patch.dict(os.environ, {"LOG_LEVEL": "INFO"}, clear=False): + app_logging.configure_logging() + + # First handler is the one we installed (basicConfig adds at most one). + handlers = logging.getLogger().handlers + self.assertTrue(handlers, "configure_logging() must install a handler") + handler = handlers[0] + self.assertIs(handler.stream, sys.stdout) + + # Confirm the format string structure by emitting a record into a buffer. + formatter = handler.formatter + record = logging.LogRecord( + name="test", level=logging.INFO, pathname=__file__, lineno=1, + msg="hello", args=(), exc_info=None, + ) + rendered = formatter.format(record) if formatter else "" + # Format is "%(asctime)s - %(name)s - %(levelname)s - %(message)s". + self.assertIn(" - test - INFO - hello", rendered) + + def test_configure_logging_does_not_swallow_subsequent_logs(self): + buffer = io.StringIO() + # Reset state, then replace stdout briefly so we can capture handler output. + with patch.dict(os.environ, {"LOG_LEVEL": "INFO"}, clear=False): + with patch.object(sys, "stdout", buffer): + app_logging.configure_logging() + logging.getLogger("test.module").info("payload-marker") + for handler in logging.getLogger().handlers: + handler.flush() + self.assertIn("payload-marker", buffer.getvalue()) + self.assertIn("INFO", buffer.getvalue()) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/test_panel_webhook_routing.py b/tests/test_panel_webhook_routing.py new file mode 100644 index 0000000..5b12f27 --- /dev/null +++ b/tests/test_panel_webhook_routing.py @@ -0,0 +1,141 @@ +"""Behaviour of ``PanelWebhookService.handle_webhook``. + +The service must: + * reject requests without the configured shared secret or a missing/bad + signature header — otherwise an attacker could forge panel events; + * acknowledge an event with HTTP 200 once it is on the queue (the worker + container does the heavy lifting); + * fall back to in-process background dispatch when Redis is unreachable so + a single-node deploy still processes events. +""" + +import asyncio +import hashlib +import hmac +import json +import unittest +from types import SimpleNamespace +from typing import Any, List +from unittest.mock import patch + +from bot.services import panel_webhook_service as pws + +SECRET = "panel-shared-secret" + + +def _sign(body: bytes) -> str: + return hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest() + + +def _make_service(): + settings = SimpleNamespace( + PANEL_WEBHOOK_SECRET=SECRET, + SUBSCRIPTION_NOTIFICATIONS_ENABLED=True, + SUBSCRIPTION_NOTIFY_DAYS_BEFORE=3, + SUBSCRIPTION_NOTIFY_ON_EXPIRE=True, + SUBSCRIPTION_NOTIFY_AFTER_EXPIRE=True, + DEFAULT_LANGUAGE="ru", + email_auth_configured=False, + ) + # Build minimally with object() placeholders — handle_webhook does not + # touch the bot/i18n/db/panel collaborators on its enqueue path. + return pws.PanelWebhookService( + bot=object(), + settings=settings, + i18n=object(), + async_session_factory=object(), + panel_service=object(), + ) + + +class HandleWebhookSecurityTests(unittest.IsolatedAsyncioTestCase): + async def test_unauthorized_when_secret_not_configured(self): + service = _make_service() + service.settings.PANEL_WEBHOOK_SECRET = "" + response = await service.handle_webhook(b"{}", _sign(b"{}")) + self.assertEqual(response.status, 401) + + async def test_unauthorized_when_signature_header_missing(self): + service = _make_service() + response = await service.handle_webhook(b'{"name":"user.expired"}', None) + self.assertEqual(response.status, 401) + + async def test_unauthorized_on_signature_mismatch(self): + service = _make_service() + body = b'{"name":"user.expired"}' + response = await service.handle_webhook(body, "deadbeef") + self.assertEqual(response.status, 401) + + async def test_bad_request_for_invalid_json(self): + service = _make_service() + body = b"not-json" + response = await service.handle_webhook(body, _sign(body)) + self.assertEqual(response.status, 400) + + async def test_ok_no_event_when_name_missing(self): + service = _make_service() + body = json.dumps({"payload": {"telegramId": 1}}).encode() + response = await service.handle_webhook(body, _sign(body)) + self.assertEqual(response.status, 200) + self.assertEqual(response.text, "ok_no_event") + + +class HandleWebhookQueueingTests(unittest.IsolatedAsyncioTestCase): + async def test_enqueues_to_redis_and_returns_ok(self): + service = _make_service() + captured: List[dict] = [] + + async def fake_enqueue(settings, provider, payload, *, event_id=None): + captured.append( + {"provider": provider, "payload": payload, "event_id": event_id} + ) + return True + + body = json.dumps( + { + "name": "user.expires_in_24_hours", + "payload": {"telegramId": 99, "uuid": "abc"}, + } + ).encode() + + with patch.object(pws, "enqueue_webhook_event", fake_enqueue): + response = await service.handle_webhook(body, _sign(body)) + + self.assertEqual(response.status, 200) + self.assertEqual(response.text, "ok") + self.assertEqual(len(captured), 1) + entry = captured[0] + self.assertEqual(entry["provider"], "panel") + self.assertEqual(entry["payload"]["event"], "user.expires_in_24_hours") + self.assertEqual(entry["payload"]["user"], {"telegramId": 99, "uuid": "abc"}) + # event_id combines event name with the strongest available identifier. + self.assertEqual(entry["event_id"], "user.expires_in_24_hours:99") + + async def test_falls_back_to_background_task_when_redis_unavailable(self): + service = _make_service() + background_seen: List[Any] = [] + + async def fake_enqueue(*args, **kwargs): + return False + + async def fake_handle_event(event_name, user_payload): + background_seen.append((event_name, user_payload)) + + body = json.dumps( + {"name": "user.expired", "payload": {"telegramId": 7}} + ).encode() + + with ( + patch.object(pws, "enqueue_webhook_event", fake_enqueue), + patch.object(service, "handle_event", fake_handle_event), + ): + response = await service.handle_webhook(body, _sign(body)) + # Yield to the event loop so the scheduled background task runs. + await asyncio.sleep(0) + + self.assertEqual(response.status, 200) + self.assertEqual(background_seen, [("user.expired", {"telegramId": 7})]) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/test_startup_banner.py b/tests/test_startup_banner.py index e37ba68..fd1d980 100644 --- a/tests/test_startup_banner.py +++ b/tests/test_startup_banner.py @@ -4,33 +4,97 @@ import unittest from contextlib import redirect_stdout from unittest.mock import patch -from startup_banner import print_startup_banner +from startup_banner import _redis_target, print_startup_banner + + +def _render(service: str, env: dict) -> str: + buffer = io.StringIO() + with patch.dict(os.environ, env, clear=False), redirect_stdout(buffer): + print_startup_banner(service) + return buffer.getvalue() class StartupBannerTests(unittest.TestCase): def test_startup_banner_marks_services_without_mojibake(self): for service in ("frontend", "backend", "worker", "migrate"): with self.subTest(service=service): - buffer = io.StringIO() - - with ( - patch.dict( - os.environ, - { - "IMAGE_TAG": "test-tag", - "POSTGRES_HOST": "postgres", - "POSTGRES_DB": "postgres", - "REDIS_URL": "redis://redis:6379/0", - }, - clear=False, - ), - redirect_stdout(buffer), - ): - print_startup_banner(service) - - output = buffer.getvalue() + output = _render( + service, + { + "IMAGE_TAG": "test-tag", + "POSTGRES_HOST": "postgres", + "POSTGRES_DB": "postgres", + "REDIS_URL": "redis://redis:6379/0", + }, + ) self.assertIn(f"container :: {service.upper()}", output) self.assertIn("image tag :: test-tag", output) self.assertIn("remnawave-minishop", output) self.assertIn("███", output) self.assertNotIn("в", output) + + +class StartupBannerServiceDetailsTests(unittest.TestCase): + def test_backend_lists_ports_postgres_and_redis(self): + output = _render( + "backend", + { + "IMAGE_TAG": "v1", + "WEB_SERVER_PORT": "8090", + "WEBAPP_SERVER_PORT": "8091", + "WEBAPP_ENABLED": "True", + "POSTGRES_HOST": "pg", + "POSTGRES_PORT": "5440", + "POSTGRES_DB": "shop", + "REDIS_URL": "redis://r:6380/2", + }, + ) + self.assertIn("webhooks :: :8090", output) + self.assertIn("webapp api :: on / :8091", output) + self.assertIn("postgres :: pg:5440/shop", output) + self.assertIn("redis :: r:6380/2", output) + + def test_worker_lists_queue_and_panel_sync_interval(self): + output = _render( + "worker", + { + "IMAGE_TAG": "v1", + "WEBHOOK_QUEUE_CONCURRENCY": "8", + "WORKER_PANEL_SYNC_INTERVAL_SECONDS": "600", + "TARIFFS_CONFIG_PATH": "data/tariffs.json", + "POSTGRES_HOST": "pg", + "POSTGRES_DB": "shop", + "REDIS_URL": "redis://r:6379/0", + }, + ) + self.assertIn("queue concurrency :: 8", output) + self.assertIn("panel sync interval :: 600s", output) + self.assertIn("tariffs config :: data/tariffs.json", output) + + def test_migrate_shows_one_shot_mode_and_data_dir(self): + output = _render( + "migrate", + { + "IMAGE_TAG": "v1", + "POSTGRES_HOST": "pg", + "POSTGRES_DB": "shop", + }, + ) + self.assertIn("mode :: one-shot migrations", output) + self.assertIn("data dir :: /app/data", output) + + +class StartupBannerRedisTargetTests(unittest.TestCase): + def test_redis_target_renders_host_port_db(self): + with patch.dict(os.environ, {"REDIS_URL": "redis://r:6379/3"}, clear=False): + self.assertEqual(_redis_target(), "r:6379/3") + + def test_redis_target_defaults_db_to_zero(self): + with patch.dict(os.environ, {"REDIS_URL": "redis://r"}, clear=False): + self.assertEqual(_redis_target(), "r/0") + + def test_redis_target_when_url_missing(self): + env = dict(os.environ) + env.pop("REDIS_URL", None) + with patch.dict(os.environ, env, clear=True): + self.assertEqual(_redis_target(), "-") diff --git a/tests/test_webhook_queue.py b/tests/test_webhook_queue.py new file mode 100644 index 0000000..8f23c76 --- /dev/null +++ b/tests/test_webhook_queue.py @@ -0,0 +1,255 @@ +import asyncio +import json +import time +import unittest +from types import SimpleNamespace +from typing import Any, Dict, List, Optional, Tuple +from unittest.mock import patch + +from bot.infra import redis as redis_infra +from bot.infra import webhook_queue + + +class FakeRedis: + """Minimal async stand-in for redis.asyncio.Redis used by tests.""" + + def __init__(self) -> None: + self._kv: Dict[str, str] = {} + self._ttl: Dict[str, float] = {} + self._lists: Dict[str, List[str]] = {} + + async def set( + self, + key: str, + value: str, + *, + nx: bool = False, + ex: Optional[int] = None, + ) -> bool: + self._expire_if_due(key) + if nx and key in self._kv: + return False + self._kv[key] = value + if ex is not None: + self._ttl[key] = time.monotonic() + ex + else: + self._ttl.pop(key, None) + return True + + async def get(self, key: str) -> Optional[str]: + self._expire_if_due(key) + return self._kv.get(key) + + async def delete(self, *keys: str) -> int: + removed = 0 + for key in keys: + if key in self._kv: + self._kv.pop(key, None) + self._ttl.pop(key, None) + removed += 1 + if key in self._lists: + self._lists.pop(key, None) + removed += 1 + return removed + + async def lpush(self, key: str, *values: str) -> int: + bucket = self._lists.setdefault(key, []) + for value in values: + bucket.insert(0, value) + return len(bucket) + + async def brpop( + self, key: str, timeout: int = 0 + ) -> Optional[Tuple[str, str]]: + bucket = self._lists.get(key) + if bucket: + return key, bucket.pop() + # In real Redis, brpop blocks. Tests never use the timeout path. + return None + + async def llen(self, key: str) -> int: + return len(self._lists.get(key, [])) + + async def eval(self, script: str, numkeys: int, *args: Any) -> int: + # The only Lua used by the code base releases redis_lock atomically. + keys = args[:numkeys] + argv = args[numkeys:] + if keys and argv and self._kv.get(keys[0]) == argv[0]: + self._kv.pop(keys[0], None) + self._ttl.pop(keys[0], None) + return 1 + return 0 + + async def aclose(self) -> None: # pragma: no cover - close path + self._kv.clear() + self._ttl.clear() + self._lists.clear() + + def _expire_if_due(self, key: str) -> None: + expires_at = self._ttl.get(key) + if expires_at is not None and time.monotonic() >= expires_at: + self._kv.pop(key, None) + self._ttl.pop(key, None) + + +def _make_settings(**overrides: Any) -> SimpleNamespace: + base: Dict[str, Any] = { + "REDIS_URL": "redis://redis:6379/0", + "REDIS_KEY_PREFIX": "remnawave-tg-shop", + "WEBHOOK_QUEUE_NAME": "webhook-events", + } + base.update(overrides) + return SimpleNamespace(**base) + + +class RedisKeyTests(unittest.TestCase): + def test_redis_key_joins_prefix_and_parts(self): + settings = _make_settings(REDIS_KEY_PREFIX="shop") + self.assertEqual(redis_infra.redis_key(settings, "queue", "events"), "shop:queue:events") + + def test_redis_key_filters_empty_or_colon_only_parts(self): + settings = _make_settings(REDIS_KEY_PREFIX="shop") + # Empty / colon-only parts are dropped; numeric and string parts coexist. + self.assertEqual( + redis_infra.redis_key(settings, "lock", "", ":", "panel-sync", 42), + "shop:lock:panel-sync:42", + ) + + def test_redis_key_strips_leading_trailing_colons(self): + settings = _make_settings(REDIS_KEY_PREFIX=":shop:") + self.assertEqual( + redis_infra.redis_key(settings, ":webhook:", "seen"), + "shop:webhook:seen", + ) + + +class WebhookQueueTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.fake = FakeRedis() + + async def fake_get_redis(_settings): + return self.fake + + self._patcher = patch.object(webhook_queue, "get_redis", fake_get_redis) + self._patcher.start() + self.addCleanup(self._patcher.stop) + + async def test_enqueue_returns_false_when_redis_unavailable(self): + async def no_redis(_settings): + return None + + with patch.object(webhook_queue, "get_redis", no_redis): + ok = await webhook_queue.enqueue_webhook_event( + _make_settings(), "yookassa", {"id": "p_1"} + ) + self.assertFalse(ok) + + async def test_enqueue_writes_payload_and_increments_depth(self): + settings = _make_settings() + ok = await webhook_queue.enqueue_webhook_event( + settings, + "yookassa", + {"id": "p_42", "amount": "100"}, + event_id="payment.succeeded:p_42", + ) + self.assertTrue(ok) + + depth = await webhook_queue.webhook_queue_depth(settings) + self.assertEqual(depth, 1) + + popped = await webhook_queue.pop_webhook_event(settings) + assert popped is not None + self.assertEqual(popped["provider"], "yookassa") + self.assertEqual(popped["event_id"], "payment.succeeded:p_42") + self.assertEqual(popped["payload"], {"id": "p_42", "amount": "100"}) + self.assertIsInstance(popped["enqueued_at"], (int, float)) + + async def test_enqueue_dedupes_repeated_event_ids(self): + settings = _make_settings() + first = await webhook_queue.enqueue_webhook_event( + settings, + "panel", + {"event": "user.expired", "user": {"telegramId": 1}}, + event_id="user.expired:1", + ) + second = await webhook_queue.enqueue_webhook_event( + settings, + "panel", + {"event": "user.expired", "user": {"telegramId": 1}}, + event_id="user.expired:1", + ) + self.assertTrue(first) + # Dedupe path is treated as a success — the event was already accepted. + self.assertTrue(second) + self.assertEqual(await webhook_queue.webhook_queue_depth(settings), 1) + + async def test_pop_returns_none_when_queue_empty(self): + settings = _make_settings() + self.assertIsNone(await webhook_queue.pop_webhook_event(settings)) + + async def test_pop_skips_invalid_payload(self): + settings = _make_settings() + await self.fake.lpush(webhook_queue.webhook_queue_key(settings), "not-json") + result = await webhook_queue.pop_webhook_event(settings) + self.assertIsNone(result) + + async def test_queue_key_uses_prefix_and_queue_name(self): + settings = _make_settings(REDIS_KEY_PREFIX="shop", WEBHOOK_QUEUE_NAME="custom-q") + self.assertEqual(webhook_queue.webhook_queue_key(settings), "shop:queue:custom-q") + + async def test_enqueue_serializes_unicode_without_escaping(self): + settings = _make_settings() + await webhook_queue.enqueue_webhook_event( + settings, "panel", {"name": "Юзер", "id": 7}, event_id="u:7" + ) + # Raw payload in Redis preserves Cyrillic without \uXXXX escapes. + raw = self.fake._lists[webhook_queue.webhook_queue_key(settings)][0] + self.assertIn("Юзер", raw) + parsed = json.loads(raw) + self.assertEqual(parsed["payload"]["name"], "Юзер") + + +class RedisLockTests(unittest.IsolatedAsyncioTestCase): + async def test_lock_is_noop_when_redis_unavailable(self): + async def no_redis(_settings): + return None + + with patch.object(redis_infra, "get_redis", no_redis): + async with redis_infra.redis_lock( + _make_settings(), "panel-sync", ttl_seconds=30 + ) as acquired: + self.assertTrue(acquired) + + async def test_lock_excludes_concurrent_holders_and_releases_on_exit(self): + fake = FakeRedis() + + async def fake_get_redis(_settings): + return fake + + with patch.object(redis_infra, "get_redis", fake_get_redis): + settings = _make_settings() + async with redis_infra.redis_lock(settings, "panel-sync", ttl_seconds=30) as first: + self.assertTrue(first) + async with redis_infra.redis_lock( + settings, "panel-sync", ttl_seconds=30 + ) as second: + self.assertFalse(second) + # After exit, the lock is released and can be re-acquired. + async with redis_infra.redis_lock(settings, "panel-sync", ttl_seconds=30) as again: + self.assertTrue(again) + + +class SleepOrStopTests(unittest.IsolatedAsyncioTestCase): + async def test_returns_promptly_when_event_set(self): + event = asyncio.Event() + event.set() + await redis_infra.sleep_or_stop(event, seconds=5) # would hang on a bug + + async def test_times_out_when_event_not_set(self): + event = asyncio.Event() + # Tiny timeout keeps the test fast; the function should not raise on timeout. + await redis_infra.sleep_or_stop(event, seconds=0.01) + + +if __name__ == "__main__": # pragma: no cover + unittest.main()