fangke/server/tracker/tests/test_dashboard.py

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
from tracker.services.stats import refresh_daily

pytestmark = pytest.mark.django_db
User = get_user_model()

# 用上海本地时刻做基准：日界是上海自然日（make_aware/localdate），
# 直接用 UTC 的 now 在本地 00:00-08:00 之间跑测试会错一天
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):
    v = Visitor.objects.create(site=site, visitor_id=vid)
    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, referrer="", at=NOW):
    return VisitSession.objects.create(site=site, visitor=visitor, session_id=sid,
        started_at=at, ended_at=at + timedelta(minutes=5), referrer_domain=referrer)


def _events(site, visitor, session, n, path, at=NOW, prefix="e"):
    for i in range(n):
        Event.objects.create(site=site, visitor=visitor, session=session,
            event_id=f"{prefix}-{i}", event_name="page_view",
            occurred_at=at + timedelta(seconds=i), path=path)


def test_overview_requires_auth(client):
    assert client.get("/api/admin/dashboard/overview").status_code == 401


def test_overview_shape_and_numbers(client, member_user):
    _auth(client, member_user)
    site = Site.objects.create(name="s", domains=["a.com"])
    v1 = _visitor(site, "v1", last_seen=timezone.now())                    # 活跃（5 分钟内）
    v2 = _visitor(site, "v2", last_seen=timezone.now() - timedelta(hours=1))  # 不活跃
    s1 = _session(site, v1, "s1", referrer="google.com")
    s2 = _session(site, v2, "s2", referrer="google.com")
    s3 = _session(site, v1, "s3", referrer="")                              # 空来源 → 直接访问
    _events(site, v1, s1, 6, "/pricing", prefix="p")                       # /pricing x6
    _events(site, v1, s3, 2, "/", prefix="r")                              # / x2
    _events(site, v2, s2, 2, "/docs", prefix="d")                          # /docs x2

    r = client.get("/api/admin/dashboard/overview")
    assert r.status_code == 200
    d = r.json()["data"]
    assert d["today"] == {"pv": 10, "uv": 2, "sessions": 3,
                          "new_visitors": 2, "active_now": 1}
    trend = d["trend"]
    assert len(trend) == 30
    assert trend[-1]["date"] == TODAY.isoformat()
    assert trend[-1]["pv"] == 10            # 今天位用实时值
    assert trend[-2]["pv"] == 0             # 昨天无 DailyStat → 补零
    assert d["sources"][0] == {"source": "google.com", "sessions": 2}
    assert d["sources"][-1] == {"source": "直接访问", "sessions": 1}
    assert d["top_pages"][0] == {"path": "/pricing", "pv": 6}


def test_overview_trend_uses_dailystat_history(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"])
    yn = NOW - timedelta(days=1)   # 昨天同一本地时刻，必落上海昨日窗口
    # a 站昨天 3 事件走真实 refresh_daily 落 DailyStat
    va = Visitor.objects.create(site=site_a, visitor_id="va")
    sa = _session(site_a, va, "sa", at=yn)
    _events(site_a, va, sa, 3, "/", at=yn, prefix="ya")
    refresh_daily(site_a, YESTERDAY)
    # b 站昨天 4 事件同样走 refresh_daily——两站各落一行，验证 site=None 的跨站 Sum
    vb = Visitor.objects.create(site=site_b, visitor_id="vb")
    sb = _session(site_b, vb, "sb", at=yn)
    _events(site_b, vb, sb, 4, "/", at=yn, prefix="yb")
    refresh_daily(site_b, YESTERDAY)

    d = client.get("/api/admin/dashboard/overview").json()["data"]
    assert d["trend"][-2] == {"date": YESTERDAY.isoformat(), "pv": 7, "uv": 2,
                              "sessions": 2, "new_visitors": 0}   # 3+4 跨站点求和
    assert d["trend"][-1]["pv"] == 0   # 今天无事件，实时覆盖为 0


def test_overview_site_filter_and_404(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"])
    va = _visitor(site_a, "va")
    vb = _visitor(site_b, "vb")
    sa = _session(site_a, va, "sa", referrer="google.com")
    sb = _session(site_b, vb, "sb", referrer="")
    _events(site_a, va, sa, 2, "/", prefix="a")
    _events(site_b, vb, sb, 5, "/b", prefix="b")
    # a 站昨天有 DailyStat 行——过滤 site_b 时不得混入
    yn = NOW - timedelta(days=1)
    _events(site_a, va, sa, 3, "/", at=yn, prefix="ay")
    refresh_daily(site_a, YESTERDAY)

    d = client.get("/api/admin/dashboard/overview").json()["data"]
    assert d["today"]["pv"] == 7           # 缺省聚合全部站点
    d = client.get("/api/admin/dashboard/overview", {"site": site_b.site_id}).json()["data"]
    assert d["today"]["pv"] == 5 and d["today"]["uv"] == 1
    assert d["today"]["active_now"] == 1     # 两访客 last_seen 均为 now，但只计本站
    assert d["sources"] == [{"source": "直接访问", "sessions": 1}]
    assert d["top_pages"][0] == {"path": "/b", "pv": 5}
    assert d["trend"][-2]["pv"] == 0       # a 站昨日行被站点过滤排除
    assert d["trend"][-1]["pv"] == 5       # 今天位仍是本站实时值

    r = client.get("/api/admin/dashboard/overview", {"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"