shaguabijia-app-server/tests/test_limit_whitelist.py

from __future__ import annotations

from copy import deepcopy
from datetime import UTC, datetime, timedelta
from uuid import uuid4

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.admin.security import create_admin_token
from app.api.v1 import auth as auth_api
from app.core import limit_policy
from app.core.config_schema import LIMIT_POLICY_GLOBAL_KEY
from app.core.security import create_token
from app.db.session import SessionLocal
from app.main import app
from app.models.admin import AdminAuditLog
from app.models.app_config import AppConfig
from app.models.comparison import ComparisonRecord
from app.models.limit_policy import LimitPolicyOverride
from app.models.risk import RiskIncident
from app.repositories import guide_video as guide_video_repo
from app.repositories import risk as risk_repo
from app.repositories import user as user_repo


def _snapshot_configs(keys: list[str]) -> dict[str, tuple[bool, object, int | None]]:
    with SessionLocal() as db:
        snapshot = {}
        for key in keys:
            row = db.get(AppConfig, key)
            snapshot[key] = (
                row is not None,
                deepcopy(row.value) if row is not None else None,
                row.updated_by_admin_id if row is not None else None,
            )
        return snapshot


def _restore_configs(snapshot: dict[str, tuple[bool, object, int | None]]) -> None:
    with SessionLocal() as db:
        for key, (existed, value, admin_id) in snapshot.items():
            row = db.get(AppConfig, key)
            if not existed:
                if row is not None:
                    db.delete(row)
                continue
            if row is None:
                db.add(
                    AppConfig(
                        key=key,
                        value=deepcopy(value),
                        updated_by_admin_id=admin_id,
                    )
                )
            else:
                row.value = deepcopy(value)
                row.updated_by_admin_id = admin_id
        db.commit()


@pytest.fixture()
def admin_headers() -> dict[str, str]:
    username = f"limit_admin_{uuid4().hex[:8]}"
    with SessionLocal() as db:
        admin = admin_repo.create_admin(
            db,
            username=username,
            password="limit-admin-pass",
            role="super_admin",
        )
        admin_id = admin.id
    token, _ = create_admin_token(admin_id=admin_id, role="super_admin")
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture()
def custom_admin_headers() -> dict[str, str]:
    username = f"limit_custom_{uuid4().hex[:8]}"
    with SessionLocal() as db:
        admin = admin_repo.create_admin(
            db,
            username=username,
            password="limit-admin-pass",
            role="custom",
            pages_override=["limit-whitelist"],
        )
        admin_id = admin.id
    token, _ = create_admin_token(admin_id=admin_id, role="custom")
    return {"Authorization": f"Bearer {token}"}


def test_custom_admin_with_page_permission_can_manage_whitelist(
    custom_admin_headers,
) -> None:
    phone = f"139{int(uuid4().hex[:8], 16) % 100000000:08d}"
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()

    with TestClient(admin_app) as client:
        created = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=custom_admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": ["compare.start.daily"],
                "expires_at": expires_at,
                "reason": "custom role page permission",
            },
        )
        assert created.status_code == 201, created.text
        override_id = created.json()[0]["id"]
        deleted = client.delete(
            f"/admin/api/limit-whitelist/{override_id}",
            headers=custom_admin_headers,
        )
        assert deleted.status_code == 204, deleted.text


def test_bulk_create_reactivates_existing_rows(admin_headers) -> None:
    phone = f"139{int(uuid4().hex[:8], 16) % 100000000:08d}"
    first_expiry = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
    second_expiry = (datetime.now(UTC) + timedelta(hours=3)).isoformat()

    with TestClient(admin_app) as client:
        first = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": ["compare.start.daily"],
                "expires_at": first_expiry,
                "reason": "first period",
            },
        )
        assert first.status_code == 201, first.text
        first_id = first.json()[0]["id"]
        reset = client.post(
            f"/admin/api/limit-whitelist/{first_id}/reset",
            headers=admin_headers,
        )
        assert reset.status_code == 200, reset.text

        recreated = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": [
                    "compare.start.daily",
                    "sms.phone.cooldown",
                ],
                "expires_at": second_expiry,
                "reason": "renewed period",
            },
        )
        assert recreated.status_code == 201, recreated.text
        rows = {item["rule_code"]: item for item in recreated.json()}
        assert rows["compare.start.daily"]["id"] == first_id
        assert rows["compare.start.daily"]["enabled"] is True
        assert rows["compare.start.daily"]["status"] == "active"
        assert rows["compare.start.daily"]["reason"] == "renewed period"
        assert rows["sms.phone.cooldown"]["enabled"] is True


def test_bulk_create_appends_rules_and_unifies_subject_period(admin_headers) -> None:
    phone = f"139{int(uuid4().hex[:8], 16) % 100000000:08d}"
    first_start = (datetime.now(UTC) + timedelta(minutes=5)).isoformat()
    first_expiry = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
    latest_start = (datetime.now(UTC) + timedelta(minutes=10)).isoformat()
    latest_expiry = (datetime.now(UTC) + timedelta(hours=3)).isoformat()

    with TestClient(admin_app) as client:
        first = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": ["compare.start.daily"],
                "starts_at": first_start,
                "expires_at": first_expiry,
                "reason": "original rule",
            },
        )
        assert first.status_code == 201, first.text

        appended = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": ["sms.phone.cooldown"],
                "starts_at": latest_start,
                "expires_at": latest_expiry,
                "reason": "appended rule",
            },
        )
        assert appended.status_code == 201, appended.text
        rows = {item["rule_code"]: item for item in appended.json()}
        assert set(rows) == {
            "compare.start.daily",
            "sms.phone.cooldown",
        }
        assert all(
            datetime.fromisoformat(item["starts_at"]) == datetime.fromisoformat(latest_start)
            for item in rows.values()
        )
        assert all(
            datetime.fromisoformat(item["expires_at"]) == datetime.fromisoformat(latest_expiry)
            for item in rows.values()
        )
        assert all(item["enabled"] is True for item in rows.values())
        assert rows["compare.start.daily"]["reason"] == "original rule"
        assert rows["sms.phone.cooldown"]["reason"] == "appended rule"
        with SessionLocal() as db:
            audit = db.scalar(
                select(AdminAuditLog)
                .where(AdminAuditLog.action == "limit.override.bulk_create")
                .order_by(AdminAuditLog.id.desc())
            )
            assert audit is not None
            assert len(audit.detail["before"]) == 1
            assert len(audit.detail["after"]) == 2


def test_legacy_writes_preserve_subject_level_period_and_enabled_state(
    admin_headers,
) -> None:
    phone = f"138{int(uuid4().hex[:8], 16) % 100000000:08d}"
    first_expiry = datetime.now(UTC) + timedelta(hours=2)
    latest_start = datetime.now(UTC) + timedelta(minutes=5)
    latest_expiry = datetime.now(UTC) + timedelta(hours=4)

    with TestClient(admin_app) as client:
        created = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": [
                    "compare.start.daily",
                    "sms.phone.cooldown",
                ],
                "expires_at": first_expiry.isoformat(),
                "reason": "initial subject",
            },
        )
        assert created.status_code == 201, created.text
        first_id = created.json()[0]["id"]

        patched = client.patch(
            f"/admin/api/limit-whitelist/{first_id}",
            headers=admin_headers,
            json={
                "starts_at": latest_start.isoformat(),
                "expires_at": latest_expiry.isoformat(),
            },
        )
        assert patched.status_code == 200, patched.text

        legacy_added = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_code": "ad.feed.daily",
                "mode": "unlimited",
                "enabled": True,
                "starts_at": latest_start.isoformat(),
                "expires_at": latest_expiry.isoformat(),
                "reason": "legacy append",
            },
        )
        assert legacy_added.status_code == 201, legacy_added.text

        listed = client.get(
            "/admin/api/limit-whitelist/subjects",
            headers=admin_headers,
            params={"keyword": phone},
        )
        assert listed.status_code == 200, listed.text
        rows = listed.json()["items"][0]["items"]
        assert len(rows) == 3
        assert {
            datetime.fromisoformat(item["starts_at"]).replace(tzinfo=None) for item in rows
        } == {latest_start.replace(tzinfo=None)}
        assert {
            datetime.fromisoformat(item["expires_at"]).replace(tzinfo=None) for item in rows
        } == {latest_expiry.replace(tzinfo=None)}
        assert all(item["enabled"] is True for item in rows)

        reset = client.post(
            f"/admin/api/limit-whitelist/{first_id}/reset",
            headers=admin_headers,
        )
        assert reset.status_code == 200, reset.text
        listed = client.get(
            "/admin/api/limit-whitelist/subjects",
            headers=admin_headers,
            params={"keyword": phone},
        )
        assert all(item["enabled"] is False for item in listed.json()["items"][0]["items"])


def test_subject_list_groups_rules_and_paginates_by_subject(admin_headers) -> None:
    suffix = f"{int(uuid4().hex[:7], 16) % 1000000:06d}"
    phones = [f"13870{suffix}", f"13871{suffix}"]
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()

    with TestClient(admin_app) as client:
        first = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phones[0],
                "rule_codes": [
                    "compare.start.daily",
                    "sms.phone.cooldown",
                ],
                "expires_at": expires_at,
                "reason": "grouped list first",
            },
        )
        assert first.status_code == 201, first.text
        second = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phones[1],
                "rule_codes": ["compare.start.daily"],
                "expires_at": expires_at,
                "reason": "grouped list second",
            },
        )
        assert second.status_code == 201, second.text

        page_one = client.get(
            "/admin/api/limit-whitelist/subjects",
            headers=admin_headers,
            params={"keyword": suffix, "limit": 1, "offset": 0},
        )
        page_two = client.get(
            "/admin/api/limit-whitelist/subjects",
            headers=admin_headers,
            params={"keyword": suffix, "limit": 1, "offset": 1},
        )
        assert page_one.status_code == 200, page_one.text
        assert page_two.status_code == 200, page_two.text
        assert page_one.json()["total"] == 2
        assert len(page_one.json()["items"]) == 1
        assert len(page_two.json()["items"]) == 1
        listed = page_one.json()["items"] + page_two.json()["items"]
        assert {item["subject_value"] for item in listed} == set(phones)
        first_subject = next(item for item in listed if item["subject_value"] == phones[0])
        assert first_subject["total_rules"] == 2
        assert sum(first_subject["group_counts"].values()) == 2
        assert {item["rule_code"] for item in first_subject["items"]} == {
            "compare.start.daily",
            "sms.phone.cooldown",
        }


def test_subject_replace_updates_rule_selection_atomically(admin_headers) -> None:
    phone = f"137{int(uuid4().hex[:8], 16) % 100000000:08d}"
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
    updated_expiry = (datetime.now(UTC) + timedelta(hours=4)).isoformat()

    with TestClient(admin_app) as client:
        created = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": [
                    "compare.start.daily",
                    "sms.phone.cooldown",
                ],
                "expires_at": expires_at,
                "reason": "before subject edit",
            },
        )
        assert created.status_code == 201, created.text
        original_created_at = min(item["created_at"] for item in created.json())

        replaced = client.put(
            "/admin/api/limit-whitelist/subjects",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": [
                    "ad.reward_video.daily",
                ],
                "expires_at": updated_expiry,
                "reason": "after subject edit",
            },
        )
        assert replaced.status_code == 200, replaced.text
        payload = replaced.json()
        assert payload["subject_value"] == phone
        assert payload["total_rules"] == 1
        assert {item["rule_code"] for item in payload["items"]} == {
            "ad.reward_video.daily",
        }
        assert payload["created_at"] == original_created_at
        assert all(item["reason"] == "after subject edit" for item in payload["items"])

        listed = client.get(
            "/admin/api/limit-whitelist/subjects",
            headers=admin_headers,
            params={"keyword": phone},
        )
        assert listed.status_code == 200, listed.text
        assert listed.json()["total"] == 1
        assert {item["rule_code"] for item in listed.json()["items"][0]["items"]} == {
            "ad.reward_video.daily",
        }
        assert listed.json()["items"][0]["created_at"] == original_created_at


def test_subject_enabled_patch_updates_all_rules_together(admin_headers) -> None:
    phone = f"136{int(uuid4().hex[:8], 16) % 100000000:08d}"
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()

    with TestClient(admin_app) as client:
        created = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": [
                    "compare.start.daily",
                    "sms.phone.cooldown",
                    "ad.reward_video.daily",
                ],
                "expires_at": expires_at,
                "reason": "subject switch",
            },
        )
        assert created.status_code == 201, created.text

        disabled = client.patch(
            "/admin/api/limit-whitelist/subjects/enabled",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "enabled": False,
            },
        )
        assert disabled.status_code == 200, disabled.text
        assert disabled.json()["total_rules"] == 3
        assert all(item["enabled"] is False for item in disabled.json()["items"])
        assert all(item["status"] == "disabled" for item in disabled.json()["items"])

        enabled = client.patch(
            "/admin/api/limit-whitelist/subjects/enabled",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "enabled": True,
            },
        )
        assert enabled.status_code == 200, enabled.text
        assert all(item["enabled"] is True for item in enabled.json()["items"])
        assert all(item["status"] == "active" for item in enabled.json()["items"])

        disabled_again = client.patch(
            "/admin/api/limit-whitelist/subjects/enabled",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "enabled": False,
            },
        )
        assert disabled_again.status_code == 200, disabled_again.text
        oldest_id = min(item["id"] for item in created.json())
        expired = client.patch(
            f"/admin/api/limit-whitelist/{oldest_id}",
            headers=admin_headers,
            json={
                "expires_at": (datetime.now(UTC) - timedelta(minutes=1)).isoformat(),
            },
        )
        assert expired.status_code == 200, expired.text

        rejected = client.patch(
            "/admin/api/limit-whitelist/subjects/enabled",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "enabled": True,
            },
        )
        assert rejected.status_code == 400, rejected.text
        listed = client.get(
            "/admin/api/limit-whitelist/subjects",
            headers=admin_headers,
            params={"keyword": phone},
        )
        assert listed.status_code == 200, listed.text
        assert all(item["enabled"] is False for item in listed.json()["items"][0]["items"])


def test_admin_whitelist_crud_and_policy_precedence(admin_headers) -> None:
    suffix = uuid4().hex[:10]
    phone = f"139{int(suffix[:8], 16) % 100000000:08d}"
    device = f"limit-device-{suffix}"
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()

    with TestClient(admin_app) as client:
        rules = client.get(
            "/admin/api/limit-whitelist/rules",
            headers=admin_headers,
        )
        assert rules.status_code == 200, rules.text
        assert {item["code"] for item in rules.json()} >= {
            "compare.start.daily",
            "sms.send.hourly",
            "ad.reward_video.daily",
            "risk.compare.daily",
        }
        rules_by_code = {item["code"]: item for item in rules.json()}
        assert all("inherit" not in item["allowed_modes"] for item in rules.json())
        assert all("override" not in item["allowed_modes"] for item in rules.json())
        assert "unlimited" in rules_by_code["sms.phone.cooldown"]["allowed_modes"]
        assert "unlimited" in rules_by_code["sms.code.failed_attempts"]["allowed_modes"]
        assert "suppress_alert" in rules_by_code["risk.compare.daily"]["allowed_modes"]

        removed_inherit = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_code": "compare.start.daily",
                "mode": "inherit",
                "enabled": True,
            },
        )
        assert removed_inherit.status_code == 422

        invalid_alert = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_code": "compare.start.daily",
                "mode": "suppress_alert",
                "enabled": True,
                "reason": "invalid",
            },
        )
        assert invalid_alert.status_code == 400

        missing_expiry = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_code": "compare.start.daily",
                "mode": "unlimited",
                "enabled": True,
                "reason": "temporary QA",
            },
        )
        assert missing_expiry.status_code == 400

        expired_unlimited = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_code": "compare.start.daily",
                "mode": "unlimited",
                "enabled": True,
                "expires_at": "2020-01-01T00:00:00Z",
                "reason": "expired temporary policy",
            },
        )
        assert expired_unlimited.status_code == 400
        assert "晚于当前时间" in expired_unlimited.json()["detail"]

        removed_override = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_code": "compare.start.daily",
                "mode": "override",
                "limit_value": 2,
                "enabled": True,
                "expires_at": expires_at,
            },
        )
        assert removed_override.status_code == 422

        phone_created = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_code": "compare.start.daily",
                "mode": "unlimited",
                "enabled": True,
                "expires_at": expires_at,
                "reason": "temporary QA",
            },
        )
        assert phone_created.status_code == 201, phone_created.text
        phone_id = phone_created.json()["id"]
        assert phone_created.json()["effective_limit"] is None

        phone_updated = client.patch(
            f"/admin/api/limit-whitelist/{phone_id}",
            headers=admin_headers,
            json={
                "enabled": True,
                "reason": "updated from admin",
            },
        )
        assert phone_updated.status_code == 200, phone_updated.text
        assert phone_updated.json()["effective_limit"] is None
        assert phone_updated.json()["reason"] == "updated from admin"
        with SessionLocal() as db:
            persisted = db.get(LimitPolicyOverride, phone_id)
            assert persisted is not None
            assert persisted.mode == "unlimited"
            assert persisted.limit_value is None
            assert persisted.reason == "updated from admin"

        removed_patch_fields = client.patch(
            f"/admin/api/limit-whitelist/{phone_id}",
            headers=admin_headers,
            json={"mode": "override", "limit_value": 4},
        )
        assert removed_patch_fields.status_code == 422

        inverted_time = client.patch(
            f"/admin/api/limit-whitelist/{phone_id}",
            headers=admin_headers,
            json={
                "starts_at": "2030-01-02T00:00:00Z",
                "expires_at": "2030-01-01T00:00:00Z",
            },
        )
        assert inverted_time.status_code == 400
        assert "晚于生效时间" in inverted_time.json()["detail"]

        duplicate = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_code": "compare.start.daily",
                "mode": "unlimited",
                "enabled": True,
                "expires_at": expires_at,
            },
        )
        assert duplicate.status_code == 409

        device_created = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "device",
                "subject_value": device,
                "rule_code": "compare.start.daily",
                "mode": "unlimited",
                "enabled": True,
                "expires_at": expires_at,
                "reason": "device QA",
            },
        )
        assert device_created.status_code == 201, device_created.text
        device_id = device_created.json()["id"]
        assert device_created.json()["effective_limit"] is None

        with SessionLocal() as db:
            effective = limit_policy.resolve(
                db,
                "compare.start.daily",
                phone=phone,
                device=device,
            )
            assert effective.unlimited is True
            assert effective.matched_subject_type == "device"

        restored = client.post(
            f"/admin/api/limit-whitelist/{phone_id}/reset",
            headers=admin_headers,
        )
        assert restored.status_code == 200, restored.text
        assert restored.json()["enabled"] is False
        assert restored.json()["status"] == "disabled"
        assert restored.json()["effective_limit"] == restored.json()["global_limit"]
        assert restored.json()["reset_at"] is None
        with SessionLocal() as db:
            effective = limit_policy.resolve(
                db,
                "compare.start.daily",
                phone=phone,
                device=None,
            )
            assert effective.override_id is None
            assert effective.limit == effective.global_limit

        ordered = client.get(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            params={"rule_code": "compare.start.daily", "limit": 500},
        )
        assert ordered.status_code == 200, ordered.text
        ordered_ids = [item["id"] for item in ordered.json()["items"]]
        assert ordered_ids.index(device_id) < ordered_ids.index(phone_id)

        listed = client.get(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            params={"keyword": suffix[:5]},
        )
        assert listed.status_code == 200
        assert listed.json()["total"] >= 1

        deleted = client.delete(
            f"/admin/api/limit-whitelist/{device_id}",
            headers=admin_headers,
        )
        assert deleted.status_code == 204
        client.delete(
            f"/admin/api/limit-whitelist/{phone_id}",
            headers=admin_headers,
        )


def test_device_candidates_are_rule_aware_and_searchable(admin_headers) -> None:
    suffix = uuid4().hex[:8]
    phone = f"135{int(suffix, 16) % 100000000:08d}"
    auth_device = f"android-id-{suffix}"
    compare_device = f"device-PJZ110-{suffix}"
    now = datetime.now(UTC).replace(microsecond=0)
    with SessionLocal() as db:
        user = user_repo.upsert_user_for_login(
            db,
            phone=phone,
            register_channel="sms",
        )
        user.nickname = f"候选设备用户{suffix}"
        risk_repo.record_behavior_event(
            db,
            event_type=risk_repo.EVENT_ONECLICK_LOGIN,
            subject_type="device",
            subject_id=auth_device,
            user_id=user.id,
            device_id=auth_device,
            device_model="OPPO Find X8",
            phone=phone,
            outcome="success",
            occurred_at=now,
        )
        db.add(
            ComparisonRecord(
                user_id=user.id,
                device_id=compare_device,
                trace_id=f"device-candidate-{suffix}",
                device_model="PJZ110",
                status="success",
                items=[],
                comparison_results=[],
                skipped_dish_names=[],
                created_at=now.replace(tzinfo=None),
            )
        )
        db.commit()
        user_id = user.id
        username = user.username

    with TestClient(admin_app) as client:
        for keyword in (phone, str(user_id), username, "Find X8"):
            response = client.get(
                "/admin/api/limit-whitelist/device-candidates",
                headers=admin_headers,
                params={
                    "rule_code": "risk.oneclick.daily",
                    "keyword": keyword,
                },
            )
            assert response.status_code == 200, response.text
            item = next(row for row in response.json() if row["device_id"] == auth_device)
            assert item["source_label"] == "一键登录"
            assert item["user_id"] == user_id
            assert item["phone"] == phone
            assert item["device_model"] == "OPPO Find X8"

        comparison = client.get(
            "/admin/api/limit-whitelist/device-candidates",
            headers=admin_headers,
            params={
                "rule_code": "compare.start.daily",
                "keyword": "PJZ110",
            },
        )
        assert comparison.status_code == 200, comparison.text
        assert len(comparison.json()) == 1
        comparison_item = comparison.json()[0]
        assert comparison_item["device_id"] == compare_device
        assert comparison_item["source"] == "comparison_record"
        assert comparison_item["source_label"] == "比价记录"
        assert comparison_item["user_id"] == user_id
        assert comparison_item["username"] == username
        assert comparison_item["phone"] == phone
        assert comparison_item["nickname"] == f"候选设备用户{suffix}"
        assert comparison_item["device_model"] == "PJZ110"
        assert comparison_item["last_active_at"].startswith(now.replace(tzinfo=None).isoformat())

        phone_only = client.get(
            "/admin/api/limit-whitelist/device-candidates",
            headers=admin_headers,
            params={"rule_code": "risk.compare.daily"},
        )
        assert phone_only.status_code == 400
        assert "不支持设备白名单" in phone_only.json()["detail"]


def test_legacy_ip_is_not_a_device_candidate_or_valid_whitelist_subject(
    admin_headers,
) -> None:
    suffix = uuid4().hex[:8]
    legacy_device = f"legacy-ip:203.0.113.{int(suffix[:2], 16) % 200 + 1}"
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
    with SessionLocal() as db:
        risk_repo.record_behavior_event(
            db,
            event_type=risk_repo.EVENT_SMS_SEND,
            subject_type="device",
            subject_id=legacy_device,
            device_id=None,
            phone=f"136{int(suffix, 16) % 100000000:08d}",
            outcome="success",
        )

    with TestClient(admin_app) as client:
        candidates = client.get(
            "/admin/api/limit-whitelist/device-candidates",
            headers=admin_headers,
            params={
                "rule_code": "sms.send.hourly",
                "keyword": legacy_device,
            },
        )
        assert candidates.status_code == 200, candidates.text
        assert all(row["device_id"] != legacy_device for row in candidates.json())

        single = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "device",
                "subject_value": legacy_device,
                "rule_code": "sms.send.hourly",
                "mode": "unlimited",
                "enabled": True,
                "expires_at": expires_at,
            },
        )
        assert single.status_code == 400, single.text
        assert "未上报真实设备 ID" in single.json()["detail"]

        bulk = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "device",
                "subject_value": legacy_device,
                "rule_codes": ["sms.send.hourly", "sms.send.daily"],
                "enabled": True,
                "expires_at": expires_at,
            },
        )
        assert bulk.status_code == 400, bulk.text
        assert "未上报真实设备 ID" in bulk.json()["detail"]


def test_bulk_create_automatically_selects_unlimited_and_suppress_modes(
    admin_headers,
) -> None:
    suffix = uuid4().hex[:8]
    phone = f"137{int(suffix, 16) % 100000000:08d}"
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()

    with TestClient(admin_app) as client:
        created = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": [
                    "compare.start.daily",
                    "risk.compare.daily",
                ],
                "expires_at": expires_at,
                "reason": "批量白名单测试",
            },
        )
        assert created.status_code == 201, created.text
        rows = {item["rule_code"]: item for item in created.json()}
        assert rows["compare.start.daily"]["mode"] == "unlimited"
        assert rows["compare.start.daily"]["effective_limit"] is None
        assert rows["risk.compare.daily"]["mode"] == "suppress_alert"
        assert (
            rows["risk.compare.daily"]["effective_limit"]
            == rows["risk.compare.daily"]["global_limit"]
        )

        appended_batch = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "phone",
                "subject_value": phone,
                "rule_codes": [
                    "compare.start.daily",
                    "sms.send.daily",
                ],
                "expires_at": expires_at,
            },
        )
        assert appended_batch.status_code == 201, appended_batch.text
        appended_rows = {item["rule_code"]: item for item in appended_batch.json()}
        assert set(appended_rows) == {
            "compare.start.daily",
            "risk.compare.daily",
            "sms.send.daily",
        }

        for item in appended_batch.json():
            deleted = client.delete(
                f"/admin/api/limit-whitelist/{item['id']}",
                headers=admin_headers,
            )
            assert deleted.status_code == 204


def test_device_bulk_accepts_rules_from_different_business_categories(
    admin_headers,
) -> None:
    device_id = f"manually-entered-device-{uuid4().hex[:8]}"
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
    with TestClient(admin_app) as client:
        response = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "device",
                "subject_value": device_id,
                "rule_codes": [
                    "compare.start.daily",
                    "sms.send.hourly",
                    "risk.oneclick.daily",
                ],
                "expires_at": expires_at,
            },
        )
    assert response.status_code == 201, response.text
    rows = {item["rule_code"]: item for item in response.json()}
    assert set(rows) == {
        "compare.start.daily",
        "sms.send.hourly",
        "risk.oneclick.daily",
    }
    assert all(item["subject_value"] == device_id for item in rows.values())

    with SessionLocal() as db:
        compare = limit_policy.resolve(
            db,
            "compare.start.daily",
            device=device_id,
        )
        sms = limit_policy.resolve(
            db,
            "sms.send.hourly",
            device=device_id,
        )
        alert = limit_policy.resolve(
            db,
            "risk.oneclick.daily",
            device=device_id,
        )
        assert compare.unlimited is True
        assert sms.unlimited is True
        assert alert.suppressed is True


def test_device_subject_can_append_and_replace_cross_category_rules(
    admin_headers,
) -> None:
    device_id = f"opaque-device-value-{uuid4().hex[:8]}"
    expires_at = (datetime.now(UTC) + timedelta(hours=2)).isoformat()
    with TestClient(admin_app) as client:
        initial = client.post(
            "/admin/api/limit-whitelist/bulk",
            headers=admin_headers,
            json={
                "subject_type": "device",
                "subject_value": device_id,
                "rule_codes": ["compare.start.daily"],
                "expires_at": expires_at,
            },
        )
        assert initial.status_code == 201, initial.text

        appended = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "device",
                "subject_value": device_id,
                "rule_code": "risk.oneclick.daily",
                "mode": "suppress_alert",
                "enabled": True,
                "expires_at": expires_at,
            },
        )
        assert appended.status_code == 201, appended.text

        replaced = client.put(
            "/admin/api/limit-whitelist/subjects",
            headers=admin_headers,
            json={
                "subject_type": "device",
                "subject_value": device_id,
                "rule_codes": [
                    "compare.start.daily",
                    "sms.send.daily",
                    "risk.oneclick.daily",
                ],
                "enabled": True,
                "expires_at": expires_at,
            },
        )
        assert replaced.status_code == 200, replaced.text
        assert {
            item["rule_code"] for item in replaced.json()["items"]
        } == {
            "compare.start.daily",
            "sms.send.daily",
            "risk.oneclick.daily",
        }


def test_legacy_ip_device_whitelist_matches_sms_policy(monkeypatch) -> None:
    suffix = uuid4().hex[:8]
    client_ip = f"203.0.113.{int(suffix[:2], 16) % 200 + 1}"
    legacy_device = f"legacy-ip:{client_ip}"
    phone = f"134{int(suffix, 16) % 100000000:08d}"
    expires_at = datetime.now(UTC) + timedelta(hours=2)
    with SessionLocal() as db:
        row = LimitPolicyOverride(
            subject_type="device",
            subject_value=legacy_device,
            rule_code="sms.send.hourly",
            mode="unlimited",
            enabled=True,
            expires_at=expires_at,
        )
        db.add(row)
        db.commit()
        override_id = row.id

    original_resolve = auth_api.limit_policy.resolve
    matched_override_ids: list[int | None] = []

    def spy_resolve(db, rule_code, **subjects):
        result = original_resolve(db, rule_code, **subjects)
        if rule_code == "sms.send.hourly":
            matched_override_ids.append(result.override_id)
        return result

    monkeypatch.setattr(auth_api.limit_policy, "resolve", spy_resolve)
    try:
        with TestClient(app) as client:
            response = client.post(
                "/api/v1/auth/sms/send",
                headers={"x-forwarded-for": client_ip},
                json={"phone": phone},
            )
        assert response.status_code == 200, response.text
        assert matched_override_ids == [override_id]
    finally:
        with SessionLocal() as db:
            row = db.get(LimitPolicyOverride, override_id)
            if row is not None:
                db.delete(row)
                db.commit()


def test_global_guide_limit_stays_in_sync_with_legacy_config(admin_headers) -> None:
    snapshot = _snapshot_configs(
        [
            LIMIT_POLICY_GLOBAL_KEY,
            "coupon_guide_video",
            "comparison_guide_video",
        ]
    )
    try:
        with TestClient(admin_app) as client:
            changed = client.patch(
                "/admin/api/limit-whitelist/rules/guide.video.lifetime",
                headers=admin_headers,
                json={"value": 7},
            )
            assert changed.status_code == 200, changed.text
            assert changed.json()["global_limit"] == 7

            legacy_page = client.get(
                "/admin/api/guide-video",
                headers=admin_headers,
            )
            assert legacy_page.status_code == 200, legacy_page.text
            assert legacy_page.json()["max_plays"] == 7
            assert legacy_page.json()["scene"] == "coupon"

            comparison_changed = client.patch(
                "/admin/api/guide-video",
                headers=admin_headers,
                params={"scene": "comparison"},
                json={"max_plays": 9},
            )
            assert comparison_changed.status_code == 200, comparison_changed.text
            assert comparison_changed.json()["scene"] == "comparison"
            assert comparison_changed.json()["max_plays"] == 9

            coupon_again = client.get(
                "/admin/api/guide-video",
                headers=admin_headers,
                params={"scene": "coupon"},
            )
            assert coupon_again.status_code == 200, coupon_again.text
            assert coupon_again.json()["max_plays"] == 7
    finally:
        _restore_configs(snapshot)


def test_legacy_guide_limit_audit_uses_synced_global_value(admin_headers) -> None:
    snapshot = _snapshot_configs([LIMIT_POLICY_GLOBAL_KEY, "coupon_guide_video"])
    try:
        with TestClient(admin_app) as client:
            changed = client.patch(
                "/admin/api/guide-video",
                headers=admin_headers,
                params={"scene": "coupon"},
                json={"max_plays": 11},
            )
            assert changed.status_code == 200, changed.text
            assert changed.json()["max_plays"] == 11

        with SessionLocal() as db:
            audit = db.scalar(
                select(AdminAuditLog)
                .where(AdminAuditLog.action == "guide_video.update")
                .order_by(AdminAuditLog.id.desc())
            )
            assert audit is not None
            assert audit.detail["after"]["max_plays"] == 11
    finally:
        _restore_configs(snapshot)


def test_zero_global_guide_limit_keeps_legacy_config_editable(admin_headers) -> None:
    snapshot = _snapshot_configs([LIMIT_POLICY_GLOBAL_KEY, "coupon_guide_video"])
    try:
        with TestClient(admin_app) as client:
            zeroed = client.patch(
                "/admin/api/limit-whitelist/rules/guide.video.lifetime",
                headers=admin_headers,
                json={"value": 0},
            )
            assert zeroed.status_code == 200, zeroed.text
            assert zeroed.json()["global_limit"] == 0

            legacy_changed = client.patch(
                "/admin/api/guide-video",
                headers=admin_headers,
                params={"scene": "coupon"},
                json={"reward_coin": 200},
            )
            assert legacy_changed.status_code == 200, legacy_changed.text
            assert legacy_changed.json()["max_plays"] == 0
            assert legacy_changed.json()["reward_coin"] == 200
    finally:
        _restore_configs(snapshot)


def test_guide_video_v2_honours_unlimited_override_without_reusing_seq() -> None:
    snapshot = _snapshot_configs([LIMIT_POLICY_GLOBAL_KEY, "coupon_guide_video"])
    suffix = uuid4().hex[:8]
    phone = f"139{int(suffix, 16) % 100_000_000:08d}"
    override_id: int | None = None
    try:
        with SessionLocal() as db:
            user = user_repo.upsert_user_for_login(
                db,
                phone=phone,
                register_channel="sms",
            )
            user_id = user.id
            guide_video_repo.set_video(
                db,
                "/media/guide_video/whitelist-v2-test.mp4",
                analysis={
                    "duration_ms": 10_000,
                    "video_codec": "h264",
                    "audio_codec": "aac",
                    "analysis_status": "valid",
                    "analysis_error": None,
                },
                scene="coupon",
                admin_id=1,
            )
            guide_video_repo.update_config(
                db,
                scene="coupon",
                enabled=True,
                max_plays=1,
                reward_coin=100,
                admin_id=1,
            )

            first_plan = guide_video_repo.prepare_play(db, user_id, scene="coupon")
            first_start = guide_video_repo.start_play(
                db, user_id, play_token=first_plan["play_token"]
            )
            assert first_start["seq"] == 1
            assert (
                guide_video_repo.prepare_play(db, user_id, scene="coupon")["reason"]
                == "play_limit_reached"
            )

            override = LimitPolicyOverride(
                rule_code="guide.video.lifetime",
                subject_type="phone",
                subject_value=phone,
                mode="unlimited",
                limit_value=None,
                starts_at=datetime.now(UTC) - timedelta(seconds=1),
                expires_at=datetime.now(UTC) + timedelta(hours=1),
                reset_at=datetime.now(UTC),
                enabled=True,
                reason="新版十圈视频白名单回归测试",
            )
            db.add(override)
            db.commit()
            db.refresh(override)
            override_id = override.id

            second_plan = guide_video_repo.prepare_play(db, user_id, scene="coupon")
            assert second_plan["should_play"] is True
            second_start = guide_video_repo.start_play(
                db, user_id, play_token=second_plan["play_token"]
            )
            assert second_start["seq"] == 2
            assert second_start["remaining"] > 0
    finally:
        with SessionLocal() as db:
            if override_id is not None:
                override = db.get(LimitPolicyOverride, override_id)
                if override is not None:
                    db.delete(override)
                    db.commit()
        _restore_configs(snapshot)


def test_legacy_ad_limit_update_syncs_split_global_rules(admin_headers) -> None:
    snapshot = _snapshot_configs(["ad_daily_limit", LIMIT_POLICY_GLOBAL_KEY])
    try:
        with TestClient(admin_app) as client:
            changed = client.patch(
                "/admin/api/config/ad_daily_limit",
                headers=admin_headers,
                json={"value": 321},
            )
            assert changed.status_code == 200, changed.text
            assert changed.json()["value"] == 321

            rules = client.get(
                "/admin/api/limit-whitelist/rules",
                headers=admin_headers,
            )
            by_code = {item["code"]: item for item in rules.json()}
            assert by_code["ad.reward_video.daily"]["global_limit"] == 321
            assert by_code["ad.feed.daily"]["global_limit"] == 321
    finally:
        _restore_configs(snapshot)


def test_zero_disables_cooldown_style_global_rules(admin_headers) -> None:
    snapshot = _snapshot_configs([LIMIT_POLICY_GLOBAL_KEY])
    try:
        with TestClient(admin_app) as client:
            for rule_code in ("sms.phone.cooldown", "phone.rebind.days"):
                changed = client.patch(
                    f"/admin/api/limit-whitelist/rules/{rule_code}",
                    headers=admin_headers,
                    json={"value": 0},
                )
                assert changed.status_code == 200, changed.text
                assert changed.json()["global_limit"] == 0
    finally:
        _restore_configs(snapshot)


def test_global_limits_are_stored_in_one_complete_json(admin_headers) -> None:
    keys = [LIMIT_POLICY_GLOBAL_KEY, *limit_policy.LIMIT_CONFIG_KEYS]
    snapshot = _snapshot_configs(keys)
    try:
        with TestClient(admin_app) as client:
            first = client.patch(
                "/admin/api/limit-whitelist/rules/compare.start.daily",
                headers=admin_headers,
                json={"value": 73},
            )
            second = client.patch(
                "/admin/api/limit-whitelist/rules/sms.send.hourly",
                headers=admin_headers,
                json={"value": 9},
            )
            assert first.status_code == 200, first.text
            assert second.status_code == 200, second.text
            direct_bundle_write = client.patch(
                f"/admin/api/config/{LIMIT_POLICY_GLOBAL_KEY}",
                headers=admin_headers,
                json={"value": {"compare.start.daily": 1}},
            )
            assert direct_bundle_write.status_code == 404

        with SessionLocal() as db:
            bundle = db.get(AppConfig, LIMIT_POLICY_GLOBAL_KEY)
            assert bundle is not None
            assert isinstance(bundle.value, dict)
            assert set(bundle.value) == set(limit_policy.RULE_MAP)
            assert len(bundle.value) == 16
            assert bundle.value["compare.start.daily"] == 73
            assert bundle.value["sms.send.hourly"] == 9
            sparse_rows = (
                db.query(AppConfig)
                .filter(AppConfig.key.in_(limit_policy.LIMIT_CONFIG_KEYS))
                .count()
            )
            assert sparse_rows == 0
    finally:
        _restore_configs(snapshot)


def test_compare_start_uses_phone_override_and_reset_baseline() -> None:
    suffix = uuid4().hex[:8]
    phone = f"136{int(suffix, 16) % 100000000:08d}"
    device = f"compare-policy-{suffix}"
    with SessionLocal() as db:
        user = user_repo.upsert_user_for_login(
            db,
            phone=phone,
            register_channel="sms",
        )
        db.add(
            LimitPolicyOverride(
                subject_type="phone",
                subject_value=phone,
                rule_code="compare.start.daily",
                mode="override",
                limit_value=1,
                enabled=True,
            )
        )
        db.commit()
        user_id = user.id

    token, _ = create_token(user_id=user_id, token_type="access")
    headers = {"Authorization": f"Bearer {token}"}
    with TestClient(app) as client:
        first = client.post(
            "/api/v1/compare/start",
            headers=headers,
            json={
                "trace_id": f"limit-policy-first-{suffix}",
                "business_type": "food",
                "device_id": device,
            },
        )
        assert first.status_code == 200, first.text
        assert first.json() == {"limit": 1, "used": 1, "remaining": 0}

        blocked = client.post(
            "/api/v1/compare/start",
            headers=headers,
            json={
                "trace_id": f"limit-policy-blocked-{suffix}",
                "business_type": "food",
                "device_id": device,
            },
        )
        assert blocked.status_code == 429

        with SessionLocal() as db:
            override = (
                db.query(LimitPolicyOverride)
                .filter(
                    LimitPolicyOverride.subject_type == "phone",
                    LimitPolicyOverride.subject_value == phone,
                    LimitPolicyOverride.rule_code == "compare.start.daily",
                )
                .one()
            )
            override.reset_at = datetime.now(UTC)
            db.commit()

        after_reset = client.post(
            "/api/v1/compare/start",
            headers=headers,
            json={
                "trace_id": f"limit-policy-reset-{suffix}",
                "business_type": "food",
                "device_id": device,
            },
        )
        assert after_reset.status_code == 200, after_reset.text
        assert after_reset.json() == {"limit": 1, "used": 1, "remaining": 0}


def test_suppress_alert_keeps_events_but_creates_no_incident() -> None:
    suffix = uuid4().hex[:10]
    device = f"suppress-device-{suffix}"
    phone = f"138{int(suffix[:8], 16) % 100000000:08d}"
    now = risk_repo.utcnow().replace(microsecond=0)
    with SessionLocal() as db:
        db.add(
            LimitPolicyOverride(
                subject_type="device",
                subject_value=device,
                rule_code="risk.sms.hourly",
                mode="suppress_alert",
                enabled=True,
            )
        )
        db.commit()
        for index in range(10):
            risk_repo.record_behavior_event(
                db,
                event_type=risk_repo.EVENT_SMS_SEND,
                subject_type="device",
                subject_id=device,
                device_id=device,
                phone=phone,
                outcome="success",
                occurred_at=now + timedelta(seconds=index),
                evaluate_rule=risk_repo.RULE_SMS_HOURLY,
            )

    with SessionLocal() as db:
        incident = (
            db.query(RiskIncident)
            .filter(
                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
                RiskIncident.subject_id == device,
            )
            .one_or_none()
        )
        assert incident is None


def test_adding_suppress_alert_resolves_existing_incident(admin_headers) -> None:
    suffix = uuid4().hex[:10]
    device = f"suppress-existing-{suffix}"
    now = risk_repo.utcnow().replace(microsecond=0)
    with SessionLocal() as db:
        for index in range(5):
            risk_repo.record_behavior_event(
                db,
                event_type=risk_repo.EVENT_SMS_SEND,
                subject_type="device",
                subject_id=device,
                device_id=device,
                phone=f"13700001{index:03d}",
                outcome="success",
                occurred_at=now + timedelta(seconds=index),
                evaluate_rule=risk_repo.RULE_SMS_HOURLY,
            )
        incident = (
            db.query(RiskIncident)
            .filter(
                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
                RiskIncident.subject_id == device,
            )
            .one()
        )
        assert incident.status == "open"

    with TestClient(admin_app) as client:
        created = client.post(
            "/admin/api/limit-whitelist",
            headers=admin_headers,
            json={
                "subject_type": "device",
                "subject_value": device,
                "rule_code": "risk.sms.hourly",
                "mode": "suppress_alert",
                "enabled": True,
                "expires_at": (datetime.now(UTC) + timedelta(hours=2)).isoformat(),
                "reason": "QA device",
            },
        )
        assert created.status_code == 201, created.text

    with SessionLocal() as db:
        incident = (
            db.query(RiskIncident)
            .filter(
                RiskIncident.rule_code == risk_repo.RULE_SMS_HOURLY,
                RiskIncident.subject_id == device,
            )
            .one()
        )
        assert incident.status == "resolved"
        assert incident.action_reason == risk_repo.AUTO_RESOLVED_REASON