2268b6a77a
Add app credentials, account-scoped idempotency and per-app terminal notifications. Consolidate the unreleased migration baseline, preserve rule descriptions as AI context, and regenerate provider contracts with regression coverage.
96 lines
4.0 KiB
Python
96 lines
4.0 KiB
Python
from copy import deepcopy
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from ai_platform.lead_scoring.application.scoring import calculate_run
|
|
from ai_platform.presentation.http.run_responses import RunDetail, RunSummary
|
|
from ai_platform.presentation.http.schemas import RunRequest
|
|
|
|
FIXTURE_DIR = Path(__file__).resolve().parents[1] / "contracts/lead-scoring/v2/fixtures"
|
|
|
|
|
|
@pytest.mark.parametrize("use_ai", [False, True])
|
|
def test_web_fixture_matches_run_request_contract(use_ai):
|
|
request = RunRequest.model_validate_json(
|
|
(FIXTURE_DIR / "run-request.json").read_text(encoding="utf-8")
|
|
)
|
|
payload = request.model_dump(mode="json")
|
|
if use_ai:
|
|
payload["scoring_config"].update(rule_weight=0.5, ai_weight=0.5)
|
|
payload["ai_profile"] = "medium"
|
|
RunRequest.model_validate(payload)
|
|
score, _ = calculate_run(payload)
|
|
assert score == 50
|
|
|
|
|
|
def test_web_fixture_has_ordered_conversation_with_unique_messages():
|
|
request = RunRequest.model_validate_json((FIXTURE_DIR / "run-request.json").read_text())
|
|
messages = [message for chat in request.snapshot.conversations for message in chat.messages]
|
|
assert len(messages) == 12
|
|
assert len({message.message_id for message in messages}) == len(messages)
|
|
assert {message.sender_type for message in messages} == {"agent", "contact"}
|
|
sent_at = [message.sent_at for message in messages]
|
|
assert sent_at == sorted(sent_at)
|
|
assert all(0 <= (request.snapshot.captured_at - at).days < 90 for at in sent_at)
|
|
|
|
|
|
@pytest.mark.parametrize("value", [None, "", " ", 3, "x" * 256])
|
|
def test_invalid_account_ids_are_rejected(valid_run_payload, value):
|
|
with pytest.raises(ValidationError):
|
|
RunRequest.model_validate({**valid_run_payload, "account_id": value})
|
|
|
|
|
|
def test_account_identity_is_preserved(valid_run_payload):
|
|
request = RunRequest.model_validate({**valid_run_payload, "account_id": " Account-3 "})
|
|
assert request.account_id == " Account-3 "
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["app_id", "tenant_id", "callback_url"])
|
|
def test_self_supplied_identity_and_callback_are_rejected(valid_run_payload, field):
|
|
with pytest.raises(ValidationError):
|
|
RunRequest.model_validate({**valid_run_payload, field: "supplied"})
|
|
|
|
|
|
def test_subject_must_match_snapshot_lead(valid_run_payload):
|
|
with pytest.raises(ValidationError):
|
|
RunRequest.model_validate({**valid_run_payload, "subject_ref": "another-lead"})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value", ["F81C7C4E-A92A-4B19-8D7A-3A5C08EB4E6E", "f81c7c4ea92a4b198d7a3a5c08eb4e6e", "bad"]
|
|
)
|
|
def test_request_id_requires_canonical_uuid(valid_run_payload, value):
|
|
with pytest.raises(ValidationError):
|
|
RunRequest.model_validate({**valid_run_payload, "request_id": value})
|
|
|
|
|
|
def test_response_examples_match_provider_models():
|
|
summary = RunSummary.model_validate_json((FIXTURE_DIR / "run-summary.json").read_text())
|
|
detail = RunDetail.model_validate_json((FIXTURE_DIR / "run-detail.json").read_text())
|
|
assert summary.app_id == detail.app_id == "omni-local"
|
|
assert summary.account_id == detail.account_id == "3"
|
|
assert summary.request_id == detail.request_id
|
|
assert summary.run_id == detail.run_id
|
|
request = RunRequest.model_validate_json((FIXTURE_DIR / "run-request.json").read_text())
|
|
score, breakdown = calculate_run(request.model_dump(mode="json"))
|
|
assert summary.rule_score == detail.rule_score == score
|
|
assert summary.final_score == detail.final_score == score
|
|
assert summary.breakdown == detail.breakdown == breakdown
|
|
assert detail.snapshot == request.snapshot
|
|
|
|
|
|
def test_run_request_contains_caller_rules_and_weights(valid_run_payload):
|
|
request = RunRequest.model_validate(valid_run_payload)
|
|
assert request.scoring_config.rules[0].points == 25
|
|
assert request.scoring_config.ai_weight == 0
|
|
|
|
|
|
def test_run_request_requires_weights_to_sum_to_one(valid_run_payload):
|
|
data = deepcopy(valid_run_payload)
|
|
data["scoring_config"].update(rule_weight=0.5, ai_weight=0.4)
|
|
|
|
with pytest.raises(ValidationError):
|
|
RunRequest.model_validate(data)
|