浅念
133 lines
6.8 KiB
Python
133 lines
6.8 KiB
Python
import json
|
|
import pathlib
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
from django.utils import timezone
|
|
from tracker import constants as C
|
|
from tracker.services.validation import validate_event, sanitize_url
|
|
|
|
VECTORS = json.loads((pathlib.Path(__file__).parent / "fixtures" / "sanitize_vectors.json").read_text(encoding="utf-8"))
|
|
|
|
NOW = timezone.now()
|
|
|
|
def _ev(**kw):
|
|
base = dict(event_id="e1", event_name="page_view",
|
|
occurred_at=NOW, path="/p", title="t", properties={})
|
|
base.update(kw)
|
|
return base
|
|
|
|
def test_ok_event():
|
|
assert validate_event(_ev()) is None # None = 通过
|
|
|
|
def test_unknown_event_name():
|
|
assert validate_event(_ev(event_name="xxx")) == "event_name_not_allowed"
|
|
|
|
def test_sensitive_property_rejected():
|
|
assert validate_event(_ev(properties={"user_password": "1"})) == "sensitive_field"
|
|
|
|
def test_shape_type_and_whitelist():
|
|
assert validate_event(["x"]) == "event_invalid_shape"
|
|
assert validate_event(_ev(event_id=123)) == "invalid_event_id"
|
|
assert validate_event(_ev(properties={"foo": "bar"})) == "property_not_allowed"
|
|
|
|
def test_too_many_props_and_long_string():
|
|
assert validate_event(_ev(properties={f"k{i}": "v" for i in range(40)})) == "properties_invalid"
|
|
assert validate_event(_ev(properties={"referrer": "x" * 600})) == "properties_too_long"
|
|
|
|
def test_future_and_stale_time():
|
|
assert validate_event(_ev(occurred_at=NOW + timedelta(days=2))) == "occurred_at_out_of_window" # 超窗拒绝
|
|
assert validate_event(_ev(occurred_at=NOW - timedelta(hours=2))) is None # 窗内迟到可收
|
|
|
|
def test_sanitize_url_drops_sensitive_params():
|
|
u = sanitize_url("/p?email=a@b.com&token=x&id=3&utm_source=google&monkey=1")
|
|
assert "email=" not in u and "token=" not in u and "utm_source=google" in u
|
|
assert "monkey=1" in u # fullmatch：含 key 子串的正常参数不误伤
|
|
|
|
|
|
def test_pageview_referrer_url_sanitized():
|
|
"""page_view 的 properties.referrer 是完整 URL：敏感参数整键剔除后写回。"""
|
|
ev = _ev(properties={"referrer": "https://www.bing.com/search?q=x&email=a@b.com&token=t"})
|
|
assert validate_event(ev) is None
|
|
assert ev["properties"]["referrer"] == "https://www.bing.com/search?q=x"
|
|
|
|
|
|
def test_pageview_referrer_truncated_after_reencode():
|
|
"""清洗重编码（中文 → %XX 九倍膨胀）可能超出原长：出门统一截 MAX_STR_LEN=500。
|
|
用 utm 值驱动：utm 参数三轨制下无条件保留（不走值判别），膨胀路径仍可达。"""
|
|
ev = _ev(properties={"referrer": "https://www.bing.com/search?utm_term=" + "词" * 200})
|
|
assert validate_event(ev) is None # 原文 237 字符过长度门
|
|
ref = ev["properties"]["referrer"]
|
|
assert len(ref) == 500 and ref.startswith("https://www.bing.com/search?utm_term=%E8%AF%8D")
|
|
|
|
def test_nested_sensitive_key_rejected():
|
|
assert validate_event(_ev(properties={"utm": {"utm_a": 1, "user_password": "x"}})) == "sensitive_field" # utm 在白名单内，嵌套敏感键仍须命中
|
|
|
|
def test_naive_datetime_rejected():
|
|
assert validate_event(_ev(occurred_at=NOW.replace(tzinfo=None))) == "invalid_occurred_at"
|
|
|
|
def test_prop_size_gate_counts_bytes():
|
|
r = validate_event(_ev(properties={"utm": {"a": "汉" * 1500}}))
|
|
assert r == "properties_too_large" # 字符数约 2K 未超，UTF-8 约 4.5KB 超 4096，须按字节算
|
|
|
|
def test_nested_string_length_capped():
|
|
assert validate_event(_ev(properties={"utm": {"a": "x" * 600}})) == "properties_too_long"
|
|
|
|
def test_falsy_non_dict_properties_rejected():
|
|
assert validate_event(_ev(properties=[])) == "properties_invalid"
|
|
|
|
def test_lone_surrogate_event_id_rejected():
|
|
assert validate_event(_ev(event_id="\ud800")) == "invalid_event_id"
|
|
|
|
def test_lone_surrogate_field_rejected():
|
|
assert validate_event(_ev(path="/p\ud800")) == "invalid_field_type"
|
|
assert validate_event(_ev(title="t\ud800")) == "invalid_field_type"
|
|
|
|
def test_lone_surrogate_props_rejected():
|
|
assert validate_event(_ev(properties={"utm": {"a": "\ud800"}})) == "properties_invalid"
|
|
|
|
def test_non_str_path_rejected():
|
|
assert validate_event(_ev(path=123)) == "invalid_field_type"
|
|
|
|
def test_prop_depth_exceeded():
|
|
assert validate_event(_ev(properties={"utm": {"a": {"b": 1}}})) == "properties_depth"
|
|
|
|
def test_sanitize_url_vectors():
|
|
for v in VECTORS["vectors"]:
|
|
assert sanitize_url(v["in"]) == v["out"], v["in"]
|
|
|
|
def test_sanitize_rule_version_exported():
|
|
assert C.SANITIZE_RULE_VERSION == VECTORS["rule_version"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 业务事件契约（spec §1/§2）：+6 事件白名单与属性枚举/格式校验
|
|
# ---------------------------------------------------------------------------
|
|
SITE_ID = "site_test" # validate_event 不查 DB，本地定义即可
|
|
|
|
def _biz(name, **props):
|
|
return {"event_id": f"biz-{name}", "event_name": name, "occurred_at": timezone.now(),
|
|
"path": "/p", "title": "t", "site_id": SITE_ID, "visitor_id": "v1",
|
|
"session_id": "s1", "page_id": "p1", "seq": 1, "properties": props}
|
|
|
|
@pytest.mark.parametrize("ev,expect_reason", [
|
|
(_biz("contact_click", channel="whatsapp"), None),
|
|
(_biz("contact_click", channel="im"), "property_not_allowed"),
|
|
(_biz("form_start", form_id="cf7-156"), None),
|
|
(_biz("form_start", form_id="wpcf7-f156-p20"), "property_not_allowed"),
|
|
(_biz("form_submit", form_id="cf7-156", submission_id="9c8b7a6d-1111-4222-8333-444455556666"), None),
|
|
(_biz("form_submit", form_id="cf7-156", submission_id=""), None),
|
|
(_biz("form_submit", form_id="cf7-156", submission_id="not-uuid"), "property_not_allowed"),
|
|
(_biz("form_error", form_id="cf7-156", submission_id="", error_code="invalid", field_count=2), None),
|
|
(_biz("form_error", form_id="cf7-156", submission_id="", error_code="bogus"), "property_not_allowed"),
|
|
(_biz("form_error", form_id="cf7-156", submission_id="", error_code="invalid", field_count=-1), "property_not_allowed"),
|
|
(_biz("form_submit_result", form_id="cf7-156", submission_id="", result_code="mail_sent"), None),
|
|
(_biz("form_submit_result", form_id="cf7-156", submission_id="", result_code="ok"), "property_not_allowed"),
|
|
(_biz("product_view", product_id="byd-equation-leopard-5", inferred=True, rule_version="v1"), None),
|
|
(_biz("product_view", product_id="x"*101, inferred=True, rule_version="v1"), "property_not_allowed"),
|
|
# 尾换行脏值：$ 锚匹配串尾换行前，\Z 才是真正串尾——collect 端点公开，须拒
|
|
(_biz("form_start", form_id="cf7-156\n"), "property_not_allowed"),
|
|
(_biz("form_submit", form_id="cf7-156", submission_id="9c8b7a6d-1111-4222-8333-444455556666\n"), "property_not_allowed"),
|
|
])
|
|
def test_biz_event_validation(ev, expect_reason):
|
|
assert validate_event(ev, v2=True, batch_site_id=SITE_ID, batch_visitor_id="v1") == expect_reason
|