浅念
128 lines
6.1 KiB
Python
128 lines
6.1 KiB
Python
import json
|
||
import pytest
|
||
from django.core.cache import cache
|
||
from django.utils import timezone
|
||
from tracker.models import Site
|
||
|
||
NOW = timezone.now()
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _clear_cache():
|
||
cache.clear() # locmem 限流历史跨测试残留，会让 429 用例假红
|
||
|
||
def _post(client, body, origin="https://www.a.com", extra=None):
|
||
return client.post("/api/collect/batch", data=json.dumps(body),
|
||
content_type="text/plain", HTTP_ORIGIN=origin, **(extra or {}))
|
||
|
||
def _body(n=1, sid="s1", vid="v1"):
|
||
return {"site_id": "site_x", "visitor_id": vid, "session_id": sid,
|
||
"context": {"referrer_domain": "", "path": "/", "device_type": "PC"},
|
||
"events": [{"event_id": f"e{i}", "event_name": "page_view",
|
||
"occurred_at": NOW.isoformat(), "path": "/", "title": "t",
|
||
"properties": {}} for i in range(n)]}
|
||
|
||
def test_collect_ok_creates_events(db, client):
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
r = _post(client, _body(2))
|
||
assert r.status_code == 200
|
||
body = r.json() # 包含式断言：v2 响应新增 rejected/request_id/server_time 字段
|
||
assert body["code"] == 200 and body["message"] == "ok"
|
||
assert body["data"]["accepted"] == 2 and body["data"]["duplicated"] == 0
|
||
assert body["data"]["rejected"] == [] and body["data"]["request_id"]
|
||
assert r["Access-Control-Allow-Origin"] == "https://www.a.com"
|
||
|
||
def test_origin_not_whitelisted_rejected(db, client):
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
r = _post(client, _body(1), origin="https://evil.com")
|
||
assert r.status_code == 403 and r.json()["code"] == 403
|
||
assert r.json()["message"] == "not_allowed" # 与未知站点同 message，防枚举
|
||
|
||
def test_inactive_site_rejected(db, client):
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x", is_active=False)
|
||
assert _post(client, _body(1)).status_code == 403
|
||
|
||
def test_unknown_site_403_not_404(db, client): # 不泄露站点是否存在
|
||
r = _post(client, _body(1))
|
||
assert r.status_code == 403
|
||
assert r.json()["message"] == "not_allowed" # message 与上例一致
|
||
|
||
def test_replay_duplicated(db, client):
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
_post(client, _body(1))
|
||
r = _post(client, _body(1))
|
||
assert r.json()["data"]["duplicated"] == 1
|
||
|
||
def test_invalid_json_400_not_500(db, client):
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
r = client.post("/api/collect/batch", data="{bad json",
|
||
content_type="text/plain", HTTP_ORIGIN="https://www.a.com")
|
||
assert r.status_code == 400
|
||
|
||
def test_throttle_429(db, client, settings):
|
||
settings.REST_FRAMEWORK = {**settings.REST_FRAMEWORK,
|
||
"DEFAULT_THROTTLE_RATES": {"collect": "3/min"}}
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
for _ in range(3):
|
||
assert _post(client, _body(1)).status_code == 200
|
||
assert _post(client, _body(1)).status_code == 429
|
||
|
||
# ---- 审查修复锁定测试 ----
|
||
|
||
def test_throttle_429_envelope_and_acao(db, client, settings):
|
||
"""429 走 finalize_response：信封 + ACAO——否则浏览器 SDK 读不到错误体误判网络错误重试，反而打爆限流器。"""
|
||
settings.REST_FRAMEWORK = {**settings.REST_FRAMEWORK,
|
||
"DEFAULT_THROTTLE_RATES": {"collect": "3/min"}}
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
for _ in range(3):
|
||
assert _post(client, _body(1)).status_code == 200
|
||
r = _post(client, _body(1))
|
||
assert r.status_code == 429
|
||
assert r["Access-Control-Allow-Origin"] == "https://www.a.com"
|
||
body = r.json()
|
||
assert body["code"] == 429 and body["data"] is None and body["message"]
|
||
|
||
def test_invalid_json_envelope_and_acao(db, client):
|
||
"""ParseError 400 由 DRF 默认异常处理器产出 {detail}——finalize_response 统一转信封并补 ACAO。"""
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
r = client.post("/api/collect/batch", data="{bad json",
|
||
content_type="text/plain", HTTP_ORIGIN="https://www.a.com")
|
||
assert r.status_code == 400
|
||
assert r.json() == {"code": 400, "data": None, "message": "invalid_json"}
|
||
assert r["Access-Control-Allow-Origin"] == "https://www.a.com"
|
||
|
||
def test_origin_url_in_query_bypass_blocked(db, client):
|
||
"""split("//")[-1] 提取 host 会被查询串里的白名单域名绕过——urlsplit 后必 403。"""
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
r = _post(client, _body(1), origin="https://evil.com/?x=https://www.a.com")
|
||
assert r.status_code == 403 and r.json()["message"] == "not_allowed"
|
||
|
||
def test_malformed_content_length_400_not_500(db, client):
|
||
"""Content-Length 非数字（"abc"）：isdigit/isdecimal 防御，拒 400 而非 int() 崩 500。"""
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
r = _post(client, _body(1), extra={"CONTENT_LENGTH": "abc"})
|
||
assert r.status_code == 400 and r.json()["message"] == "payload_too_large"
|
||
|
||
|
||
# ---- 响应契约（阶段A Task3）----
|
||
|
||
def test_response_contract_rejected_details(db, client):
|
||
"""部分拒绝批：rejected 明细（index/event_id/reason）回传——HTTP 200 不再隐含全量成功。"""
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
b = _body(2)
|
||
b["events"][0]["event_name"] = "bad_name"
|
||
r = _post(client, b)
|
||
assert r.status_code == 200
|
||
data = r.json()["data"]
|
||
assert data["accepted"] == 1 and data["duplicated"] == 0
|
||
assert data["rejected"] == [{"index": 0, "event_id": "e0",
|
||
"reason": "event_name_not_allowed"}]
|
||
|
||
def test_request_id_in_response_and_raw_batch(db, client):
|
||
"""request_id（uuid4 hex 32）随响应回传，且与 RawBatch.request_id 同值落库（对账）。"""
|
||
from tracker.models import RawBatch
|
||
Site.objects.create(name="s", domains=["www.a.com"], site_id="site_x")
|
||
r = _post(client, _body(1))
|
||
rid = r.json()["data"]["request_id"]
|
||
assert isinstance(rid, str) and len(rid) == 32
|
||
assert RawBatch.objects.get().request_id == rid
|