浅念
293 lines
16 KiB
Python
293 lines
16 KiB
Python
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, VisitSession, Visitor
|
||
|
||
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):
|
||
return Event.objects.create(site=site, visitor=visitor, session=session,
|
||
event_id=eid, event_name=name, occurred_at=at, path=path, title=title,
|
||
properties={"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", 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", "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", "online"}
|
||
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["online"] is True
|
||
row2 = d["list"][1]
|
||
assert row2["first_source"] == "baidu" # first_referrer 空 → first_utm 兜底
|
||
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"]
|
||
assert ids(d) == ["old-visitor"] # last_seen 落在昨天
|
||
|
||
|
||
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")
|
||
_session(site, v, "s-old", at=NOW - timedelta(hours=2),
|
||
referrer_domain="google.com", stay_ms=60000, duration_ms=300000, event_count=3)
|
||
_session(site, v, "s-new", at=NOW, utm={"utm_source": "wechat"},
|
||
landing_path="/new", stay_ms=30000, duration_ms=120000, event_count=2)
|
||
|
||
r = client.get("/api/admin/visitors/v1")
|
||
assert r.status_code == 200
|
||
d = r.json()["data"]
|
||
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 [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_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"]
|