peter
ce8ccfc3c0
quota_events had no retention: one row per API spend plus a zero-unit row per live search, kept forever. The daily history cleanup (audit_gc) now also prunes it past quota_retention_days (a new Quota setting, default 365 so the admin usage view's longest window stays whole; 0 = forever). The floor of two days keeps today's rows, which the per-user search cap counts.
47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
"""`quota_events` has a retention: the daily history-cleanup job (`audit_gc`) prunes rows older than
|
|
`quota_retention_days`, as it does the audit log. Without it the table only ever grew — one row per
|
|
API spend plus a zero-unit row per live search."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app import scheduler, sysconfig
|
|
from app.models import QuotaEvent
|
|
|
|
|
|
def _event(days_ago: int) -> QuotaEvent:
|
|
return QuotaEvent(
|
|
user_id=None,
|
|
action="videos.list",
|
|
units=1,
|
|
created_at=datetime.now(timezone.utc) - timedelta(days=days_ago),
|
|
)
|
|
|
|
|
|
def _remaining(db) -> list[int]:
|
|
db.expire_all()
|
|
return sorted(e.units for e in db.execute(select(QuotaEvent)).scalars())
|
|
|
|
|
|
def test_the_history_cleanup_prunes_quota_events_past_the_retention(db):
|
|
old, recent = _event(400), _event(10)
|
|
recent.units = 7
|
|
db.add_all([old, recent])
|
|
db.commit()
|
|
|
|
scheduler.JOB_FUNCS["audit_gc"]()
|
|
|
|
assert _remaining(db) == [7] # the default keeps a year, which the admin usage view can show
|
|
|
|
|
|
def test_a_zero_retention_keeps_quota_history_forever(db):
|
|
db.add(_event(4000))
|
|
db.commit()
|
|
sysconfig.set_value(db, "quota_retention_days", 0)
|
|
db.commit()
|
|
|
|
scheduler.JOB_FUNCS["audit_gc"]()
|
|
|
|
assert _remaining(db) == [1]
|