fangke/server/tracker/tests/test_validation.py

from datetime import datetime, timedelta
from django.utils import timezone
from tracker.services.validation import validate_event, sanitize_url

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。"""
    ev = _ev(properties={"referrer": "https://www.bing.com/search?q=" + "词" * 200})
    assert validate_event(ev) is None              # 原文 227 字符过长度门
    ref = ev["properties"]["referrer"]
    assert len(ref) == 500 and ref.startswith("https://www.bing.com/search?q=%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"