Presence & status model + endpoints (Redis heartbeat) -- recovered
checkpoint / checkpoint (push) Successful in 2m25s

Real feature built in August, checkpointed mid-task by the old fleet's
graceful-shutdown mechanism, then orphaned when checkpoint_focus_
feature_ids scoped the fleet away from M8 and it was never resumed --
the PM task had shown done/reviewed for a month with zero actual code on
main. Ported forward from the 225-commits-stale branch onto current
main's conventions (DbDep pattern, migration rebased onto 0005's head),
fixing two real bugs found in the process: the original PresenceStore's
explicit connect() was never called from anywhere (would 500 on every
real request), and a naive fix hits 'event loop is closed' under this
repo's TestClient(app) pattern -- fixed with lazy per-event-loop client
caching, same class of issue already documented for asyncpg in
conftest.py. conftest.py extended to also bring up ephemeral Redis
symmetrically with its existing Postgres bootstrap.

144/144 backend tests independently re-run via the fully automated
hermetic path (zero manual setup, matching what a fresh contributor
gets), plus a manual end-to-end heartbeat/TTL-expiry/batch-query check
against real Postgres+Redis.
This commit is contained in:
2026-09-15 00:01:24 +00:00
7 changed files with 623 additions and 38 deletions
@@ -0,0 +1,33 @@
"""add presence columns to users
Revision ID: 0006_add_presence_columns
Revises: 0005_add_world_content_tables
Create Date: 2026-09-14 00:00:00.000000
Ported forward from the orphaned agent/dd2a126b57a1 branch (originally
2a9b9873244a_add_presence_columns.py, down_revision 0001_initial) and
rebased onto the current migration head. See app/services/presence.py /
app/routers/presence.py.
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "0006_add_presence_columns"
down_revision: Union[str, None] = "0005_add_world_content_tables"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Add last_seen_at and status_message columns to users table."""
op.add_column("users", sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True))
op.add_column("users", sa.Column("status_message", sa.Text(), nullable=True))
def downgrade() -> None:
"""Remove presence columns from users table."""
op.drop_column("users", "status_message")
op.drop_column("users", "last_seen_at")
+2 -1
View File
@@ -6,7 +6,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings
from app.routers import auth, users, worlds, assets, avatar, world_content, internal
from app.routers import auth, users, worlds, assets, avatar, world_content, internal, presence
from app.services.game_servers import supervisor
@@ -44,6 +44,7 @@ app.include_router(world_content.router, prefix="/api/worlds", tags=["world-cont
app.include_router(assets.router, prefix="/api/assets", tags=["assets"])
app.include_router(avatar.router, prefix="/api/avatar", tags=["avatar"])
app.include_router(internal.router, prefix="/api/internal", tags=["internal"])
app.include_router(presence.router, prefix="/api/presence", tags=["presence"])
@app.get("/api/health")
+5 -1
View File
@@ -1,7 +1,7 @@
from datetime import date
from typing import Optional
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Date
from sqlalchemy import Column, Integer, String, Boolean, DateTime, Date, Text
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
@@ -20,6 +20,10 @@ class User(Base):
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Presence / status (backend/app/services/presence.py, app/routers/presence.py)
last_seen_at = Column(DateTime(timezone=True), nullable=True)
status_message = Column(Text, nullable=True)
sessions = relationship("Session", back_populates="user", cascade="all, delete-orphan")
refresh_tokens = relationship("RefreshToken", back_populates="user", cascade="all, delete-orphan")
worlds = relationship("World", back_populates="owner")
+146
View File
@@ -0,0 +1,146 @@
"""Presence & status router.
Endpoints:
- POST /api/presence/heartbeat -- record a heartbeat (sets Redis TTL key)
- GET /api/presence/presence -- query presence for a list of user ids
- PATCH /api/presence/status -- update the authenticated user's status message
"""
from datetime import datetime, timezone
from typing import Annotated
from fastapi import APIRouter, HTTPException, status, Depends
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.dependencies import CurrentUser
from app.models.user import User
from app.services.presence import presence_store
router = APIRouter()
DbDep = Annotated[AsyncSession, Depends(get_db)]
# ──────────────────────────── Pydantic schemas ────────────────────────────────
_STATUS_MAX_LEN = 140
class HeartbeatRequest(BaseModel):
"""Empty body -- the caller's identity comes from the JWT."""
class PresenceUser(BaseModel):
user_id: int
is_online: bool
last_seen_at: datetime | None
status_message: str | None
class PresenceResponse(BaseModel):
users: list[PresenceUser]
class StatusUpdate(BaseModel):
status_message: str | None = Field(default=None, max_length=_STATUS_MAX_LEN)
@field_validator("status_message")
@classmethod
def strip_message(cls, v: str | None) -> str | None:
if v is not None:
v = v.strip()
return v
# ──────────────────────────────── Endpoints ───────────────────────────────────
@router.post("/heartbeat", response_model=dict)
async def heartbeat(
_req: HeartbeatRequest,
current_user: CurrentUser,
db: DbDep,
):
"""Record a heartbeat for the authenticated user.
Sets a Redis key with a TTL so the server can derive online/offline
status without trusting client claims. Also persists last_seen_at to
the database.
"""
# Update Redis TTL (presence heartbeat)
await presence_store.heartbeat(current_user.id)
# Persist last_seen_at to the database
now = datetime.now(timezone.utc)
result = await db.execute(select(User).where(User.id == current_user.id))
user = result.scalar_one_or_none()
if user:
user.last_seen_at = now
await db.commit()
return {"status": "ok", "user_id": current_user.id}
@router.get("/presence", response_model=PresenceResponse)
async def get_presence(
user_ids: str,
current_user: CurrentUser,
db: DbDep,
):
"""Return presence info for a comma-separated list of user ids.
Online/offline is derived from the Redis heartbeat TTL.
last_seen_at and status_message come from the database.
"""
ids = [int(uid.strip()) for uid in user_ids.split(",") if uid.strip()]
if not ids:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="user_ids must contain at least one integer",
)
# Check Redis for online status (single round trip via MGET).
online_ids = await presence_store.is_online_batch(ids)
# Fetch DB records for last_seen_at and status_message
result = await db.execute(select(User).where(User.id.in_(ids)))
users = {u.id: u for u in result.scalars().all()}
presence_users = []
for uid in ids:
user = users.get(uid)
if user is None:
continue
presence_users.append(
PresenceUser(
user_id=uid,
is_online=uid in online_ids,
last_seen_at=user.last_seen_at,
status_message=user.status_message,
)
)
return PresenceResponse(users=presence_users)
@router.patch("/status", response_model=dict)
async def update_status(
body: StatusUpdate,
current_user: CurrentUser,
db: DbDep,
):
"""Update the authenticated user's status message (length-limited)."""
result = await db.execute(select(User).where(User.id == current_user.id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
user.status_message = body.status_message
await db.commit()
return {"status": "ok", "status_message": body.status_message}
+145
View File
@@ -0,0 +1,145 @@
"""Redis-backed presence store.
Presence is determined by a heartbeat TTL key in Redis:
- heartbeat(user_id) sets a TTL key per user (default 60 s).
- is_online(user_id) / is_online_batch(user_ids) check whether the key(s)
exist.
status_message and last_seen_at are persisted in the ``users`` table by the
router (app/routers/presence.py); this store only owns the Redis side.
The Redis client is created lazily, on first actual use, rather than
requiring an explicit ``connect()`` call during app startup. This matters
because FastAPI's lifespan handler does not run under the plain
``TestClient(app)`` instantiation this codebase's tests use (no
``with`` block), so anything that only connected during lifespan startup
would never connect in tests. Lazy connection means the store works
identically whether the app was started for real or is being driven by
pytest.
The client is also cached per-*event loop* rather than globally: the same
plain ``TestClient(app)`` pattern can dispatch different requests onto
different event loops (see the NullPool comment on ``_override_db`` in
tests/conftest.py for the analogous asyncpg problem), and an
``aioredis.Redis`` connection, like an asyncpg connection, breaks with
"Event loop is closed" if reused across loops. In the real running server
there is exactly one event loop for the process lifetime, so this
degrades to a single cached, pooled connection there.
"""
from __future__ import annotations
import asyncio
import logging
import redis.asyncio as aioredis
from app.config import settings
logger = logging.getLogger(__name__)
# Key prefix for heartbeat entries.
_PRESENCE_KEY_PREFIX = "presence:heartbeat:"
# Default heartbeat TTL in seconds (1 minute).
# Clients should heartbeat at least once per interval.
_DEFAULT_HEARTBEAT_TTL = 60
class PresenceStore:
"""Redis-backed presence store with async, lazily-connected operations."""
def __init__(self, redis_url: str | None = None, heartbeat_ttl: int = _DEFAULT_HEARTBEAT_TTL):
# redis_url is resolved from settings at first use (not at
# construction time) so tests that override settings.REDIS_URL
# before the first call still take effect -- this module is
# imported (and the module-level singleton below constructed)
# well before any such override runs.
self._redis_url = redis_url
self._heartbeat_ttl = heartbeat_ttl
# Keyed by id(running loop) -- see the module docstring.
self._clients: dict[int, aioredis.Redis] = {}
self._connect_lock = asyncio.Lock()
# ------------------------------------------------------------------
# Lifecycle
# ------------------------------------------------------------------
async def _ensure_client(self) -> aioredis.Redis:
"""Return a connected client for the current event loop, connecting
on first use per loop."""
loop_key = id(asyncio.get_running_loop())
client = self._clients.get(loop_key)
if client is None:
async with self._connect_lock:
client = self._clients.get(loop_key)
if client is None:
url = self._redis_url or settings.REDIS_URL
client = aioredis.from_url(
url,
encoding="utf-8",
decode_responses=True,
)
try:
await client.ping()
except Exception:
logger.warning("Redis ping failed for %s", url)
raise
self._clients[loop_key] = client
return client
async def close(self) -> None:
"""Close the Redis connection for the current event loop, if one was
opened."""
loop_key = id(asyncio.get_running_loop())
client = self._clients.pop(loop_key, None)
if client is not None:
await client.aclose()
# ------------------------------------------------------------------
# Heartbeat
# ------------------------------------------------------------------
async def heartbeat(self, user_id: int) -> None:
"""Record a heartbeat for *user_id*.
Sets a TTL key so that the user is considered online as long as
heartbeats arrive at least once per heartbeat_ttl seconds.
"""
client = await self._ensure_client()
await client.set(self._heartbeat_key(user_id), "1", ex=self._heartbeat_ttl)
# ------------------------------------------------------------------
# Online checks
# ------------------------------------------------------------------
async def is_online(self, user_id: int) -> bool:
"""Return True if the user has a valid heartbeat key."""
client = await self._ensure_client()
exists = await client.exists(self._heartbeat_key(user_id))
return bool(exists)
async def is_online_batch(self, user_ids: list[int]) -> set[int]:
"""Return the subset of user_ids that currently have a valid
heartbeat key. Uses MGET (one round trip) rather than one EXISTS
per user.
"""
if not user_ids:
return set()
client = await self._ensure_client()
keys = [self._heartbeat_key(uid) for uid in user_ids]
values = await client.mget(keys)
return {uid for uid, v in zip(user_ids, values) if v is not None}
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _heartbeat_key(user_id: int) -> str:
return f"{_PRESENCE_KEY_PREFIX}{user_id}"
# Singleton instance shared by the router, so all requests share the same
# connection pool.
presence_store = PresenceStore()
+71 -36
View File
@@ -1,15 +1,18 @@
"""Shared pytest fixtures + hermetic-DB bootstrap.
"""Shared pytest fixtures + hermetic-DB/Redis bootstrap.
The integration tests hit a real Postgres (Postgres-specific types: JSONB,
native ENUMs). To make ``cd backend && pytest`` work from a clean checkout
with no manual setup, ``pytest_configure`` below brings up an ephemeral
Postgres via docker compose and runs the migrations IF ``DATABASE_URL`` is
not already pointing at a reachable database. When it is (dev loop, or the
checkpoint runner which stands the DB up itself), that path is skipped and
the existing database is reused.
native ENUMs) and, for presence tests, a real Redis (TTL semantics aren't
worth faking). To make ``cd backend && pytest`` work from a clean checkout
with no manual setup, ``pytest_configure`` below brings up ephemeral
Postgres/Redis containers via docker compose and runs the migrations IF
``DATABASE_URL``/``REDIS_URL`` aren't already pointing at something
reachable. When they are (dev loop, or the checkpoint runner which stands
these up itself), that path is skipped and the existing services are
reused.
If Docker isn't available and no DB is reachable, the whole DB-backed suite
is skipped with a clear message rather than erroring out 60+ times.
If Docker isn't available and nothing is reachable, the whole
DB/Redis-backed suite is skipped with a clear message rather than erroring
out dozens of times.
"""
import asyncio
@@ -44,8 +47,9 @@ TEST_TABLES = (
BACKEND_DIR = Path(__file__).resolve().parents[1]
_COMPOSE_PROJECT = "agp-pytest"
_EPHEMERAL_PG_PORT = 25432
_EPHEMERAL_REDIS_PORT = 26379
# Set by _ensure_database() when it starts a container it owns; torn down in
# Set by pytest_configure() when it starts container(s) it owns; torn down in
# pytest_unconfigure.
_started_container = False
_db_unavailable_reason: str | None = None
@@ -64,6 +68,11 @@ def _db_reachable(url: str) -> bool:
return _port_open(parts.hostname or "localhost", parts.port or 5432)
def _redis_reachable(url: str) -> bool:
parts = urlsplit(url)
return _port_open(parts.hostname or "localhost", parts.port or 6379)
def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
return subprocess.run(cmd, cwd=BACKEND_DIR, capture_output=True, text=True, **kw)
@@ -71,14 +80,17 @@ def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
def pytest_configure(config):
global _started_container, _db_unavailable_reason
if _db_reachable(settings.DATABASE_URL):
return # dev loop / checkpoint runner already provided one
need_db = not _db_reachable(settings.DATABASE_URL)
need_redis = not _redis_reachable(settings.REDIS_URL)
if not need_db and not need_redis:
return # dev loop / checkpoint runner already provided both
if not _run(["docker", "--version"]).returncode == 0:
_db_unavailable_reason = (
"no reachable DATABASE_URL and docker is unavailable -- run the "
"checkpoint suite (scripts/run_checkpoint_tests.py) or start Postgres "
"manually"
"no reachable DATABASE_URL/REDIS_URL and docker is unavailable -- run "
"the checkpoint suite (scripts/run_checkpoint_tests.py) or start "
"Postgres/Redis manually"
)
return
@@ -86,33 +98,55 @@ def pytest_configure(config):
env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT
env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT)
env["POSTGRES_PORT"] = str(_EPHEMERAL_PG_PORT)
up = _run(["docker", "compose", "up", "-d", "postgres"], env=env)
env["REDIS_HOST_PORT"] = str(_EPHEMERAL_REDIS_PORT)
services = [s for s, need in (("postgres", need_db), ("redis", need_redis)) if need]
up = _run(["docker", "compose", "up", "-d", *services], env=env)
if up.returncode != 0:
_db_unavailable_reason = f"docker compose up postgres failed:\n{up.stderr[-800:]}"
_db_unavailable_reason = f"docker compose up {' '.join(services)} failed:\n{up.stderr[-800:]}"
return
_started_container = True
for _ in range(30):
r = _run(
["docker", "compose", "exec", "-T", "postgres",
"pg_isready", "-U", "agp_user", "-d", "agp_db"],
env=env,
)
if r.returncode == 0:
break
time.sleep(1)
else:
_db_unavailable_reason = "ephemeral Postgres never became ready"
return
if need_db:
for _ in range(30):
r = _run(
["docker", "compose", "exec", "-T", "postgres",
"pg_isready", "-U", "agp_user", "-d", "agp_db"],
env=env,
)
if r.returncode == 0:
break
time.sleep(1)
else:
_db_unavailable_reason = "ephemeral Postgres never became ready"
return
url = f"postgresql+asyncpg://agp_user:agp_password@localhost:{_EPHEMERAL_PG_PORT}/agp_db"
os.environ["DATABASE_URL"] = url
settings.DATABASE_URL = url
url = f"postgresql+asyncpg://agp_user:agp_password@localhost:{_EPHEMERAL_PG_PORT}/agp_db"
os.environ["DATABASE_URL"] = url
settings.DATABASE_URL = url
mig = _run([sys.executable, "-m", "alembic", "upgrade", "head"],
env={**os.environ, "DATABASE_URL": url})
if mig.returncode != 0:
_db_unavailable_reason = f"alembic upgrade head failed:\n{mig.stderr[-800:]}"
mig = _run([sys.executable, "-m", "alembic", "upgrade", "head"],
env={**os.environ, "DATABASE_URL": url})
if mig.returncode != 0:
_db_unavailable_reason = f"alembic upgrade head failed:\n{mig.stderr[-800:]}"
return
if need_redis:
for _ in range(30):
r = _run(
["docker", "compose", "exec", "-T", "redis", "redis-cli", "ping"],
env=env,
)
if r.returncode == 0 and "PONG" in r.stdout:
break
time.sleep(1)
else:
_db_unavailable_reason = "ephemeral Redis never became ready"
return
redis_url = f"redis://localhost:{_EPHEMERAL_REDIS_PORT}/0"
os.environ["REDIS_URL"] = redis_url
settings.REDIS_URL = redis_url
def pytest_unconfigure(config):
@@ -122,6 +156,7 @@ def pytest_unconfigure(config):
env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT
env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT)
env["POSTGRES_PORT"] = str(_EPHEMERAL_PG_PORT)
env["REDIS_HOST_PORT"] = str(_EPHEMERAL_REDIS_PORT)
_run(["docker", "compose", "down", "-v"], env=env)
+221
View File
@@ -0,0 +1,221 @@
"""Integration tests for the presence feature:
- POST /api/presence/heartbeat
- GET /api/presence/presence
- PATCH /api/presence/status
Hits the real app against the configured Postgres *and* a real Redis
(conftest.py brings up an ephemeral instance of each when neither is
already reachable) -- presence is fundamentally "does a TTL key exist in
Redis", which isn't worth faking with a mock.
Also covers the regression fixed in 47ab492 ("Fix StatusUpdate field
annotation and is_online batch-call mismatch"): the router used to call
``PresenceStore.is_online(ids)`` with a *list* of ids and then evaluate
``uid in online_ids`` against the returned bool, which raised a TypeError
on the first real multi-id batch call, and ``StatusUpdate.status_message``
used the invalid annotation ``str | Field(...)``. Both are exercised below.
"""
import asyncio
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.services.presence import PresenceStore
def _register(client: TestClient, username: str) -> dict:
resp = client.post(
"/api/auth/register",
json={
"username": username,
"email": f"{username}@example.com",
"password": "secret123",
"date_of_birth": "1990-01-01",
},
)
assert resp.status_code == 201, resp.text
token = resp.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
me = client.get("/api/users/me", headers=headers)
assert me.status_code == 200, me.text
return {"headers": headers, "id": me.json()["id"]}
class TestHeartbeat:
def test_heartbeat_sets_online(self):
client = TestClient(app)
user = _register(client, "presence_hb1")
resp = client.post("/api/presence/heartbeat", json={}, headers=user["headers"])
assert resp.status_code == 200, resp.text
assert resp.json() == {"status": "ok", "user_id": user["id"]}
presence = client.get(
"/api/presence/presence",
params={"user_ids": str(user["id"])},
headers=user["headers"],
)
assert presence.status_code == 200, presence.text
body = presence.json()["users"]
assert len(body) == 1
assert body[0]["user_id"] == user["id"]
assert body[0]["is_online"] is True
assert body[0]["last_seen_at"] is not None
def test_requires_auth(self):
client = TestClient(app)
resp = client.post("/api/presence/heartbeat", json={})
assert resp.status_code == 401
def test_never_seen_user_is_offline_with_null_last_seen(self):
client = TestClient(app)
user = _register(client, "presence_never_hb")
presence = client.get(
"/api/presence/presence",
params={"user_ids": str(user["id"])},
headers=user["headers"],
)
assert presence.status_code == 200
body = presence.json()["users"][0]
assert body["is_online"] is False
assert body["last_seen_at"] is None
class TestBatchPresence:
"""Covers the batch-call path fixed in 47ab492 (is_online_batch via MGET)."""
def test_batch_query_mixes_online_offline_and_unknown_ids(self):
client = TestClient(app)
online_user = _register(client, "presence_batch_online")
offline_user = _register(client, "presence_batch_offline")
client.post("/api/presence/heartbeat", json={}, headers=online_user["headers"])
unknown_id = max(online_user["id"], offline_user["id"]) + 999_999
ids = f"{online_user['id']},{offline_user['id']},{unknown_id}"
resp = client.get(
"/api/presence/presence",
params={"user_ids": ids},
headers=online_user["headers"],
)
assert resp.status_code == 200, resp.text
by_id = {u["user_id"]: u for u in resp.json()["users"]}
# Unknown id is silently dropped (not in the DB), not a 500 or 404.
assert unknown_id not in by_id
assert by_id[online_user["id"]]["is_online"] is True
assert by_id[offline_user["id"]]["is_online"] is False
def test_empty_user_ids_is_a_400(self):
client = TestClient(app)
user = _register(client, "presence_empty_ids")
resp = client.get(
"/api/presence/presence",
params={"user_ids": ""},
headers=user["headers"],
)
assert resp.status_code == 400
class TestStatusMessage:
def test_update_and_read_back_status_message(self):
client = TestClient(app)
user = _register(client, "presence_status1")
resp = client.patch(
"/api/presence/status",
json={"status_message": " brb, making coffee "},
headers=user["headers"],
)
assert resp.status_code == 200, resp.text
# The router strips whitespace (StatusUpdate.strip_message).
assert resp.json()["status_message"] == "brb, making coffee"
presence = client.get(
"/api/presence/presence",
params={"user_ids": str(user["id"])},
headers=user["headers"],
)
assert presence.json()["users"][0]["status_message"] == "brb, making coffee"
def test_status_message_over_max_length_is_rejected(self):
client = TestClient(app)
user = _register(client, "presence_status_toolong")
resp = client.patch(
"/api/presence/status",
json={"status_message": "x" * 141},
headers=user["headers"],
)
# StatusUpdate.status_message: str | None = Field(..., max_length=140)
# This is exactly the annotation 47ab492 fixed (it used to be the
# invalid `str | Field(...)`, which breaks model construction
# entirely rather than validating length).
assert resp.status_code == 422
def test_status_message_at_max_length_is_accepted(self):
client = TestClient(app)
user = _register(client, "presence_status_maxlen")
resp = client.patch(
"/api/presence/status",
json={"status_message": "x" * 140},
headers=user["headers"],
)
assert resp.status_code == 200, resp.text
def test_null_status_message_clears_it(self):
client = TestClient(app)
user = _register(client, "presence_status_clear")
client.patch("/api/presence/status", json={"status_message": "hi"}, headers=user["headers"])
resp = client.patch("/api/presence/status", json={"status_message": None}, headers=user["headers"])
assert resp.status_code == 200
assert resp.json()["status_message"] is None
class TestPresenceStoreTTL:
"""Exercises the real Redis TTL semantics directly against the service,
with a short heartbeat_ttl so the test doesn't need to sleep for the
real 60s default used in production. This is the cheap, real proxy for
"TTL expiry means offline" the task asked for -- same code path
(real Redis SET ... EX / EXISTS / MGET), just a much shorter window.
"""
async def test_ttl_expiry_flips_online_to_offline(self):
store = PresenceStore(heartbeat_ttl=1)
try:
uid = 987654321 # doesn't need to exist in Postgres; Redis-only check
assert await store.is_online(uid) is False
await store.heartbeat(uid)
assert await store.is_online(uid) is True
await asyncio.sleep(1.5)
assert await store.is_online(uid) is False
finally:
await store.close()
async def test_is_online_batch_matches_is_online_semantics(self):
"""Regression coverage for 47ab492: is_online_batch must return a
set of ids usable with `in`, not a bool."""
store = PresenceStore(heartbeat_ttl=30)
try:
online_uid, offline_uid = 111222333, 444555666
await store.heartbeat(online_uid)
result = await store.is_online_batch([online_uid, offline_uid])
assert isinstance(result, set)
assert online_uid in result
assert offline_uid not in result
finally:
await store.close()
async def test_is_online_batch_empty_list(self):
store = PresenceStore()
try:
assert await store.is_online_batch([]) == set()
finally:
await store.close()