fangke/server/tracker/tests/test_sites.py

import pytest
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient
from tracker.models import Site

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


@pytest.fixture
def client():   # 覆盖 pytest-django 的 Django Client——认证用例需要 credentials()/format="json"
    return APIClient()


@pytest.fixture
def admin_user():
    return User.objects.create_user(username="admin", password="adminpass123", is_staff=True)


@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 test_sites_require_auth(client):
    assert client.get("/api/admin/sites").status_code == 401


def test_create_site_normalizes_domains(client, member_user):
    _auth(client, member_user)
    r = client.post("/api/admin/sites",
        {"name": "官网", "domains": ["https://Example.com/", "example.com", "  bad.com  ",
                                     "http://Sub.Example.com:8443/path?q=1"]}, format="json")
    assert r.status_code == 200 and r.json()["code"] == 200
    d = r.json()["data"]
    assert d["domains"] == ["example.com", "bad.com", "sub.example.com"]   # 去协议/端口/路径/去重/strip/小写
    assert d["site_id"].startswith("site_") and d["is_active"] is True
    assert Site.objects.filter(site_id=d["site_id"]).exists()   # 真落库


def test_create_site_validates(client, member_user):
    _auth(client, member_user)
    assert client.post("/api/admin/sites", {"name": "x", "domains": []},
                       format="json").status_code == 400
    assert client.post("/api/admin/sites", {"domains": ["a.com"]},
                       format="json").status_code == 400
    assert client.post("/api/admin/sites", ["arr"], format="json").status_code == 400   # 畸形 body 不 500


def test_list_and_patch(client, member_user):
    _auth(client, member_user)
    site = Site.objects.create(name="a", domains=["a.com"])
    r = client.get("/api/admin/sites")
    assert len(r.json()["data"]) == 1 and r.json()["data"][0]["id"] == site.pk
    r = client.patch(f"/api/admin/sites/{site.pk}",
        {"domains": ["new.com"], "is_active": False}, format="json")
    assert r.status_code == 200
    site.refresh_from_db()
    assert site.domains == ["new.com"] and site.is_active is False
    # name/allow_no_origin 也可更新
    r = client.patch(f"/api/admin/sites/{site.pk}",
        {"name": "  新名  ", "allow_no_origin": True}, format="json")
    site.refresh_from_db()
    assert site.name == "新名" and site.allow_no_origin is True


def test_patch_validates_and_404_envelope(client, member_user):
    _auth(client, member_user)
    site = Site.objects.create(name="a", domains=["a.com"])
    assert client.patch(f"/api/admin/sites/{site.pk}",
        {"domains": []}, format="json").status_code == 400
    assert client.patch(f"/api/admin/sites/{site.pk}",
        {"name": " "}, format="json").status_code == 400
    r = client.patch("/api/admin/sites/99999", {"name": "x"}, format="json")
    assert r.status_code == 404
    body = r.json()   # 404 同样走信封（Task17 前端单条解析路径）
    assert body["code"] == 404 and body["data"] is None and body["message"]


def test_delete_member_403_admin_ok(client, member_user, admin_user):
    _auth(client, member_user)
    site = Site.objects.create(name="a", domains=["a.com"])
    r = client.delete(f"/api/admin/sites/{site.pk}")
    assert r.status_code == 403 and r.json()["code"] == 403   # 权限拒绝也带信封
    _auth(client, admin_user)
    r = client.delete(f"/api/admin/sites/{site.pk}")
    assert r.status_code == 200
    site.refresh_from_db()
    assert site.is_active is False            # 软删：行还在
    assert Site.objects.filter(pk=site.pk).exists()


def test_delete_soft_keeps_collect_rejects(client, admin_user):
    """软删与 collect 联动：is_active=False 后上报 403 not_allowed（Task7 口径）。"""
    import json
    _auth(client, admin_user)
    site = Site.objects.create(name="a", domains=["a.com"], allow_no_origin=True)
    assert client.delete(f"/api/admin/sites/{site.pk}").status_code == 200
    r = client.post("/api/collect/batch",
        data=json.dumps({"site_id": site.site_id, "visitor_id": "v1",
                         "session_id": "s1", "events": []}),
        content_type="text/plain", HTTP_ORIGIN="https://a.com")
    assert r.status_code == 403 and r.json()["message"] == "not_allowed"


# ---- 审查修复锁定测试 ----

def test_scalar_domains_400_not_500(client, member_user):
    """domains 传标量：数字曾直接 500（for d in 123）、字符串曾逐字符迭代成单字符域名静默落库——统一 400。"""
    _auth(client, member_user)
    assert client.post("/api/admin/sites",
        {"name": "x", "domains": "example.com"}, format="json").status_code == 400
    assert client.post("/api/admin/sites",
        {"name": "x", "domains": 5}, format="json").status_code == 400


def test_bool_string_parsed_by_literal(client, member_user):
    """bool("false") 是 True，对 allow_no_origin 这类放宽开关方向不利——"false" 字符串须按字面落库 False。"""
    _auth(client, member_user)
    r = client.post("/api/admin/sites",
        {"name": "x", "domains": ["a.com"], "allow_no_origin": "false"}, format="json")
    assert r.status_code == 200
    site = Site.objects.get(pk=r.json()["data"]["id"])
    assert site.allow_no_origin is False
    client.patch(f"/api/admin/sites/{site.pk}", {"is_active": "false"}, format="json")
    site.refresh_from_db()
    assert site.is_active is False
    client.patch(f"/api/admin/sites/{site.pk}", {"allow_no_origin": "true"}, format="json")
    site.refresh_from_db()
    assert site.allow_no_origin is True


def test_delete_member_403_before_404(client, member_user):
    """权限先于存在性：member DELETE 不存在的 pk 也是 403——不泄露站点是否存在。"""
    _auth(client, member_user)
    r = client.delete("/api/admin/sites/99999")
    assert r.status_code == 403 and r.json()["code"] == 403