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

167 lines
6.8 KiB
Python
Raw Normal View History

"""原始上报批次(raw_batch):脱敏快照、拒批截断、重放留痕、API 分页、30 天清理。
设计契约(与用户确认):
- 敏感字段值替换为 [FILTERED](结构保留)——sanitize_payload 复用 SENSITIVE_KEY_PATTERNS;
- 被拒批次也存(body 截断 4KB);invalid_json(解析失败)不存;
- 原始表写入失败不得影响主采集流程(best-effort);
- 每次接收都记一行("原始记录"语义,重放不合并)。
"""
import json
from datetime import timedelta
from io import StringIO
import pytest
from django.contrib.auth import get_user_model
from django.core.cache import cache
from django.core.management import call_command
from django.utils import timezone
from rest_framework.test import APIClient
from tracker.models import RawBatch, Site, Visitor
User = get_user_model()
NOW = timezone.now()
@pytest.fixture(autouse=True)
def _clear_cache():
cache.clear() # locmem 限流历史跨测试残留,会让压测 429(与 test_collect 同款)
@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 _post(client, body, origin="https://www.a.com"):
return client.post("/api/collect/batch", data=json.dumps(body),
content_type="text/plain", HTTP_ORIGIN=origin)
def _body(n=1, sid="s1", vid="v1", events=None):
return {"site_id": "site_x", "visitor_id": vid, "session_id": sid,
"context": {"referrer_domain": "", "path": "/", "device_type": "PC"},
"events": events if events is not None else
[{"event_id": f"e{i}", "event_name": "page_view",
"occurred_at": NOW.isoformat(), "path": "/", "title": "t",
"properties": {}} for i in range(n)]}
def _site():
return Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
# ---- a) 正常上报 ----
def test_raw_batch_accepted(db, client):
_site()
r = _post(client, _body(2, vid="v9"))
assert r.status_code == 200
rb = RawBatch.objects.get()
assert rb.status == "accepted" and rb.event_count == 2 and rb.rejected_count == 0
vis = Visitor.objects.get(visitor_id="v9")
assert rb.visitor_id == vis.pk and rb.site_id == vis.site_id # 访客关联
body = json.loads(rb.body)
assert body["visitor_id"] == "v9" # 完整结构在场
assert body["context"]["device_type"] == "PC"
assert [e["path"] for e in body["events"]] == ["/", "/"] # 快照先于 ingest 就地清洗
# ---- b) 敏感键脱敏 + partial ----
def test_raw_batch_partial_and_filtered(db, client):
_site()
events = [
{"event_id": "ok1", "event_name": "page_view", "occurred_at": NOW.isoformat(),
"path": "/", "title": "t", "properties": {}},
{"event_id": "bad1", "event_name": "page_view", "occurred_at": NOW.isoformat(),
"path": "/login", "title": "t", "properties": {"user_password": "hunter2"}},
]
r = _post(client, _body(events=events, vid="v1"))
assert r.json()["data"]["accepted"] == 1 # 事件层:敏感事件被拒
rb = RawBatch.objects.get()
assert rb.status == "partial" and rb.event_count == 1 and rb.rejected_count == 1
body = json.loads(rb.body)
assert body["events"][1]["properties"]["user_password"] == "[FILTERED]" # 值替换结构保留
assert body["events"][1]["properties"] != {"user_password": "hunter2"}
# ---- c) 超批 rejected + 截断 ----
def test_raw_batch_rejected_truncated(db, client):
_site()
r = _post(client, _body(51, vid="v1")) # >50 事件整批拒
assert r.json()["data"]["accepted"] == 0
rb = RawBatch.objects.get()
assert rb.status == "rejected" and rb.event_count == 0 and rb.rejected_count == 1
# 51 事件原始 JSON 远超 4KB → 截断到 4096 + 标记后缀(≤ ~4.1KB)
assert len(rb.body) == 4096 + len("...[TRUNCATED]")
assert rb.body.endswith("...[TRUNCATED]")
assert rb.visitor is None or rb.visitor.visitor_id == "v1" # 超批不建访客,允许 None
# ---- d) 重放每次都记 ----
def test_raw_batch_replay_two_rows(db, client):
_site()
_post(client, _body(1, vid="v1"))
_post(client, _body(1, vid="v1"))
assert RawBatch.objects.count() == 2 # 原始记录语义:每次接收一行
# ---- e) 管理台 API:分页与字段 ----
def test_raw_batches_api_pagination_and_fields(db, client, member_user):
_auth(client, member_user)
site = _site()
v = Visitor.objects.create(site=site, visitor_id="v1")
for i in range(15):
rb = RawBatch.objects.create(site=site, visitor=v, batch_id=f"b{i:02d}",
body='{"visitor_id":"v1"}', event_count=i % 3,
rejected_count=0, status="accepted", ip="1.2.3.4")
# received_at 是 auto_now_add,update 绕开制造时间差(与 test_visitors 同款手法)
RawBatch.objects.filter(pk=rb.pk).update(received_at=NOW - timedelta(minutes=i))
r = client.get("/api/admin/visitors/v1/raw-batches")
assert r.status_code == 200
d = r.json()["data"]
assert d["total"] == 15 and d["page"] == 1 and len(d["list"]) == 10 # page_size 10
row = d["list"][0]
assert set(row) == {"id", "batch_id", "status", "event_count",
"rejected_count", "received_at", "body"}
assert row["batch_id"] == "b00" # -received_at:最新在前
d2 = client.get("/api/admin/visitors/v1/raw-batches", {"page": "2"}).json()["data"]
assert d2["page"] == 2 and len(d2["list"]) == 5
assert [x["batch_id"] for x in d2["list"]][-1] == "b14"
r = client.get("/api/admin/visitors/nope/raw-batches")
assert r.status_code == 404
body = r.json()
assert body["code"] == 404 and body["data"] is None and body["message"] == "visitor_not_found"
# ---- f) 清理命令 ----
def test_cleanup_raw_command(db):
site = _site()
v = Visitor.objects.create(site=site, visitor_id="v1")
old = RawBatch.objects.create(site=site, visitor=v, body="{}")
RawBatch.objects.filter(pk=old.pk).update(received_at=NOW - timedelta(days=31))
new = RawBatch.objects.create(site=site, visitor=v, body="{}") # 刚收到
out = StringIO()
call_command("cleanup_raw", "--days", 30, stdout=out)
assert not RawBatch.objects.filter(pk=old.pk).exists() # 30 天前删除
assert RawBatch.objects.filter(pk=new.pk).exists() # 新的保留
assert "1" in out.getvalue() # 输出删除数