dotb_ai-platform/tests/test_request_contract.py

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": [1, 2, 3]}, "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})))