浅念
f938118fee
LoginLimiter._fails 与 rate_limit._buckets 只增不删——懒清理只覆盖被 回访的键，扫号换 IP/设备轮换产生的键永不被回访 → 进程长跑缓慢累积。 两处各加 sweep()（全表清「全部时间戳均已出窗」的键，窗沿 <= 与既有 保留判据 t > now - W 同沿），main.lifespan 起周期任务（300s loop） 启动创建/关停取消（cancel 后 await 落地，不留 pending task）。 Co-Authored-By: Claude Code <noreply@anthropic.com>
96 lines
4.1 KiB
Python
96 lines
4.1 KiB
Python
"""limiter 周期清扫(spec 2026-10-19 §7——日志/内存上限铁律的进程内面)。
|
|
|
|
两处限流内存态只增不删的兜底:
|
|
- app.platform.auth.LoginLimiter._fails: (username, ip) → list[float]——
|
|
allow/record_fail 只顺手 _prune 被访问的键;失败一次即消失的键(扫号
|
|
换 IP/换用户名)永不被回访 → 永久滞留。
|
|
- app.engine.services.rate_limit._buckets: (tenant, ip, device_id) →
|
|
list[float]——check() 同款单键懒清,键随客户端轮换只增不减。
|
|
sweep() 全表清「全部时间戳均已出窗」的键,由 main.lifespan 的周期任务
|
|
(300s sleep loop)调用。时间推进手法照 test_auth 既有 lockout 用例:
|
|
直改内部时间戳,免真实等待。
|
|
"""
|
|
import asyncio
|
|
import time
|
|
|
|
from tests.conftest import (
|
|
JIANCAI_HOST,
|
|
SEED_DIR,
|
|
app_test_client,
|
|
build_app,
|
|
)
|
|
|
|
|
|
class TestLoginLimiterSweep:
|
|
def test_清过期保活跃(self):
|
|
from app.platform.auth import LoginLimiter
|
|
limiter = LoginLimiter(window_seconds=900.0, max_fails=5)
|
|
limiter.record_fail("u", "1.1.1.1")
|
|
limiter.record_fail("u", "2.2.2.2")
|
|
limiter._fails[("u", "1.1.1.1")] = [0.0] # 远古 → 待清
|
|
n = limiter.sweep(now=time.time())
|
|
assert n == 1
|
|
assert ("u", "1.1.1.1") not in limiter._fails
|
|
assert ("u", "2.2.2.2") in limiter._fails
|
|
|
|
def test_全活跃时清零(self):
|
|
from app.platform.auth import LoginLimiter
|
|
limiter = LoginLimiter(window_seconds=900.0)
|
|
limiter.record_fail("u", "1.1.1.1")
|
|
assert limiter.sweep(now=time.time()) == 0
|
|
assert ("u", "1.1.1.1") in limiter._fails
|
|
|
|
def test_窗沿时间戳算过期_空键也清(self):
|
|
# ts[-1] <= now - window 与 _prune 的保留判据 t > now - window 同沿
|
|
from app.platform.auth import LoginLimiter
|
|
limiter = LoginLimiter(window_seconds=100.0)
|
|
now = time.time()
|
|
limiter._fails[("u", "1.1.1.1")] = [now - 100.0] # 恰在窗沿 → 出窗
|
|
limiter._fails[("u", "2.2.2.2")] = [] # 空值键 → 待清
|
|
assert limiter.sweep(now=now) == 2
|
|
assert not list(limiter._fails) # 全清空
|
|
|
|
|
|
class TestRateLimitSweep:
|
|
def test_同款语义(self):
|
|
from app.engine.services import rate_limit
|
|
try:
|
|
now = time.time()
|
|
rate_limit._buckets[("a", "1.1.1.1", "")] = [0.0] # 远古
|
|
rate_limit._buckets[("b", "2.2.2.2", "")] = [now] # 活跃
|
|
rate_limit._buckets[("c", "3.3.3.3", "")] = [] # 空值键
|
|
rate_limit._buckets[("d", "4.4.4.4", "")] = [
|
|
now - rate_limit.WINDOW] # 恰在窗沿
|
|
n = rate_limit.sweep(now=now)
|
|
assert n == 3
|
|
assert set(rate_limit._buckets) == {("b", "2.2.2.2", "")}
|
|
finally:
|
|
rate_limit._buckets.clear() # 模块级状态:清现场(conftest 亦兜底)
|
|
|
|
|
|
class TestLifespanSweepTask:
|
|
def test_注册与取消(self, tmp_path, monkeypatch):
|
|
"""lifespan 起周期清扫任务,关停时取消。
|
|
|
|
手法: monkeypatch asyncio.create_task 包一层记录真实 task(行为
|
|
不变);TestClient with 进出 = lifespan 启停。不真等 300s——loop
|
|
首步即 sleep,任务常 pending;取消断言 cancelled/cancelling 任一
|
|
成立即可(实现侧 cancel 后 await 落地,常态是 cancelled)。"""
|
|
created = []
|
|
real_create_task = asyncio.create_task
|
|
|
|
def _rec(coro, **kwargs):
|
|
t = real_create_task(coro, **kwargs)
|
|
created.append(t)
|
|
return t
|
|
|
|
monkeypatch.setattr(asyncio, "create_task", _rec)
|
|
a = build_app(SEED_DIR, tmp_path)
|
|
with app_test_client(a, f"http://{JIANCAI_HOST}", tmp_path):
|
|
tasks = [t for t in created
|
|
if getattr(t.get_coro(), "__name__", "")
|
|
== "_limiter_sweep_loop"]
|
|
assert len(tasks) == 1 # 启动期创建且恰好一份
|
|
assert not tasks[0].done() # 300s sleep 中不会自然完成
|
|
assert tasks[0].cancelled() or tasks[0].cancelling() > 0
|