Codex Release Bot
69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
"""Tests for Redis-backed BT Panel bootstrap lock."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from server.infrastructure.btpanel.bootstrap_lock import BootstrapLockTimeout, bootstrap_lock
|
|
|
|
|
|
class _FakeRedis:
|
|
def __init__(self) -> None:
|
|
self.data: dict[str, str] = {}
|
|
|
|
async def set(self, key: str, value: str, ex: int | None = None, nx: bool = False):
|
|
if nx and key in self.data:
|
|
return False
|
|
self.data[key] = value
|
|
return True
|
|
|
|
async def eval(self, _script: str, _numkeys: int, key: str, token: str):
|
|
if self.data.get(key) == token:
|
|
del self.data[key]
|
|
return 1
|
|
return 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bootstrap_lock_releases_after_success(monkeypatch):
|
|
redis = _FakeRedis()
|
|
monkeypatch.setattr("server.infrastructure.btpanel.bootstrap_lock.get_redis", lambda: redis)
|
|
|
|
async with bootstrap_lock(12, wait_timeout_seconds=0):
|
|
assert "btpanel:bootstrap:lock:12" in redis.data
|
|
|
|
assert redis.data.get("btpanel:bootstrap:lock:12") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bootstrap_lock_timeout_when_already_locked(monkeypatch):
|
|
redis = _FakeRedis()
|
|
redis.data["btpanel:bootstrap:lock:12"] = "other-token"
|
|
monkeypatch.setattr("server.infrastructure.btpanel.bootstrap_lock.get_redis", lambda: redis)
|
|
|
|
with pytest.raises(BootstrapLockTimeout):
|
|
async with bootstrap_lock(12, wait_timeout_seconds=0):
|
|
pass
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bootstrap_lock_waits_for_release(monkeypatch):
|
|
redis = _FakeRedis()
|
|
monkeypatch.setattr("server.infrastructure.btpanel.bootstrap_lock.get_redis", lambda: redis)
|
|
events: list[str] = []
|
|
|
|
async def first():
|
|
async with bootstrap_lock(12, wait_timeout_seconds=0):
|
|
events.append("first-enter")
|
|
await asyncio.sleep(0.05)
|
|
events.append("first-exit")
|
|
|
|
async def second():
|
|
await asyncio.sleep(0.01)
|
|
async with bootstrap_lock(12, wait_timeout_seconds=1):
|
|
events.append("second-enter")
|
|
|
|
await asyncio.gather(first(), second())
|
|
|
|
assert events == ["first-enter", "first-exit", "second-enter"]
|