dotb_ai-platform/tests/test_score_endpoint.py

import json
import threading
from uuid import uuid4

import httpx
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import func, select

from ai_platform.infrastructure.persistence.models import AIProfile, ScoringRun

URL = "/v1/lead-scoring/score"
ASSESSMENT = {
    "signals": [
        {
            "polarity": "positive",
            "signal": "Hỏi giá",
            "detail": "Hỏi học phí.",
            "source": "message-789",
        }
    ],
    "score": 90,
    "summary": "Khách quan tâm rõ ràng.",
    "suggested_action": "Gọi lại trong ngày.",
}


def _completion(content: dict) -> httpx.Response:
    return httpx.Response(
        200, json={"choices": [{"message": {"content": json.dumps(content, ensure_ascii=False)}}]}
    )


def _ai(calls: list[httpx.Request]):
    def handler(request: httpx.Request) -> httpx.Response:
        calls.append(request)
        return _completion(ASSESSMENT)

    return handler


def _post(client, env, body, *, key=None, ai_key=False, **headers):
    headers = {**env["headers"](key), **({"X-AI-Key": env["ai_key"]} if ai_key else {}), **headers}
    return client.post(URL, headers=headers, json=body)


def _hybrid(request: dict) -> dict:
    request["scoring_config"].update(rule_weight=0.5, ai_weight=0.5)
    return request


def test_rules_only_never_calls_ai(score_env, canonical_request):
    calls: list[httpx.Request] = []
    with TestClient(score_env["app_for"](_ai(calls))) as client:
        response = _post(client, score_env, canonical_request)
    body = response.json()
    assert response.status_code == 200
    assert (body["result_mode"], body["final_score"], body["rule_score"]) == ("rules_only", 50, 50)
    assert (body["ai"], body["metrics"]["message_count"]) == (None, 12)
    assert body["metadata"] == canonical_request["metadata"]
    assert not calls


def test_hybrid_uses_one_ai_call_and_never_stores_or_logs_the_ai_key(
    score_env, canonical_request, caplog
):
    calls: list[httpx.Request] = []
    with TestClient(score_env["app_for"](_ai(calls))) as client:
        response = _post(client, score_env, _hybrid(canonical_request), ai_key=True)
    body = response.json()
    assert response.status_code == 200
    assert (body["result_mode"], body["ai_score"], body["final_score"]) == ("hybrid", 90, 70)
    assert body["ai"]["signals"][0]["source"] == "message-789"
    assert len(calls) == 1
    assert calls[0].headers["authorization"] == f"Bearer {score_env['ai_key']}"
    assert (
        "Thời điểm chấm: Thứ Năm, 25/09/2025 17:00"
        in json.loads(calls[0].content)["messages"][1]["content"]
    )

    stored = score_env["read"](
        lambda db: db.scalar(select(ScoringRun.request).where(ScoringRun.id == body["run_id"]))
    )
    assert score_env["ai_key"] not in response.text + json.dumps(stored) + caplog.text
    assert score_env["app_key"] not in response.text + caplog.text


@pytest.mark.parametrize(
    ("ai_key", "handler", "error_code"),
    [
        (False, lambda r: httpx.Response(500), "ai_key_missing"),
        (True, lambda r: httpx.Response(500), "ai_http_error"),
        (True, lambda r: _completion({"score": "bad"}), "ai_invalid_assessment"),
    ],
)
def test_ai_failure_degrades_to_rule_score(
    score_env, canonical_request, ai_key, handler, error_code
):
    with TestClient(score_env["app_for"](handler)) as client:
        response = _post(client, score_env, _hybrid(canonical_request), ai_key=ai_key)
    body = response.json()
    assert response.status_code == 200
    assert (body["result_mode"], body["error_code"], body["ai_score"]) == (
        "rules_only_fallback",
        error_code,
        None,
    )
    assert body["final_score"] == body["rule_score"] == 50


def test_missing_active_profile_degrades_to_rule_score(score_env, canonical_request):
    async def deactivate(db):
        for profile in (await db.scalars(select(AIProfile))).all():
            profile.active = False

    score_env["read"](deactivate)
    with TestClient(score_env["app_for"](_ai([]))) as client:
        body = _post(client, score_env, _hybrid(canonical_request), ai_key=True).json()
    assert (body["result_mode"], body["error_code"]) == (
        "rules_only_fallback",
        "ai_profile_missing",
    )


def test_same_key_replays_without_calling_ai_and_a_changed_body_is_422(
    score_env, canonical_request
):
    calls: list[httpx.Request] = []
    request, key = _hybrid(canonical_request), str(uuid4())
    changed = {**request, "metadata": {"other": 1}}
    with TestClient(score_env["app_for"](_ai(calls))) as client:
        first = _post(client, score_env, request, key=key, ai_key=True)
        again = _post(client, score_env, request, key=key, ai_key=True)
        conflict = _post(client, score_env, changed, key=key, ai_key=True)
    assert first.status_code == again.status_code == 200
    assert again.json() == first.json()
    assert len(calls) == 1
    assert conflict.status_code == 422


def test_concurrent_requests_with_one_key_store_one_run(score_env, canonical_request):
    request, key = _hybrid(canonical_request), str(uuid4())
    statuses: list[int] = []

    def send() -> None:
        with TestClient(score_env["app_for"](_ai([]))) as client:
            statuses.append(_post(client, score_env, request, key=key, ai_key=True).status_code)

    threads = [threading.Thread(target=send) for _ in range(2)]
    [t.start() for t in threads]
    [t.join() for t in threads]
    count = score_env["read"](
        lambda db: db.scalar(
            select(func.count()).select_from(ScoringRun).where(ScoringRun.idempotency_key == key)
        )
    )
    assert (statuses, count) == ([200, 200], 1)


def test_auth_and_input_errors(score_env, canonical_request):
    legacy = {**canonical_request, "account_id": "3"}
    with TestClient(score_env["app_for"](_ai([]))) as client:
        no_key = client.post(URL, headers={"Idempotency-Key": str(uuid4())}, json=canonical_request)
        bad_key = _post(client, score_env, canonical_request, Authorization="Bearer wrong")
        extra_field = _post(client, score_env, legacy)
        long_ai_key = client.post(
            URL,
            headers={**score_env["headers"](), "X-AI-Key": "k" * 5_000},
            json=canonical_request,
        )
    assert (no_key.status_code, bad_key.status_code) == (401, 401)
    assert (extra_field.status_code, long_ai_key.status_code) == (422, 422)
    assert all(set(error) == {"loc", "msg"} for error in extra_field.json()["detail"])