File size: 4,194 Bytes
4879fc7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | from __future__ import annotations
from types import SimpleNamespace
import pytest
from fastapi.testclient import TestClient
from app.announce import reset_stamp_guard
from app.audit import AuditLogger
from app.config import Settings
from app.dedup import PromotionLRU
from app.deps import (
get_audit,
get_bucket_write_limiter,
get_dedup,
get_hub,
get_notifier,
get_org_roles,
get_raw_message_limiter,
get_read_model,
get_registration_limiter,
get_settings_dep,
get_verification_status,
get_verifier,
)
from app.main import app as fastapi_app
from app.notify import Notifier
from app.org_roles import OrgRoles
from app.rate_limit import CompoundLimiter, TokenBucket
from app.read_model import ReadModel
from app.verification import VerificationStatusStore
from app.verifier import Verifier
from fakes import FakeHub, FakeJobRunner
@pytest.fixture
def make_env():
"""Build an app environment around a FakeHub, with optional Settings
overrides (passed by env-var alias, e.g. MENTION_FANOUT_CAP=2)."""
def _make(**settings_overrides):
# The per-author monotonic stamp guard is process-global by design;
# without a reset, one test's frozen clock would leak into the next
# test's filenames.
reset_stamp_guard()
settings = Settings(
HF_TOKEN="test-token",
ORG="test-org",
COLLAB_SLUG="test",
AUDIT_BUCKET="auditor/test-audit",
VERIFIER_AGENT="test-verifier",
**settings_overrides,
)
hub = FakeHub(settings)
read_model = ReadModel(hub, settings)
# One LRU per env, like production's process-wide singleton — a
# per-request instance would make cross-request dedup untestable.
dedup = PromotionLRU(10_000)
verification = VerificationStatusStore(
hub, runs_prefix=settings.verification_runs_prefix
)
# Fresh registry per env so parked waiters can't leak across tests, and
# so the caps are whatever this test's Settings say.
notifier = Notifier(
max_waiters_per_owner=settings.longpoll_max_waiters_per_owner,
max_waiters_total=settings.longpoll_max_waiters_total,
wake_spread_s=settings.longpoll_wake_spread_s,
wake_spread_threshold=settings.longpoll_wake_spread_threshold,
)
runner = FakeJobRunner()
# Inline spawn: the verdict watcher runs synchronously inside the POST,
# so tests assert final state without thread coordination.
verifier = Verifier(
settings, hub, read_model, verification, runner,
spawn=lambda _name, fn: fn(),
notifier=notifier,
)
generous = lambda: CompoundLimiter( # noqa: E731 — tests never rate limit
TokenBucket(capacity=1000, refill_per_minute=1000),
TokenBucket(capacity=1000, refill_per_minute=1000),
)
fastapi_app.dependency_overrides.update(
{
get_settings_dep: lambda: settings,
get_hub: lambda: hub,
get_read_model: lambda: read_model,
get_notifier: lambda: notifier,
get_org_roles: lambda: OrgRoles(hub, settings),
get_audit: lambda: AuditLogger(hub),
get_dedup: lambda: dedup,
get_verification_status: lambda: verification,
get_verifier: lambda: verifier,
get_bucket_write_limiter: generous,
get_raw_message_limiter: generous,
get_registration_limiter: lambda: TokenBucket(
capacity=1000, refill_per_minute=1000
),
}
)
return SimpleNamespace(
settings=settings,
hub=hub,
read_model=read_model,
verification=verification,
runner=runner,
verifier=verifier,
notifier=notifier,
client=TestClient(fastapi_app),
)
yield _make
fastapi_app.dependency_overrides.clear()
@pytest.fixture
def env(make_env):
return make_env()
|