Files
fangke/server/tracker/tests/test_ingest.py
T
2026-09-15 12:59:13 +08:00

137 lines
6.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from datetime import timedelta
from django.utils import timezone
from tracker.models import Site, Visitor, VisitSession, Event
from tracker.services.geo import resolver as _geo
from tracker.services.ingest import ingest_batch
NOW = timezone.now()
def _batch(events, vid="v1", sid="s1"):
return {"visitor_id": vid, "session_id": sid, "context": {
"referrer_domain": "google.com", "utm": {"utm_source": "google"},
"path": "/landing", "device_type": "PC", "os": "Windows",
"browser": "Chrome", "language": "en-US", "screen": "1920x1080"},
"events": events}
def _ev(i, name="page_view", **kw):
d = dict(event_id=f"e{i}", event_name=name, occurred_at=NOW,
path="/p", title="T", properties={})
d.update(kw)
return d
# 1. 首批建档
def test_first_batch_creates_visitor_session_events(db):
site = Site.objects.create(name="s", domains=["a.com"])
r = ingest_batch(site, _batch([_ev(1), _ev(2)]), ip="114.114.114.114")
assert (r["accepted"], r["duplicated"], r["rejected"]) == (2, 0, [])
v = Visitor.objects.get(site=site, visitor_id="v1")
assert v.first_referrer_domain == "google.com" and v.visit_count == 1
assert v.active_days == 1 and v.last_ip == "114.114.114.114"
if _geo._searcher: # 地理断言仅在有真实 xdb 时执行,CI/fresh clone 不因缺库变红
assert v.country == "中国" and v.city == "南京市" # 114 DNS(江苏南京)
assert Event.objects.filter(site=site).count() == 2
# 2. 重放幂等 + 不抬高会话聚合
def test_duplicate_events_never_inflate_aggregates(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([_ev(1), _ev(2, "page_stay", properties={"active_ms": 3000, "path": "/p"})]))
r = ingest_batch(site, _batch([_ev(1)])) # 重放
assert r["duplicated"] == 1 and Event.objects.count() == 2
sess = VisitSession.objects.get()
assert sess.event_count == 2 and sess.stay_ms == 3000 # 重放没有重复加分
# 3. 新会话递增 visit_count(F 表达式)
def test_new_session_increments_visit_count(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([_ev(1)], sid="s1"))
ingest_batch(site, _batch([_ev(2)], sid="s2"))
v = Visitor.objects.get(visitor_id="v1")
assert v.visit_count == 2 and VisitSession.objects.count() == 2
# 4. 服务端 30 分钟兜底切分(同 sid 间隔>30min 切出 s1#2,且同批后续事件复用同段)
def test_server_backstop_splits_idle_gap(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([_ev(1, occurred_at=NOW - timedelta(minutes=90))]))
r = ingest_batch(site, _batch([_ev(2), _ev(3)])) # 两条同批
assert VisitSession.objects.count() == 2 # s1 与 s1#2,不会切成 3 段
assert r["accepted"] == 2
# 5. 拒绝事件带原因上报
def test_rejected_events_reported_with_reason(db):
site = Site.objects.create(name="s", domains=["a.com"])
r = ingest_batch(site, _batch([_ev(1, event_name="bad"), _ev(2)]))
assert r["accepted"] == 1 and r["rejected"][0]["reason"] == "event_name_not_allowed"
# 6. page_stay 的 active_ms 累计进 stay_ms
def test_stay_ms_accumulates(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([
_ev(1), _ev(2, "page_stay", properties={"active_ms": 5000, "path": "/p"})]))
ingest_batch(site, _batch([
_ev(3, occurred_at=NOW + timedelta(minutes=1)),
_ev(4, "page_stay", properties={"active_ms": 7000, "path": "/p"},
occurred_at=NOW + timedelta(minutes=1))]))
assert VisitSession.objects.get().stay_ms == 12000
# 7. 爬虫 UA 标记(访客+事件)
def test_bot_ua_marked(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([_ev(1)]), ip="66.249.66.1", ua="Mozilla/5.0 (compatible; Googlebot/2.1)")
assert Visitor.objects.get().is_bot is True
assert Event.objects.get().is_bot is True
# 8. 超批拒绝
def test_batch_size_invalid(db):
site = Site.objects.create(name="s", domains=["a.com"])
r = ingest_batch(site, _batch([_ev(i) for i in range(60)]))
assert r["rejected"] == [{"reason": "batch_size_invalid"}]
# 9. 同批重复 event_id:去重后只入一份,聚合不灌倍
def test_intra_batch_duplicate_event_id_deduped(db):
site = Site.objects.create(name="s", domains=["a.com"])
evs = [_ev(1, "page_stay", properties={"active_ms": 1000, "path": "/p"})
for _ in range(3)] # 3 份同 id
r = ingest_batch(site, _batch(evs))
assert (r["accepted"], r["duplicated"]) == (1, 2)
sess = VisitSession.objects.get()
assert sess.event_count == 1 and sess.stay_ms == 1000
# 10. 切分后同 sid 小间隔连发:空闲比较取会话族最大 ended_at,不再每批新开段
def test_split_then_small_gap_reuses_segment(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([_ev(1, occurred_at=NOW - timedelta(minutes=90))]))
ingest_batch(site, _batch([_ev(2)])) # 90min 间隔 → 切出 s1#2
ingest_batch(site, _batch([_ev(3, occurred_at=NOW + timedelta(minutes=1))]))
ingest_batch(site, _batch([_ev(4, occurred_at=NOW + timedelta(minutes=2))]))
assert VisitSession.objects.count() == 2 # 保持 s1 与 s1#2,不切 s1#3/s1#4
assert Visitor.objects.get(visitor_id="v1").visit_count == 2
# 11. invariant 4:乱序批次(后入更早发生)ended_at 不回退
def test_ended_at_never_regresses_on_out_of_order_batch(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([_ev(1, occurred_at=NOW + timedelta(minutes=1))]))
ingest_batch(site, _batch([_ev(2, occurred_at=NOW)]))
sess = VisitSession.objects.get()
assert abs((sess.ended_at - (NOW + timedelta(minutes=1))).total_seconds()) < 1
# 12. 首触持久:第二批换 referrer 不改首触字段
def test_first_touch_immutable_across_batches(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([_ev(1)]))
b = _batch([_ev(2)])
b["context"]["referrer_domain"] = "bing.com"
ingest_batch(site, b)
v = Visitor.objects.get(visitor_id="v1")
assert v.first_referrer_domain == "google.com"
assert v.first_utm == {"utm_source": "google"}
# 13. payload 缺 visitor_id/session_id 提前拒绝
def test_missing_payload_fields_rejected(db):
site = Site.objects.create(name="s", domains=["a.com"])
p = _batch([_ev(1)])
del p["visitor_id"]
assert ingest_batch(site, p)["rejected"] == [{"reason": "invalid_payload"}]
p = _batch([_ev(1)])
del p["session_id"]
assert ingest_batch(site, p)["rejected"] == [{"reason": "invalid_payload"}]