shaguabijia-app-server/tests/test_compare_daily_limit.py

from __future__ import annotations

import time
from datetime import datetime, timedelta

from sqlalchemy import func, select

from app.core.rewards import CN_TZ
from app.core.security import decode_token
from app.db.session import SessionLocal
from app.models.comparison import ComparisonRecord


def _login(client) -> tuple[str, int]:
    phone = f"137{int(time.time() * 1000) % 100000000:08d}"
    sent = client.post("/api/v1/auth/sms/send", json={"phone": phone})
    assert sent.status_code == 200, sent.text
    logged_in = client.post(
        "/api/v1/auth/sms/login",
        json={"phone": phone, "code": "123456"},
    )
    assert logged_in.status_code == 200, logged_in.text
    token = logged_in.json()["access_token"]
    return token, int(decode_token(token, expected_type="access")["sub"])


def _headers(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


def test_compare_start_requires_login(client) -> None:
    response = client.post(
        "/api/v1/compare/start",
        json={"trace_id": "quota-no-auth", "business_type": "food"},
    )
    assert response.status_code == 401


def test_compare_start_is_idempotent_by_trace_id(client) -> None:
    token, user_id = _login(client)
    payload = {
        "trace_id": f"quota-idempotent-{user_id}",
        "business_type": "ecom",
        "device_id": "quota-device",
    }

    first = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))
    retry = client.post("/api/v1/compare/start", json=payload, headers=_headers(token))

    assert first.status_code == 200, first.text
    # trace_id 回显客户端带来的值(老协议幂等路径)
    assert first.json() == {
        "limit": 100, "used": 1, "remaining": 99, "trace_id": payload["trace_id"],
    }
    assert retry.status_code == 200, retry.text
    assert retry.json() == first.json()
    with SessionLocal() as db:
        count = db.scalar(
            select(func.count(ComparisonRecord.id)).where(
                ComparisonRecord.trace_id == payload["trace_id"]
            )
        )
        record = db.execute(
            select(ComparisonRecord).where(
                ComparisonRecord.trace_id == payload["trace_id"]
            )
        ).scalar_one()
    assert count == 1
    assert record.user_id == user_id
    assert record.status == "running"
    assert record.business_type == "ecom"
    assert record.device_id == "quota-device"


def test_compare_start_issues_trace_id_when_absent(client) -> None:
    """新客户端不带 trace_id → 服务端签发并随响应返回,running 行以签发 id 建。"""
    token, user_id = _login(client)
    response = client.post(
        "/api/v1/compare/start",
        json={"business_type": "food", "device_id": "quota-device-issue"},
        headers=_headers(token),
    )
    assert response.status_code == 200, response.text
    body = response.json()
    issued = body["trace_id"]
    assert issued  # 非空签发
    assert body["used"] == 1
    with SessionLocal() as db:
        record = db.execute(
            select(ComparisonRecord).where(ComparisonRecord.trace_id == issued)
        ).scalar_one()
    assert record.user_id == user_id
    assert record.status == "running"
    assert record.device_id == "quota-device-issue"


def test_compare_start_rejects_101st_beijing_day_attempt(client) -> None:
    token, user_id = _login(client)
    now = datetime.now(CN_TZ).replace(tzinfo=None)
    with SessionLocal() as db:
        db.add_all(
            [
                ComparisonRecord(
                    user_id=user_id,
                    trace_id=f"quota-full-{user_id}-{index}",
                    status="failed",
                    created_at=now,
                )
                for index in range(99)
            ]
        )
        db.add(
            ComparisonRecord(
                user_id=user_id,
                trace_id=f"quota-yesterday-{user_id}",
                status="success",
                created_at=now - timedelta(days=1),
            )
        )
        db.commit()

    final_allowed_trace = f"quota-final-allowed-{user_id}"
    allowed = client.post(
        "/api/v1/compare/start",
        json={"trace_id": final_allowed_trace, "business_type": "food"},
        headers=_headers(token),
    )
    assert allowed.status_code == 200, allowed.text
    assert allowed.json() == {
        "limit": 100, "used": 100, "remaining": 0, "trace_id": final_allowed_trace,
    }

    rejected_trace = f"quota-rejected-{user_id}"
    response = client.post(
        "/api/v1/compare/start",
        json={"trace_id": rejected_trace, "business_type": "food"},
        headers=_headers(token),
    )

    assert response.status_code == 429
    assert response.json()["detail"] == "今日已比价超过100次，请明天再试"
    with SessionLocal() as db:
        assert db.scalar(
            select(func.count(ComparisonRecord.id)).where(
                ComparisonRecord.trace_id == rejected_trace
            )
        ) == 0