183 lines
7.2 KiB
Python
183 lines
7.2 KiB
Python
from copy import deepcopy
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from ai_platform.lead_scoring.application.scoring import calculate_run
|
|
from ai_platform.presentation.http.schemas import RunRequest
|
|
|
|
|
|
def _with(request, **changes):
|
|
data = deepcopy(request)
|
|
for path, value in changes.items():
|
|
*parents, leaf = path.split("__")
|
|
target = data
|
|
for part in parents:
|
|
target = target[int(part)] if part.isdigit() else target[part]
|
|
target[int(leaf) if leaf.isdigit() else leaf] = value
|
|
return data
|
|
|
|
|
|
def test_canonical_request_is_valid_and_scores_deterministically(canonical_request):
|
|
request = RunRequest.model_validate(canonical_request)
|
|
messages = [m for c in request.snapshot.conversations for m in c.messages]
|
|
assert len({m.message_id for m in messages}) == len(messages) == 12
|
|
assert [m.sent_at for m in messages] == sorted(m.sent_at for m in messages)
|
|
assert calculate_run(request.model_dump(mode="json"))[0] == 50
|
|
assert request.metadata == canonical_request["metadata"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"legacy",
|
|
[
|
|
{"request_id": "f81c7c4e-a92a-4b19-8d7a-3a5c08eb4e6e"},
|
|
{"account_id": "3"},
|
|
{"subject_ref": "lead-123"},
|
|
{"ai_profile": "medium"},
|
|
],
|
|
)
|
|
def test_client_specific_top_level_fields_are_rejected(canonical_request, legacy):
|
|
with pytest.raises(ValidationError, match="Extra inputs"):
|
|
RunRequest.model_validate({**canonical_request, **legacy})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change",
|
|
[
|
|
{"scoring_config__version": "v1"},
|
|
{"snapshot__version": "1.0.0"},
|
|
{"snapshot__lead_profile__ems_lead_id": "x"},
|
|
{"snapshot__conversations__0__assigned_agent_id": "x"},
|
|
{"snapshot__conversations__0__messages__0__extra": "x"},
|
|
],
|
|
)
|
|
def test_unmodeled_fields_are_rejected_not_discarded(canonical_request, change):
|
|
with pytest.raises(ValidationError, match="Extra inputs"):
|
|
RunRequest.model_validate(_with(canonical_request, **change))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("change", "error"),
|
|
[
|
|
({"snapshot__conversations": [{"conversation_id": str(i)} for i in range(11)]}, "10"),
|
|
({"snapshot__conversations__0__messages__0__content": "x" * 20_001}, "20000"),
|
|
({"snapshot__conversations__0__messages__0__sent_at": "2025-09-24T09:30:00"}, "timezone"),
|
|
({"snapshot__conversations__0__messages__0__sent_at": "2025-09-26T09:30:00Z"}, "after"),
|
|
({"metadata": {"blob": "x" * 10_001}}, "metadata exceeds"),
|
|
({"scoring_config__ai_weight": 0.4}, "sum to 1"),
|
|
({"scoring_config__rule_weight": True}, "valid number"),
|
|
],
|
|
)
|
|
def test_invalid_evidence_and_config_is_rejected(canonical_request, change, error):
|
|
with pytest.raises(ValidationError, match=error):
|
|
RunRequest.model_validate(_with(canonical_request, **change))
|
|
|
|
|
|
def test_old_messages_are_accepted(canonical_request):
|
|
data = _with(
|
|
canonical_request, snapshot__conversations__0__messages__0__sent_at="2020-01-01T00:00:00Z"
|
|
)
|
|
assert RunRequest.model_validate(data).snapshot.captured_at.year == 2025
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "operator", "value"),
|
|
[
|
|
("custom_fields.interest.region", "eq", "north"),
|
|
("lead_profile.external_id", "exists", None),
|
|
("metrics.contact_message_recency_hours", "gte", 1.5),
|
|
("metrics.customer_initiated_return", "eq", True),
|
|
],
|
|
)
|
|
def test_supported_rule_paths(canonical_request, field, operator, value):
|
|
rule = {"field": field, "operator": operator, "value": value}
|
|
request = RunRequest.model_validate(
|
|
_with(canonical_request, scoring_config__rules__0=_rule(rule))
|
|
)
|
|
assert request.scoring_config.rules[0].field == field
|
|
|
|
|
|
def _rule(overrides: dict) -> dict:
|
|
return {"id": "r", "points": 1, **overrides}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "operator", "value", "points"),
|
|
[
|
|
("lead_profile.custom_fields.x", "eq", "a", 1),
|
|
("metrics.customer_message_count", "exists", None, 1),
|
|
("metrics.unknown_metric", "eq", 1, 1),
|
|
("snapshot.conversations", "exists", None, 1),
|
|
("metrics.conversation_turn_count", "eq", 1.0, 1),
|
|
("metrics.customer_initiated_return", "eq", 1, 1),
|
|
("custom_fields.age", "gte", "18", 1),
|
|
("custom_fields.age", "eq", None, 1),
|
|
("custom_fields.region", "eq", {"nested": 1}, 1),
|
|
("custom_fields.age", "eq", 18, 1.5),
|
|
("custom_fields.region", "eq", "x" * 1_001, 1),
|
|
],
|
|
)
|
|
def test_invalid_rules_are_rejected(canonical_request, field, operator, value, points):
|
|
rule = _rule({"field": field, "operator": operator, "value": value, "points": points})
|
|
with pytest.raises(ValidationError):
|
|
RunRequest.model_validate(_with(canonical_request, scoring_config__rules__0=rule))
|
|
|
|
|
|
@pytest.mark.parametrize("description", [None, "", "ế" * 99])
|
|
def test_rule_description_is_optional_and_never_changes_the_score(canonical_request, description):
|
|
plain = RunRequest.model_validate(canonical_request).model_dump(mode="json")
|
|
data = _with(canonical_request, scoring_config__rules__0__description=description)
|
|
described = RunRequest.model_validate(data).model_dump(mode="json")
|
|
assert calculate_run(described) == calculate_run(plain)
|
|
|
|
|
|
@pytest.mark.parametrize("description", ["x" * 100, 99])
|
|
def test_rule_description_rejects_overlong_or_non_strings(canonical_request, description):
|
|
with pytest.raises(ValidationError):
|
|
RunRequest.model_validate(
|
|
_with(canonical_request, scoring_config__rules__0__description=description)
|
|
)
|
|
|
|
|
|
def _only_rule(**rule):
|
|
return {"scoring_config__rules": [{"id": "r", "points": 5, **rule}]}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"rule",
|
|
[
|
|
{"field": "custom_fields.x", "operator": "between", "value": [1, 2]},
|
|
{"field": "custom_fields.x", "operator": "in", "value": ["hot", 3, True]},
|
|
{"field": "metrics.message_count", "operator": "between", "value": [3, 10]},
|
|
{"field": "metrics.message_count", "operator": "in", "value": [3, 10]},
|
|
],
|
|
)
|
|
def test_between_and_in_rules_are_accepted(canonical_request, rule):
|
|
RunRequest.model_validate(_with(canonical_request, **_only_rule(**rule)))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("rule", "error"),
|
|
[
|
|
({"operator": "between", "value": [1]}, "min, max"),
|
|
({"operator": "between", "value": [5, 1]}, "exceed"),
|
|
({"operator": "between", "value": [1, "2"]}, "min, max"),
|
|
({"operator": "between", "value": [True, 2]}, "min, max"),
|
|
({"operator": "between", "value": 3}, "list"),
|
|
({"operator": "in", "value": []}, "non-empty"),
|
|
({"operator": "in", "value": ["a"] * 101}, "100"),
|
|
({"operator": "in", "value": [["a"]]}, "scalar"),
|
|
({"operator": "in", "value": "hot"}, "list"),
|
|
({"operator": "eq", "value": ["a"]}, "scalar"),
|
|
({"operator": "in", "value": [1.5], "field": "metrics.message_count"}, "integer"),
|
|
(
|
|
{"operator": "in", "value": [True], "field": "metrics.customer_initiated_return"},
|
|
"eq/ne",
|
|
),
|
|
],
|
|
)
|
|
def test_invalid_between_and_in_rules_are_rejected(canonical_request, rule, error):
|
|
base = {"field": "custom_fields.x"}
|
|
with pytest.raises(ValidationError, match=error):
|
|
RunRequest.model_validate(_with(canonical_request, **_only_rule(**{**base, **rule})))
|