175 lines
6.5 KiB
Python
175 lines
6.5 KiB
Python
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)
|
|
reordered = dict(reversed(list(request.items())))
|
|
again = _post(client, score_env, reordered, 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"])
|