unknown
944 lines
33 KiB
Python
944 lines
33 KiB
Python
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 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 (
|
|
AD_FEED_DAILY_LIMIT_KEY,
|
|
AD_REWARD_VIDEO_DAILY_LIMIT_KEY,
|
|
GUIDE_VIDEO_MAX_PLAYS_KEY,
|
|
PHONE_REBIND_DAYS_KEY,
|
|
SMS_PHONE_COOLDOWN_SECONDS_KEY,
|
|
)
|
|
from app.core.security import create_token
|
|
from app.db.session import SessionLocal
|
|
from app.main import app
|
|
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 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_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"]
|
|
|
|
duplicate_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 duplicate_batch.status_code == 409
|
|
with SessionLocal() as db:
|
|
rolled_back = (
|
|
db.query(LimitPolicyOverride)
|
|
.filter(
|
|
LimitPolicyOverride.subject_type == "phone",
|
|
LimitPolicyOverride.subject_value == phone,
|
|
LimitPolicyOverride.rule_code == "sms.send.daily",
|
|
)
|
|
.one_or_none()
|
|
)
|
|
assert rolled_back is None
|
|
|
|
for item in created.json():
|
|
deleted = client.delete(
|
|
f"/admin/api/limit-whitelist/{item['id']}",
|
|
headers=admin_headers,
|
|
)
|
|
assert deleted.status_code == 204
|
|
|
|
|
|
def test_device_bulk_rejects_mixed_candidate_namespaces(admin_headers) -> None:
|
|
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": f"mixed-device-{uuid4().hex[:8]}",
|
|
"rule_codes": [
|
|
"compare.start.daily",
|
|
"sms.send.hourly",
|
|
],
|
|
"expires_at": expires_at,
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "不能在同一白名单中混选" in response.json()["detail"]
|
|
|
|
|
|
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(
|
|
[
|
|
GUIDE_VIDEO_MAX_PLAYS_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_ad_limit_update_syncs_split_global_rules(admin_headers) -> None:
|
|
snapshot = _snapshot_configs(
|
|
["ad_daily_limit", AD_REWARD_VIDEO_DAILY_LIMIT_KEY, AD_FEED_DAILY_LIMIT_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
|
|
|
|
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(
|
|
[SMS_PHONE_COOLDOWN_SECONDS_KEY, PHONE_REBIND_DAYS_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_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
|