xinyouhe/fangke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c9967b6d707e30e2ac62939843e21fc617516541
fangke/server/tracker/tests/test_visitors.py
T

574 lines
31 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.
import pytest
from datetime import timedelta
from django.contrib.auth import get_user_model
from django.utils import timezone
from rest_framework.test import APIClient
from tracker.models import Event, Site, SourceClassification, VisitSession, Visitor
from tracker.services.ingest import ingest_batch
from tracker.tests.test_ingest import SITE_ID, _batch, _ev, _v2_batch, _v2_event
pytestmark = pytest.mark.django_db
User = get_user_model()
# 用上海本地时刻做基准：时间线/日期筛选按本地自然日分组（localdate/localtime），
# 直接用 UTC 的 now 在本地 00:00-08:00 之间跑测试会错一天（与 test_dashboard 同款防 flaky）
NOW = timezone.localtime(timezone.now())
TODAY = NOW.date()
YESTERDAY = TODAY - timedelta(days=1)
@pytest.fixture
def client(): # 覆盖 pytest-django 的 Django Client——认证用例需要 credentials()
return APIClient()
@pytest.fixture
def member_user():
return User.objects.create_user(username="member", password="memberpass123")
def _auth(client, user):
from rest_framework_simplejwt.tokens import RefreshToken
token = RefreshToken.for_user(user)
client.credentials(HTTP_AUTHORIZATION=f"Bearer {token.access_token}")
def _visitor(site, vid, last_seen=None, **kw):
v = Visitor.objects.create(site=site, visitor_id=vid, **kw)
if last_seen is not None: # last_seen_at 是 auto_now_add，只能 queryset.update 绕开
Visitor.objects.filter(pk=v.pk).update(last_seen_at=last_seen)
return v
def _session(site, visitor, sid, at=NOW, referrer_domain="", utm=None,
landing_path="", event_count=0, duration_ms=0, stay_ms=0):
return VisitSession.objects.create(site=site, visitor=visitor, session_id=sid,
started_at=at, ended_at=at + timedelta(minutes=5), referrer_domain=referrer_domain,
utm=utm or {}, landing_path=landing_path, event_count=event_count,
duration_ms=duration_ms, stay_ms=stay_ms)
def _event(site, visitor, session, eid, name, at, path="", title="", active_ms=None,
properties=None):
return Event.objects.create(site=site, visitor=visitor, session=session,
event_id=eid, event_name=name, occurred_at=at, path=path, title=title,
properties=properties if properties is not None else (
{"active_ms": active_ms} if active_ms is not None else {}))
def test_list_fields_and_default_bot_excluded(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v1 = _visitor(site, "v1", country="中国", province="上海", city="上海",
device_type="desktop", os="Windows", browser="Chrome", language="zh-CN",
first_referrer_domain="google.com", first_landing_path="/from-google",
visit_count=2, active_days=1, last_seen=timezone.now())
v2 = _visitor(site, "v2", first_utm={"utm_source": "baidu"},
last_seen=timezone.now() - timedelta(hours=1))
bot = _visitor(site, "bot-v")
Visitor.objects.filter(pk=bot.pk).update(is_bot=True)
_session(site, v1, "s1", at=NOW - timedelta(hours=2), stay_ms=120000)
_session(site, v1, "s2", at=NOW, referrer_domain="bing.com", stay_ms=30000)
_session(site, v2, "s3", stay_ms=5000)
_session(site, bot, "sb", stay_ms=999)
r = client.get("/api/admin/visitors")
assert r.status_code == 200
d = r.json()["data"]
assert d["total"] == 2 and d["page"] == 1 and len(d["list"]) == 2
assert [x["visitor_id"] for x in d["list"]] == ["v1", "v2"] # last_seen 倒序
row = d["list"][0]
assert set(row) == {"visitor_id", "site_id", "site_name", "country", "province", "city",
"device_type", "os", "browser", "language", "first_seen_at",
"last_seen_at", "visit_count", "active_days", "session_count",
"total_stay_ms", "first_source", "last_source", "first_landing_path",
"last_path", "online", "channel", "platform", "variant", "search_term",
"resolution", "device_model"}
assert row["country"] == "中国" and row["os"] == "Windows" and row["browser"] == "Chrome"
assert row["session_count"] == 2
assert row["total_stay_ms"] == 150000 # 120000+30000，有效停留口径
assert row["first_source"] == "google.com"
assert row["last_source"] == "bing.com" # 最近会话 s2 的 referrer
assert row["first_landing_path"] == "/from-google"
assert row["online"] is True
row2 = d["list"][1]
assert row2["first_source"] == "baidu" # first_referrer 空 → first_utm 兜底
assert row2["first_landing_path"] == "" # 未采集 → 空串（前端显示 —）
assert row2["last_source"] == "直接访问" # 会话无来源 → 兜底
assert row2["total_stay_ms"] == 5000 and row2["online"] is False
d = client.get("/api/admin/visitors", {"show_bot": "1"}).json()["data"]
assert d["total"] == 3 and len(d["list"]) == 3
bot_row = next(x for x in d["list"] if x["visitor_id"] == "bot-v")
assert bot_row["first_source"] == "直接访问" # 无 referrer 无 utm
def test_list_filters_and_search(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
_visitor(site, "cn-desktop", country="中国", device_type="desktop",
last_ip="1.2.3.4", last_seen=timezone.now())
_visitor(site, "us-mobile", country="美国", device_type="mobile",
last_ip="5.6.7.8", last_seen=timezone.now() - timedelta(hours=2))
old = _visitor(site, "old-visitor", last_seen=NOW - timedelta(days=1))
Visitor.objects.filter(pk=old.pk).update(first_seen_at=NOW - timedelta(days=1))
ids = lambda d: [x["visitor_id"] for x in d["list"]] # noqa: E731
d = client.get("/api/admin/visitors", {"country": "中国"}).json()["data"]
assert ids(d) == ["cn-desktop"]
d = client.get("/api/admin/visitors", {"device_type": "mobile"}).json()["data"]
assert ids(d) == ["us-mobile"]
d = client.get("/api/admin/visitors", {"online": "1"}).json()["data"]
assert ids(d) == ["cn-desktop"]
d = client.get("/api/admin/visitors", {"search": "CN-"}).json()["data"] # istartswith 大小写不敏感
assert ids(d) == ["cn-desktop"]
d = client.get("/api/admin/visitors", {"search": "5.6.7"}).json()["data"] # last_ip 前缀
assert ids(d) == ["us-mobile"]
d = client.get("/api/admin/visitors", {"search": "zzz"}).json()["data"]
assert d["total"] == 0 and d["list"] == []
d = client.get("/api/admin/visitors", {"is_new": "1"}).json()["data"] # first_seen 在今天
assert set(ids(d)) == {"cn-desktop", "us-mobile"}
d = client.get("/api/admin/visitors", {"date_to": YESTERDAY.isoformat()}).json()["data"]
# 断言 date_to 的确定边界：昨天者必收、今天者必排除。us-mobile 的 now-2h 在本地
# 00:00-02:00 间跑测试会落到昨天（被一并收入）——属时钟归属，不作断言对象（防 flaky）。
assert "old-visitor" in ids(d) and "cn-desktop" not in ids(d)
def test_list_site_fields_and_site_filter(client, member_user):
_auth(client, member_user)
site_a = Site.objects.create(name="站点A", domains=["a.com"])
site_b = Site.objects.create(name="站点B", domains=["b.com"])
_visitor(site_a, "va", last_seen=timezone.now())
_visitor(site_b, "vb", last_seen=timezone.now() - timedelta(hours=1))
d = client.get("/api/admin/visitors").json()["data"] # 缺省聚合全部站点
assert d["total"] == 2 and len(d["list"]) == 2
by_id = {x["visitor_id"]: x for x in d["list"]}
assert by_id["va"]["site_id"] == site_a.site_id and by_id["va"]["site_name"] == "站点A"
assert by_id["vb"]["site_id"] == site_b.site_id and by_id["vb"]["site_name"] == "站点B"
d = client.get("/api/admin/visitors", {"site": site_b.site_id}).json()["data"]
assert d["total"] == 1 and [x["visitor_id"] for x in d["list"]] == ["vb"] # 只剩本站访客
def test_list_last_path(client, member_user):
"""列表行 last_path：该访客最近一条事件的 path（Subquery 注解，无 N+1）；
无事件访客为空串。"""
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v1 = _visitor(site, "v1", last_seen=timezone.now())
v2 = _visitor(site, "v2", last_seen=timezone.now() - timedelta(hours=1))
s1 = _session(site, v1, "s1", at=NOW - timedelta(hours=2))
_event(site, v1, s1, "e1", "page_view", NOW - timedelta(hours=2), path="/old")
_event(site, v1, s1, "e2", "page_view", NOW - timedelta(minutes=90), path="/stay")
_event(site, v1, s1, "e3", "page_stay", NOW - timedelta(minutes=30), path="/stay",
active_ms=1000) # occurred_at 早于 e2 → 不改判"最近"
_event(site, v1, s1, "e4", "page_view", NOW, path="/newest")
d = client.get("/api/admin/visitors").json()["data"]
by_id = {x["visitor_id"]: x for x in d["list"]}
assert by_id["v1"]["last_path"] == "/newest" # 最近一条事件的 path
assert by_id["v2"]["last_path"] == "" # 无事件 → 空串（前端显示 —）
def test_list_site_not_found(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
_visitor(site, "v1", last_seen=timezone.now())
r = client.get("/api/admin/visitors", {"site": "site_nope"})
assert r.status_code == 404
body = r.json()
assert body["code"] == 404 and body["data"] is None and body["message"] == "site_not_found"
def test_list_pagination_and_order(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
for i in range(25):
_visitor(site, f"v{i:02d}", last_seen=timezone.now() - timedelta(minutes=i))
d = client.get("/api/admin/visitors").json()["data"]
assert d["total"] == 25 and d["page"] == 1 and len(d["list"]) == 20
assert d["list"][0]["visitor_id"] == "v00" and d["list"][-1]["visitor_id"] == "v19"
d = client.get("/api/admin/visitors", {"page": "2"}).json()["data"]
assert d["total"] == 25 and d["page"] == 2 and len(d["list"]) == 5
assert [x["visitor_id"] for x in d["list"]] == ["v20", "v21", "v22", "v23", "v24"]
def test_detail_shape(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1", screen="1920x1080", timezone_name="Asia/Shanghai",
last_ip="1.2.3.4", first_utm={"utm_source": "google"},
first_referrer_domain="google.com", first_landing_path="/landing")
s_old = _session(site, v, "s-old", at=NOW - timedelta(hours=2),
referrer_domain="google.com", stay_ms=60000, duration_ms=300000,
event_count=3)
s_new = _session(site, v, "s-new", at=NOW, utm={"utm_source": "wechat"},
landing_path="/new", stay_ms=30000, duration_ms=120000, event_count=2)
_event(site, v, s_old, "e1", "page_view", NOW - timedelta(hours=2), path="/a")
_event(site, v, s_old, "e2", "page_stay", NOW - timedelta(minutes=119),
path="/a", active_ms=1000)
_event(site, v, s_new, "e3", "page_view", NOW, path="/b")
r = client.get("/api/admin/visitors/v1")
assert r.status_code == 200
d = r.json()["data"]
assert d["site_id"] == site.site_id and d["site_name"] == "s" # 详情同样输出站点归属
assert d["pages_viewed"] == 2 # 只计 page_view；page_stay 是停留信号不并计
assert d["screen"] == "1920x1080" and d["timezone_name"] == "Asia/Shanghai"
assert d["last_ip"] == "1.2.3.4" and d["first_utm"] == {"utm_source": "google"}
assert d["first_landing_path"] == "/landing" and d["first_referrer_domain"] == "google.com"
assert d["session_count"] == 2 and d["total_stay_ms"] == 90000
assert d["first_source"] == "google.com"
assert d["last_source"] == "wechat" # 最近会话 s-new：referrer 空 → utm 兜底
assert d["last_path"] == "/b" # 最近一条事件 e3 的 path
assert [s["session_id"] for s in d["sessions"]] == ["s-new", "s-old"] # 最近在前
srow = d["sessions"][0]
assert set(srow) == {"session_id", "started_at", "ended_at", "referrer_domain", "utm",
"landing_path", "event_count", "stay_ms"} # duration_ms 内部字段不展示
assert srow["utm"] == {"utm_source": "wechat"}
assert srow["stay_ms"] == 30000 and srow["event_count"] == 2
def test_timeline_day_grouped_and_stay_merge(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
yn = NOW - timedelta(days=1) # 昨天同一本地时刻，必落上海昨日窗口
sy = _session(site, v, "sy", at=yn)
_event(site, v, sy, "y1", "sdk_loaded", yn, path="/")
_event(site, v, sy, "y2", "page_view", yn + timedelta(seconds=1), path="/pricing", title="Pricing")
_event(site, v, sy, "y3", "page_stay", yn + timedelta(seconds=2), path="/pricing", active_ms=60000)
st = _session(site, v, "st", at=NOW)
_event(site, v, st, "t1", "sdk_loaded", NOW, path="/")
_event(site, v, st, "t2", "page_view", NOW + timedelta(seconds=1), path="/pricing", title="Pricing")
_event(site, v, st, "t3", "page_stay", NOW + timedelta(seconds=2), path="/pricing", active_ms=78000)
_event(site, v, st, "t4", "page_stay", NOW + timedelta(seconds=3), path="/pricing", active_ms=20000)
r = client.get("/api/admin/visitors/v1/timeline")
assert r.status_code == 200
data = r.json()["data"]
assert set(data) == {"days", "truncated"}
assert data["truncated"] is False # 未触 1000 上限
groups = data["days"]
assert [g["date"] for g in groups] == [TODAY.isoformat(), YESTERDAY.isoformat()] # 日期组倒序
today_g, yest_g = groups
assert set(today_g) == {"date", "sessions"}
assert [s["session_id"] for s in today_g["sessions"]] == ["st"]
assert set(today_g["sessions"][0]) == {"session_id", "started_at", "events"}
evs = today_g["sessions"][0]["events"]
assert set(evs[0]) == {"time", "event_name", "path", "title", "active_ms"} # 白名单字段
assert evs[0]["event_name"] == "sdk_loaded" and evs[0]["active_ms"] is None
pricing = evs[1]
assert pricing["event_name"] == "page_view" and pricing["path"] == "/pricing"
assert pricing["title"] == "Pricing"
assert pricing["active_ms"] == 98000 # 78000 + 20000 两段累加进同一条 page_view
y_evs = yest_g["sessions"][0]["events"]
assert len(y_evs) == 2 # sdk_loaded + 合并后的 /pricing
assert y_evs[1]["path"] == "/pricing" and y_evs[1]["active_ms"] == 60000 # 跨天不合并
assert y_evs[1]["title"] == "Pricing"
def test_timeline_business_event_safe_fields(client, member_user):
"""时间线业务事件安全字段白名单（阶段B T6 Step0）：六类新事件行各带映射字段
（form_submit 的 submission_id 只透前 8 位），其余 properties 键一律不泄漏
（含故意塞进去的 PII 键）；单会话接口走同一序列化，同样只出白名单。"""
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
s = _session(site, v, "s1")
base = NOW
_event(site, v, s, "b1", "contact_click", base, path="/contact",
properties={"channel": "whatsapp", "label": " WhatsApp 按钮", "href": "https://wa.me/8613800000000"})
_event(site, v, s, "b2", "form_start", base + timedelta(seconds=1),
properties={"form_id": "cf7-12", "page_url": "https://a.com/contact"})
_event(site, v, s, "b3", "form_submit", base + timedelta(seconds=2),
properties={"form_id": "cf7-12", "submission_id": "abcd1234-9999-zzzz",
"fields": {"your-name": "张三", "your-email": "z@x.com"}})
_event(site, v, s, "b4", "form_error", base + timedelta(seconds=3),
properties={"error_code": "validation_failed", "field_count": 2,
"messages": ["邮箱格式不对"]})
_event(site, v, s, "b5", "form_submit_result", base + timedelta(seconds=4),
properties={"result_code": "mail_sent", "smtp_detail": "250 ok"})
_event(site, v, s, "b6", "product_view", base + timedelta(seconds=5),
properties={"product_id": "excavator-x1", "inferred": True,
"url": "https://a.com/product/excavator-x1"})
r = client.get("/api/admin/visitors/v1/timeline")
assert r.status_code == 200
evs = r.json()["data"]["days"][0]["sessions"][0]["events"]
base_keys = {"time", "event_name", "path", "title", "active_ms"}
e1, e2, e3, e4, e5, e6 = evs
assert set(e1) == base_keys | {"channel"} and e1["channel"] == "whatsapp"
assert set(e2) == base_keys | {"form_id"} and e2["form_id"] == "cf7-12"
assert set(e3) == base_keys | {"form_id", "submission_id"}
assert e3["submission_id"] == "abcd1234" # 前 8 位截断，尾部不透出
assert "abcd1234-9999" not in r.content.decode() # 全量 submission_id 不出现在响应
assert set(e4) == base_keys | {"error_code", "field_count"}
assert e4["error_code"] == "validation_failed" and e4["field_count"] == 2
assert set(e5) == base_keys | {"result_code"} and e5["result_code"] == "mail_sent"
assert set(e6) == base_keys | {"product_id", "inferred"}
assert e6["product_id"] == "excavator-x1" and e6["inferred"] is True
body = r.content.decode()
for leak in ("wa.me", "张三", "z@x.com", "邮箱格式不对", "250 ok", "smtp_detail",
"page_url", "fields", "messages", "label"):
assert leak not in body # 无其他 properties 键泄漏
d = client.get("/api/admin/visitors/v1/sessions/s1").json()["data"] # 同一序列化路径
sub = next(e for e in d["events"] if e["event_name"] == "form_submit")
assert sub["submission_id"] == "abcd1234"
assert "fields" not in sub
def test_timeline_days_param_and_404(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
sy = _session(site, v, "sy", at=NOW - timedelta(days=1))
_event(site, v, sy, "y1", "page_view", NOW - timedelta(days=1), path="/")
st = _session(site, v, "st", at=NOW)
_event(site, v, st, "t1", "page_view", NOW, path="/")
r = client.get("/api/admin/visitors/v1/timeline", {"days": "200"})
assert r.status_code == 200 # 超界夹紧到 90，不报错
assert [g["date"] for g in r.json()["data"]["days"]] == [TODAY.isoformat(), YESTERDAY.isoformat()]
r = client.get("/api/admin/visitors/v1/timeline", {"days": "1"})
assert [g["date"] for g in r.json()["data"]["days"]] == [TODAY.isoformat()] # 只含今天窗口
r = client.get("/api/admin/visitors/v1/timeline", {"days": "abc"})
assert r.status_code == 200 # 非法 days → 默认 30
r = client.get("/api/admin/visitors/nope/timeline")
assert r.status_code == 404
body = r.json()
assert body["code"] == 404 and body["data"] is None and body["message"] == "visitor_not_found"
def test_session_detail_ordered(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
s = _session(site, v, "s1", referrer_domain="d.com", landing_path="/start",
event_count=5, duration_ms=90000, stay_ms=45000)
base = NOW
_event(site, v, s, "e2", "page_view", base + timedelta(seconds=2), path="/a", title="A") # 乱序造数
_event(site, v, s, "e0", "page_stay", base, path="/x", active_ms=15000) # 先于任何 page_view → 独立
_event(site, v, s, "e4", "page_stay", base + timedelta(seconds=4), path="/a", active_ms=30000) # 并入 /a
_event(site, v, s, "e1", "sdk_loaded", base + timedelta(seconds=1), path="/")
_event(site, v, s, "e3", "page_stay", base + timedelta(seconds=3), path="/b", active_ms=7000) # 无 /b page_view → 独立
r = client.get("/api/admin/visitors/v1/sessions/s1")
assert r.status_code == 200
d = r.json()["data"]
assert set(d) == {"session_id", "started_at", "ended_at", "referrer_domain", "utm",
"landing_path", "event_count", "stay_ms", "events"} # duration_ms 不展示
assert d["session_id"] == "s1" and d["referrer_domain"] == "d.com"
assert d["landing_path"] == "/start" and d["event_count"] == 5
assert d["stay_ms"] == 45000
evs = d["events"]
assert [e["event_name"] for e in evs] == ["page_stay", "sdk_loaded", "page_view", "page_stay"] # occurred_at 正序
assert [e["path"] for e in evs] == ["/x", "/", "/a", "/b"]
assert [e["active_ms"] for e in evs] == [15000, None, 30000, 7000] # e4 并入 /a；/x /b 无匹配独立成条
r = client.get("/api/admin/visitors/v1/sessions/nope")
assert r.status_code == 404 and r.json()["message"] == "session_not_found"
def test_timeline_truncates_at_1000(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
s = _session(site, v, "s1", at=NOW - timedelta(hours=3))
base = NOW - timedelta(hours=3) # 跨不跨天皆可，按 path 集合断言不依赖分组形态
Event.objects.bulk_create([Event(site=site, visitor=v, session=s,
event_id=f"e{i}", event_name="page_view",
occurred_at=base + timedelta(seconds=i), path=f"/p{i}") for i in range(1001)])
r = client.get("/api/admin/visitors/v1/timeline")
assert r.status_code == 200
d = r.json()["data"]
assert d["truncated"] is True # 触上限（前端可区分"没动静"和"被截断"）
paths = [e["path"] for g in d["days"] for sess in g["sessions"] for e in sess["events"]]
assert len(paths) == 1000 # 条数封顶
assert "/p0" not in paths # 最旧一条被截掉
assert "/p1000" in paths # 最新保留
def test_timeline_orphan_events_without_session(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
_event(site, v, None, "o1", "sdk_loaded", NOW, path="/") # session=NULL 孤立事件
_event(site, v, None, "o2", "page_stay", NOW + timedelta(seconds=1), path="/x", active_ms=100)
r = client.get("/api/admin/visitors/v1/timeline")
assert r.status_code == 200 # 不崩溃，归入 session_id=null 桶
sess = r.json()["data"]["days"][0]["sessions"][0]
assert sess["session_id"] is None and sess["started_at"] is None
assert [e["event_name"] for e in sess["events"]] == ["sdk_loaded", "page_stay"]
def _classify(site, visitor, scope, channel, platform="", session=None, **kw):
return SourceClassification.objects.create(
site=site, visitor=visitor, session=session, scope=scope,
channel=channel, platform=platform, **kw)
def test_list_attribution_fields_and_channel_filter(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v1 = _visitor(site, "v1", last_seen=timezone.now())
v2 = _visitor(site, "v2", last_seen=timezone.now() - timedelta(hours=1))
v3 = _visitor(site, "v3", last_seen=timezone.now() - timedelta(hours=2)) # 无分类行
_classify(site, v1, "first_touch", "natural_search", "Bing",
search_term="挖掘机", evidence_kind="observed_referrer",
resolution="resolved")
_classify(site, v2, "first_touch", "ai_referral", "ChatGPT",
evidence_kind="observed_referrer", resolution="resolved")
d = client.get("/api/admin/visitors").json()["data"]
by_id = {x["visitor_id"]: x for x in d["list"]}
assert by_id["v1"]["channel"] == "natural_search"
assert by_id["v1"]["platform"] == "Bing" and by_id["v1"]["search_term"] == "挖掘机"
assert by_id["v1"]["variant"] == "" and by_id["v1"]["resolution"] == "resolved"
assert by_id["v3"]["channel"] == "" # 无分类行 → 空串（前端回落 first_source 展示）
d = client.get("/api/admin/visitors", {"channel": "natural_search"}).json()["data"]
assert d["total"] == 1 and [x["visitor_id"] for x in d["list"]] == ["v1"]
d = client.get("/api/admin/visitors", {"channel": "direct"}).json()["data"]
assert d["total"] == 0 # 无分类访客不落进任何渠道桶
d = client.get("/api/admin/visitors", {"channel": "ai_referral"}).json()["data"]
assert [x["visitor_id"] for x in d["list"]] == ["v2"]
def test_detail_attribution_block(client, member_user):
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
s1 = _session(site, v, "s1", at=NOW - timedelta(hours=2),
referrer_domain="www.bing.com")
s2 = _session(site, v, "s2", at=NOW, referrer_domain="chatgpt.com")
_classify(site, v, "first_touch", "natural_search", "Bing",
search_term="挖掘机", evidence_kind="observed_referrer",
resolution="resolved")
_classify(site, v, "session_entry", "natural_search", "Bing",
session=s1, evidence_kind="observed_referrer", resolution="resolved")
_classify(site, v, "session_entry", "ai_referral", "ChatGPT",
session=s2, evidence_kind="observed_referrer", resolution="resolved")
r = client.get("/api/admin/visitors/v1")
assert r.status_code == 200
attr = r.json()["data"]["attribution"]
ft = attr["first_touch"]
assert ft["channel"] == "natural_search" and ft["platform"] == "Bing"
assert ft["search_term"] == "挖掘机" and ft["resolution"] == "resolved"
assert ft["evidence_kind"] == "observed_referrer" and ft["rule_version"] == "v2"
assert ft["computed_at"] and ft["detail"] == {}
assert [e["session_id"] for e in attr["session_entries"]] == ["s2", "s1"] # 最近在前
assert attr["session_entries"][0]["channel"] == "ai_referral"
def test_detail_source_evidence_triplet(client, member_user):
"""三口径：首日 direct（不可识别）+ 次日 Google——first_observable=direct/unknown，
first_identifiable 回溯到次日的可识别会话入口，latest_session=最新会话入口。"""
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
s1 = _session(site, v, "s1", at=NOW - timedelta(days=1)) # 首日直接访问
s2 = _session(site, v, "s2", at=NOW, referrer_domain="google.com") # 次日 Google
_classify(site, v, "first_touch", "direct", resolution="unknown")
_classify(site, v, "session_entry", "direct", session=s1, resolution="unknown")
_classify(site, v, "session_entry", "natural_search", "Google", session=s2,
evidence_kind="observed_referrer", resolution="resolved")
r = client.get("/api/admin/visitors/v1")
assert r.status_code == 200
attr = r.json()["data"]["attribution"]
fo = attr["first_observable"]
assert fo["channel"] == "direct" and fo["resolution"] == "unknown" # 首访可观测=直接
fi = attr["first_identifiable"]
assert fi["channel"] == "natural_search" and fi["platform"] == "Google"
assert fi["resolution"] == "resolved" and fi["session_id"] == "s2" # 回溯到次日入口
ls = attr["latest_session"]
assert ls["channel"] == "natural_search" and ls["session_id"] == "s2"
def test_detail_source_evidence_no_classification(client, member_user):
"""历史数据无任何分类行：first_observable 给 unknown 占位，其余两口径为 null。"""
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
v = _visitor(site, "v1")
_session(site, v, "s1", referrer_domain="google.com")
r = client.get("/api/admin/visitors/v1")
assert r.status_code == 200
attr = r.json()["data"]["attribution"]
assert attr["first_touch"] is None
assert attr["first_observable"] == {"channel": "unknown", "resolution": "unknown"}
assert attr["first_identifiable"] is None
assert attr["latest_session"] is None
def test_device_model_and_env_extra_fields(client, member_user):
"""设备型号/扩展环境快照：新数据随行输出；旧访客空串/空对象（前端逐键判空跳行）。"""
_auth(client, member_user)
site = Site.objects.create(name="s", domains=["a.com"])
_visitor(site, "v-new", last_seen=timezone.now(), device_model="Pixel 7",
env_extra={"viewport": "393x851", "device_pixel_ratio": "2.625",
"color_scheme": "dark", "network_downlink": "10",
"network_rtt": "50", "save_data": "0", "wechat_version": "8.0.49"})
_visitor(site, "v-old", last_seen=timezone.now() - timedelta(hours=1))
d = client.get("/api/admin/visitors").json()["data"]
by_id = {x["visitor_id"]: x for x in d["list"]}
assert by_id["v-new"]["device_model"] == "Pixel 7" # 列表行随带（设备列小字用）
assert by_id["v-old"]["device_model"] == ""
d = client.get("/api/admin/visitors/v-new").json()["data"]
assert d["device_model"] == "Pixel 7"
assert d["env_extra"]["viewport"] == "393x851"
assert d["env_extra"]["network_rtt"] == "50"
d = client.get("/api/admin/visitors/v-old").json()["data"]
assert d["device_model"] == "" and d["env_extra"] == {}
def test_mixed_v1_v2_stay_totals_and_detail(client, member_user):
"""v1+v2 混合会话回归（阶段A T4）：total_stay_ms/会话 stay_ms 均读 VisitSession.stay_ms
重算值（v1 active_ms=8000 加法 + v2 同 page_id 区间 [0,15000][15000,30000] 并集 30000
= 38000）；详情/单会话/时间线对多条 page_stay 事件正常返回不崩——v2 区间事件
properties 无 active_ms，时间线合并展示为 0 累加（展示层不炸、不虚增）。"""
_auth(client, member_user)
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": 8000, "path": "/p"})], sid="sm"))
ingest_batch(site, _v2_batch([
_v2_event("page_stay", "p1", "sm", seq=1,
active_from=0, active_to=15000, flush_reason="periodic"),
_v2_event("page_stay", "p1", "sm", seq=2,
active_from=15000, active_to=30000, flush_reason="periodic")],
sid="sm"))
sess = VisitSession.objects.get(session_id="sm")
assert sess.stay_ms == 38000 and sess.event_count == 4 # 写侧：v1 SUM + v2 并集
d = client.get("/api/admin/visitors").json()["data"] # 列表：Sum(sessions__stay_ms)
assert d["total"] == 1
assert d["list"][0]["total_stay_ms"] == 38000 and d["list"][0]["session_count"] == 1
d = client.get("/api/admin/visitors/v1").json()["data"] # 详情：行字段 + 会话摘要
assert d["total_stay_ms"] == 38000
assert [s["stay_ms"] for s in d["sessions"]] == [38000]
d = client.get("/api/admin/visitors/v1/sessions/sm").json()["data"] # 单会话：多条 page_stay 正常
assert d["stay_ms"] == 38000 and d["event_count"] == 4
# 4 条事件（1 page_view + 3 page_stay）合并成一条：v1 的 8000 并入 page_view，
# v2 两段区间事件无 active_ms → 展示层累加 0（真实停留看 stay_ms=38000）
assert [(e["event_name"], e["path"], e["active_ms"]) for e in d["events"]] == \
[("page_view", "/p", 8000)]
r = client.get("/api/admin/visitors/v1/timeline") # 时间线：同款合并不崩
assert r.status_code == 200
evs = r.json()["data"]["days"][0]["sessions"][0]["events"]
assert [(e["event_name"], e["path"], e["active_ms"]) for e in evs] == \
[("page_view", "/p", 8000)]
Reference in New Issue View Git Blame Copy Permalink