fangke/server/tracker/tests/test_ingest.py

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}