203 lines
6.9 KiB
Python
203 lines
6.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""访问统计内存缓冲与批量落库测试（离线）。
|
|
|
|
运行：uv run python tests/test_visits.py
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
from reaction_board.database import Database # noqa: E402
|
|
from reaction_board.timeutil import day_bounds # noqa: E402
|
|
from reaction_board.web.visits import VisitCollector # noqa: E402
|
|
|
|
DAY_A = "2026-10-30"
|
|
DAY_B = "2026-10-31"
|
|
|
|
|
|
def ts(day: str, hour: int = 12) -> int:
|
|
return int(day_bounds(day)[0]) + hour * 3600
|
|
|
|
|
|
class _StubDB:
|
|
"""只让 transaction/write 抛错，用于验证 flush 失败时的并回行为。"""
|
|
|
|
def transaction(self, fn):
|
|
raise RuntimeError("db down")
|
|
|
|
def write(self, fn):
|
|
raise RuntimeError("db down")
|
|
|
|
|
|
def _open(tmp: str) -> Database:
|
|
db = Database(Path(tmp) / "app.db")
|
|
db.initialize()
|
|
return db
|
|
|
|
|
|
def _rows(db: Database):
|
|
return db.read(lambda conn: [
|
|
dict(row) for row in conn.execute(
|
|
"SELECT stat_date, ip, path, hits, first_at, last_at FROM visit "
|
|
"ORDER BY stat_date, ip, path"
|
|
).fetchall()
|
|
])
|
|
|
|
|
|
def test_record_and_aggregate():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db = _open(tmp)
|
|
try:
|
|
collector = VisitCollector(db, flush_seconds=0)
|
|
assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is True
|
|
assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is True
|
|
assert collector.record("1.1.1.1", "/videos", now=ts(DAY_A)) is True
|
|
assert collector.record("2.2.2.2", "/home", now=ts(DAY_A)) is True
|
|
# 跨日单独成键
|
|
assert collector.record("1.1.1.1", "/home", now=ts(DAY_B)) is True
|
|
|
|
assert collector.pending_keys == 4
|
|
rows = {(r["stat_date"], r["ip"], r["path"]): r
|
|
for r in collector.pending_rows()}
|
|
home_a = rows[(DAY_A, "1.1.1.1", "/home")]
|
|
assert home_a["hits"] == 2
|
|
assert home_a["first_at"] == ts(DAY_A)
|
|
assert home_a["last_at"] == ts(DAY_A)
|
|
assert rows[(DAY_B, "1.1.1.1", "/home")]["hits"] == 1
|
|
|
|
# 空值不计数
|
|
assert collector.record("", "/home", now=ts(DAY_A)) is False
|
|
assert collector.record("1.1.1.1", "", now=ts(DAY_A)) is False
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_flush_writes_and_is_idempotent():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db = _open(tmp)
|
|
try:
|
|
collector = VisitCollector(db, flush_seconds=0)
|
|
assert collector.flush() == 0 # 空缓冲 no-op
|
|
|
|
collector.record("1.1.1.1", "/home", now=ts(DAY_A))
|
|
collector.record("1.1.1.1", "/home", now=ts(DAY_A))
|
|
assert collector.flush() == 1
|
|
assert collector.pending_keys == 0
|
|
rows = _rows(db)
|
|
assert len(rows) == 1
|
|
assert rows[0]["hits"] == 2
|
|
assert rows[0]["ip"] == "1.1.1.1"
|
|
assert rows[0]["path"] == "/home"
|
|
|
|
# 再次 flush 不重复累加
|
|
assert collector.flush() == 0
|
|
assert _rows(db)[0]["hits"] == 2
|
|
|
|
# 与已有行累加（先在库中放一条 hits=5 的记录）
|
|
db.write(lambda conn: conn.execute(
|
|
"INSERT INTO visit(stat_date, ip, path, hits, first_at, last_at) "
|
|
"VALUES (?, '3.3.3.3', '/home', 5, ?, ?)",
|
|
(DAY_A, ts(DAY_A), ts(DAY_A)),
|
|
))
|
|
collector.record("3.3.3.3", "/home", now=ts(DAY_A, hour=13))
|
|
assert collector.flush() == 1
|
|
row = next(r for r in _rows(db) if r["ip"] == "3.3.3.3")
|
|
assert row["hits"] == 6
|
|
assert row["first_at"] == ts(DAY_A)
|
|
assert row["last_at"] == ts(DAY_A, hour=13)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_flush_failure_merges_back():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db = _open(tmp)
|
|
try:
|
|
collector = VisitCollector(db, flush_seconds=0)
|
|
collector.record("1.1.1.1", "/home", now=ts(DAY_A))
|
|
collector.record("1.1.1.1", "/home", now=ts(DAY_A))
|
|
|
|
original = collector._db
|
|
collector._db = _StubDB()
|
|
try:
|
|
assert collector.flush() == 0
|
|
assert collector.pending_keys == 1
|
|
assert collector.pending_rows()[0]["hits"] == 2
|
|
finally:
|
|
collector._db = original
|
|
|
|
# 恢复后能正常落库且计数不丢、不重复
|
|
assert collector.flush() == 1
|
|
assert _rows(db)[0]["hits"] == 2
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_max_hits_per_key_cap():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db = _open(tmp)
|
|
try:
|
|
collector = VisitCollector(db, flush_seconds=0, max_hits_per_key=2)
|
|
assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is True
|
|
assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is True
|
|
assert collector.record("1.1.1.1", "/home", now=ts(DAY_A)) is False
|
|
assert collector.pending_rows()[0]["hits"] == 2
|
|
# 不同键各自独立
|
|
assert collector.record("1.1.1.1", "/videos", now=ts(DAY_A)) is True
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_prune_old_rows():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db = _open(tmp)
|
|
try:
|
|
collector = VisitCollector(db, flush_seconds=0, keep_days=7)
|
|
# 30 天前的老记录
|
|
old_day = "2026-09-30"
|
|
db.write(lambda conn: conn.execute(
|
|
"INSERT INTO visit(stat_date, ip, path, hits, first_at, last_at) "
|
|
"VALUES (?, '9.9.9.9', '/home', 1, 1, 1)",
|
|
(old_day,),
|
|
))
|
|
collector.record("1.1.1.1", "/home", now=ts(DAY_A))
|
|
assert collector.flush() == 1
|
|
remaining = {row["stat_date"] for row in _rows(db)}
|
|
assert old_day not in remaining
|
|
assert DAY_A in remaining
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_stop_flushes_pending():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db = _open(tmp)
|
|
try:
|
|
collector = VisitCollector(db, flush_seconds=0)
|
|
collector.start() # flush_seconds=0 → 不启动线程
|
|
assert collector._thread is None
|
|
collector.record("1.1.1.1", "/home", now=ts(DAY_A))
|
|
collector.stop()
|
|
assert collector.pending_keys == 0
|
|
assert _rows(db)[0]["hits"] == 1
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def main():
|
|
test_record_and_aggregate()
|
|
test_flush_writes_and_is_idempotent()
|
|
test_flush_failure_merges_back()
|
|
test_max_hits_per_key_cap()
|
|
test_prune_old_rows()
|
|
test_stop_flushes_pending()
|
|
print("OK: 访问统计内存缓冲与批量落库测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |