Files
fangke/server/tracker/tests/test_ingest.py
T

387 lines
21 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"}]
# 14. SDK context 的 timezone 落库 Visitor.timezone_name(Task16 联调发现:键名映射缺口)
def test_context_timezone_persisted_to_visitor(db):
site = Site.objects.create(name="s", domains=["a.com"])
b = _batch([_ev(1)])
b["context"]["timezone"] = "Asia/Shanghai"
ingest_batch(site, b)
assert Visitor.objects.get(visitor_id="v1").timezone_name == "Asia/Shanghai"
# 15. 归因分类 best-effort:引擎抛异常不影响采集入库(first_touch/session_entry 双路)
def test_attribution_failure_never_breaks_ingest(db, monkeypatch):
from tracker.services import ingest as ingest_mod
site = Site.objects.create(name="s", domains=["a.com"])
monkeypatch.setattr(ingest_mod, "classify", lambda **kw: (_ for _ in ()).throw(RuntimeError("boom")))
r = ingest_batch(site, _batch([_ev(1)]))
assert r["accepted"] == 1 and Event.objects.count() == 1
assert Visitor.objects.count() == 1 and VisitSession.objects.count() == 1
# 16. 首触分类用本批第一条 page_view 的完整 referrer(搜索词可恢复),非 ctx 裸域名
def test_first_touch_classification_uses_full_pageview_referrer(db):
from tracker.models import SourceClassification
site = Site.objects.create(name="s", domains=["a.com"])
b = _batch([_ev(1, properties={"referrer": "https://m.baidu.com/s?wd=挖掘机"})])
b["context"]["utm"] = {} # 默认 fixture 带 utm_source=google,与百度矛盾 → conflict
ingest_batch(site, b)
ft = SourceClassification.objects.get(scope="first_touch")
assert ft.channel == "natural_search" and ft.platform == "百度"
assert ft.search_term == "挖掘机" and ft.resolution == "resolved"
# 17. 扩展环境键快照:device_model 独立列 + env_extra 白名单(白名单外不入库)
def test_extended_env_snapshot_persisted(db):
site = Site.objects.create(name="s", domains=["a.com"])
b = _batch([_ev(1)])
b["context"].update({
"device_model": "Pixel 7", "device_pixel_ratio": 2.625, "viewport": "393x851",
"orientation": "portrait-primary", "max_touch_points": 5, "device_memory": 8,
"languages": "zh-CN,en", "color_scheme": "dark", "standalone": "0",
"network_downlink": 10, "network_rtt": 50, "save_data": "0",
"wechat_version": "8.0.49", "platform_version": "13.0.0",
"architecture": "arm", "bitness": "64",
"evil_key": "<script>", # 白名单外:不入 env_extra
})
ingest_batch(site, b)
v = Visitor.objects.get(visitor_id="v1")
assert v.device_model == "Pixel 7"
assert v.env_extra == {
"device_model": "Pixel 7", "device_pixel_ratio": "2.625", "viewport": "393x851",
"orientation": "portrait-primary", "max_touch_points": "5", "device_memory": "8",
"languages": "zh-CN,en", "color_scheme": "dark", "standalone": "0",
"network_downlink": "10", "network_rtt": "50", "save_data": "0",
"wechat_version": "8.0.49", "platform_version": "13.0.0",
"architecture": "arm", "bitness": "64"}
# 18. 扩展字段截断:device_model 64 / env_extra 值 100
def test_extended_env_fields_truncated(db):
site = Site.objects.create(name="s", domains=["a.com"])
b = _batch([_ev(1)])
b["context"].update({"device_model": "M" * 80, "languages": "x" * 200})
ingest_batch(site, b)
v = Visitor.objects.get(visitor_id="v1")
assert v.device_model == "M" * 64 # 独立列按 CharField(64) 截断
assert v.env_extra["device_model"] == "M" * 80 # env_extra 白名单统一 [:100](80<100 原样)
assert v.env_extra["languages"] == "x" * 100
# 19. 晚到 UA-CH hints:首批无型号 → 第二批只补填 device_model 列,env_extra 保持首触快照
def test_device_model_backfilled_on_later_batch(db):
site = Site.objects.create(name="s", domains=["a.com"])
ingest_batch(site, _batch([_ev(1)])) # 首批无任何扩展键
v = Visitor.objects.get(visitor_id="v1")
assert v.device_model == "" and v.env_extra == {}
b = _batch([_ev(2)])
b["context"].update({"device_model": "Pixel 7", "viewport": "393x851"})
ingest_batch(site, b)
v.refresh_from_db()
assert v.device_model == "Pixel 7" # 空列补填一次
assert v.env_extra == {} # env_extra 不补写(首触快照语义)
b3 = _batch([_ev(3)])
b3["context"]["device_model"] = "iPhone 15 Pro"
ingest_batch(site, b3)
v.refresh_from_db()
assert v.device_model == "Pixel 7" # 已有值不再覆盖
# ---- schema v2(阶段A Task3:事件级归属 / session_resume / 区间并集 / 时间质量)----
SITE_ID = "site_v2"
def _now_iso():
return timezone.now().isoformat()
def _v2_event(name, page_id, sid, seq=1, **props):
return {"event_id": f"e-{name}-{seq}", "event_name": name,
"occurred_at": _now_iso(), "path": "/p", "title": "t",
"site_id": SITE_ID, "visitor_id": "v1", "session_id": sid,
"page_id": page_id, "seq": seq, "properties": props}
def _v2_batch(events, vid="v1", sid="s1"):
return {"schema_version": 2, "site_id": SITE_ID, "visitor_id": vid,
"session_id": sid, "sdk_version": "0.2.0",
"context": {"referrer_domain": "", "path": "/", "device_type": "PC"},
"events": events}
# 20. v2 事件级归属:同批多 session 各自落库(v1 信封 session_id 不再是唯一依据);
# 事件级 site/visitor 与批级不一致 → attribution_mismatch
def test_v2_event_level_attribution(db):
site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
evs = [_v2_event("page_view", "p1", "sA", seq=1),
_v2_event("page_view", "p2", "sB", seq=2)]
r = ingest_batch(site, _v2_batch(evs, sid="sA"))
assert (r["accepted"], r["duplicated"], r["rejected"]) == (2, 0, [])
assert VisitSession.objects.count() == 2
for sid_name, pid in (("sA", "p1"), ("sB", "p2")):
sess = VisitSession.objects.get(session_id=sid_name)
e = Event.objects.get(session=sess)
assert e.page_id == pid and e.schema_version == 2
bad_v = _v2_event("page_view", "p3", "sA", seq=3)
bad_v["visitor_id"] = "v-other" # 事件级 visitor ≠ 批级
bad_s = _v2_event("page_view", "p4", "sA", seq=4)
bad_s["site_id"] = "site_other" # 事件级 site ≠ 批级
r2 = ingest_batch(site, _v2_batch(
[bad_v, bad_s, _v2_event("page_view", "p5", "sA", seq=5)], sid="sA"))
assert r2["accepted"] == 1
assert r2["rejected"] == [
{"index": 0, "event_id": "e-page_view-3", "reason": "attribution_mismatch"},
{"index": 1, "event_id": "e-page_view-4", "reason": "attribution_mismatch"}]
# 21. session_resume:新会话 resume_of 链接 prev_session_id 对应会话;
# prev 不存在软处理(resume_of=None 不拒);v2 不做 30min 兜底切分
def test_session_resume_creates_linked_session(db):
site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
ingest_batch(site, _v2_batch([_v2_event("page_view", "p1", "s1", seq=1)], sid="s1"))
r = ingest_batch(site, _v2_batch([
_v2_event("session_resume", "", "s2", seq=2, prev_session_id="s1", reason="idle_timeout"),
_v2_event("page_view", "p2", "s2", seq=3)], sid="s2"))
assert (r["accepted"], r["rejected"]) == (2, [])
s1 = VisitSession.objects.get(session_id="s1")
s2 = VisitSession.objects.get(session_id="s2")
assert s2.resume_of == s1
r2 = ingest_batch(site, _v2_batch([
_v2_event("session_resume", "", "s3", seq=4, prev_session_id="ghost", reason="idle_timeout"),
_v2_event("page_view", "p3", "s3", seq=5)], sid="s3"))
assert (r2["accepted"], r2["rejected"]) == (2, []) # prev 无效不拒
assert VisitSession.objects.get(session_id="s3").resume_of is None
assert Visitor.objects.get(visitor_id="v1").visit_count == 3 # s1/s2/s3
# 35min 后同 sid 再发:不切 s2#2(v2 会话切换由 SDK 显式 resume 驱动)
late_ev = _v2_event("page_view", "p2b", "s2", seq=6)
late_ev["occurred_at"] = (timezone.now() + timedelta(minutes=35)).isoformat()
r3 = ingest_batch(site, _v2_batch([late_ev], sid="s2"))
assert r3["accepted"] == 1
assert VisitSession.objects.filter(session_id__startswith="s2").count() == 1
# 22. v2 page_stay 区间并集(非加法):[0,15][15,30][40,55]s → 0-30 ∪ 40-55 = 45s
def test_stay_interval_union_not_sum(db):
site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
secs = [(0, 15), (15, 30), (40, 55)]
evs = [_v2_event("page_stay", "p1", "s1", seq=i + 1,
active_from=a * 1000, active_to=b * 1000, flush_reason="periodic")
for i, (a, b) in enumerate(secs)]
r = ingest_batch(site, _v2_batch(evs, sid="s1"))
assert (r["accepted"], r["rejected"]) == (3, [])
e1 = Event.objects.get(event_id="e-page_stay-1")
assert (e1.active_from_ms, e1.active_to_ms) == (0, 15000)
assert e1.flush_reason == "periodic" and e1.sdk_version == "0.2.0"
assert VisitSession.objects.get(session_id="s1").stay_ms == 45000 # 非 sum=60000
# 23. v1 路径完全不变:stay_ms 仍为 active_ms 加法,v2 新字段留空
def test_v1_events_unchanged_path(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, "page_stay", properties={"active_ms": 3000, "path": "/p"})]))
sess = VisitSession.objects.get()
assert sess.stay_ms == 8000 and sess.event_count == 3
e = Event.objects.get(event_id="e1")
assert e.schema_version is None and e.page_id == "" and e.time_quality == ""
assert e.active_from_ms is None and e.active_to_ms is None
# 24. time_quality 分类:正常 ok / >120s 过去 late / 未来小偏移 ok / 未来超阈 skew_backwards
def test_time_quality_classification(db):
site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
cases = [(1, 0, "ok"), (2, -130, "late"), (3, 5, "ok"), (4, 130, "skew_backwards")]
evs = []
for seq, off, _ in cases:
e = _v2_event("page_view", f"p{seq}", "s1", seq=seq)
e["occurred_at"] = (timezone.now() + timedelta(seconds=off)).isoformat()
evs.append(e)
r = ingest_batch(site, _v2_batch(evs, sid="s1"))
assert (r["accepted"], r["rejected"]) == (4, [])
for seq, off, want in cases:
e = Event.objects.get(event_id=f"e-page_view-{seq}")
assert e.time_quality == want, f"seq={seq} offset={off}s"
assert e.late == (abs(off) > 120) # late 布尔语义不动
# 25. 混合解释:同会话 v1 active_ms 加法 + v2 区间并集 相加(单一解释)
def test_recompute_mixed_v1_v2(db):
site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
ingest_batch(site, _batch(
[_ev(1), _ev(2, "page_stay", properties={"active_ms": 10000, "path": "/p"})], sid="sm"))
evs = [_v2_event("page_stay", "px", "sm", seq=1, active_from=0, active_to=10000, flush_reason="periodic"),
_v2_event("page_stay", "py", "sm", seq=2, active_from=0, active_to=5000, flush_reason="periodic")]
r = ingest_batch(site, _v2_batch(evs, sid="sm"))
assert (r["accepted"], r["rejected"]) == (2, [])
assert VisitSession.objects.get(session_id="sm").stay_ms == 10000 + 15000 # v1 SUM + v2 并集
# 26. 脏区间防御:非数值/倒置区间不参与并集、不拒批(防 stay_ms 负值炸 PositiveBigInteger)
def test_dirty_intervals_soft_skipped(db):
site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
evs = [_v2_event("page_stay", "p1", "s1", seq=1, active_from=0, active_to=10000, flush_reason="periodic"),
_v2_event("page_stay", "p2", "s1", seq=2, active_from="x", active_to=9000, flush_reason="periodic"),
_v2_event("page_stay", "p3", "s1", seq=3, active_from=20000, active_to=10000, flush_reason="periodic")]
r = ingest_batch(site, _v2_batch(evs, sid="s1"))
assert (r["accepted"], r["rejected"]) == (3, []) # 脏数据软处理不拒批
e2 = Event.objects.get(event_id="e-page_stay-2")
assert e2.active_from_ms is None # 非数值 int 化为 None
assert VisitSession.objects.get(session_id="s1").stay_ms == 10000 # 仅 p1 有效区间
# 27. NaN/Infinity/超量纲区间值:NaN/Inf 是非法 JSON 数值(json.loads 默认收,但 SQLite
# JSON 有效性校验拒 + bulk_create(ignore_conflicts) 静默丢行而 accepted 已计数——幽灵
# 事件),入口 properties_invalid 拒;有限超量纲值(1e308)合法入库但区间列降级 None
def test_nonfinite_interval_values_rejected_or_degraded(db):
site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
evs = [_v2_event("page_stay", "p1", "s1", seq=1, active_from=0, active_to=10000, flush_reason="periodic"),
_v2_event("page_stay", "p2", "s1", seq=2,
active_from=float("nan"), active_to=float("inf"), flush_reason="periodic"),
_v2_event("page_stay", "p3", "s1", seq=3,
active_from=-1e308, active_to=1e308, flush_reason="periodic")]
r = ingest_batch(site, _v2_batch(evs, sid="s1"))
assert (r["accepted"], r["duplicated"]) == (2, 0)
assert r["rejected"] == [{"index": 1, "event_id": "e-page_stay-2",
"reason": "properties_invalid"}]
e3 = Event.objects.get(event_id="e-page_stay-3")
assert e3.active_from_ms is None and e3.active_to_ms is None # 超 ±2^63 量纲 → None
assert VisitSession.objects.get(session_id="s1").stay_ms == 10000
# 28. page_id 校验(invalid_page_id):session_resume 仅豁免"空值",非空值一律完整校验
# (孤代理 "\ud800" 是合规 JSON,跳过 _utf8_ok 会在存储时炸整批)
def test_v2_page_id_validation(db):
site = Site.objects.create(name="s", domains=["a.com"], site_id=SITE_ID)
bad_surrogate = _v2_event("session_resume", "", "s1", seq=1,
prev_session_id="s0", reason="idle_timeout")
bad_surrogate["page_id"] = "p\ud800" # session_resume 的非空 page_id 也不豁免孤代理
bad_long = _v2_event("page_view", "L" * 65, "s1", seq=2)
bad_type = _v2_event("page_view", 123, "s1", seq=3)
ok_resume = _v2_event("session_resume", "", "s1", seq=4,
prev_session_id="s0", reason="idle_timeout")
r = ingest_batch(site, _v2_batch([bad_surrogate, bad_long, bad_type, ok_resume], sid="s1"))
assert r["accepted"] == 1 # 空 page_id 的 session_resume 通过
assert [x["reason"] for x in r["rejected"]] == ["invalid_page_id"] * 3
assert {x["index"] for x in r["rejected"]} == {0, 1, 2}