X-Financial/server/tests/test_steward_query_executors.py

from __future__ import annotations

from types import SimpleNamespace

from app.schemas.steward import StewardActionExecuteRequest, StewardTask
from app.services.steward_query_executors import (
    build_travel_standard_query_steps,
    execute_travel_standard_query,
    resolve_travel_standard_snapshot,
)


def _build_current_user(grade: str = "P5") -> SimpleNamespace:
    return SimpleNamespace(
        username="test_user",
        name="测试员工",
        grade=grade,
        department_name="技术部",
        position="工程师",
        role_codes=["employee"],
        is_admin=False,
        employee_no="E00001",
        manager_name="李总",
    )


def _build_request(
    *,
    location: str = "武汉",
    employee_grade: str = "",
    standard_category: str = "",
    message: str = "我去武汉出差的住宿标准是多少",
) -> StewardActionExecuteRequest:
    ontology_fields: dict[str, str] = {}
    if location:
        ontology_fields["location"] = location
    if employee_grade:
        ontology_fields["employee_grade"] = employee_grade
    if standard_category:
        ontology_fields["standard_category"] = standard_category
    task = StewardTask(
        task_id="task_query_001",
        task_type="query_travel_standard",
        assigned_agent="policy_query_assistant",
        title="差旅标准查询",
        summary=message,
        ontology_fields=ontology_fields,
    )
    return StewardActionExecuteRequest(
        action_type="execute_travel_standard_query",
        message=message,
        task=task,
    )


def test_resolve_travel_standard_snapshot_returns_lodging_and_transport_for_known_city_and_grade():
    snapshot = resolve_travel_standard_snapshot(
        location="武汉",
        employee_grade="P5",
    )
    assert snapshot["location"] == "武汉"
    assert snapshot["city_tier"] == "tier_2"  # 武汉是二类城市
    assert snapshot["employee_grade"] == "P5"
    assert snapshot["lodging"] is not None
    assert snapshot["lodging"]["daily_cap"] == "480.00"  # P5 + tier_2
    assert snapshot["transport"] is not None
    assert snapshot["transport"]["flight_level"] == 1
    assert snapshot["matched_any"] is True


def test_resolve_travel_standard_snapshot_filters_by_standard_category():
    lodging_only = resolve_travel_standard_snapshot(
        location="北京",
        employee_grade="P7",
        standard_category="lodging",
    )
    assert lodging_only["lodging"] is not None
    assert lodging_only["lodging"]["daily_cap"] == "900.00"  # P7 + tier_1
    assert lodging_only["transport"] is None

    transport_only = resolve_travel_standard_snapshot(
        location="北京",
        employee_grade="P7",
        standard_category="transport",
    )
    assert transport_only["transport"] is not None
    assert transport_only["transport"]["flight_level"] == 3  # P7 飞机等级
    assert transport_only["lodging"] is None


def test_resolve_travel_standard_snapshot_normalizes_grade_variants():
    # "p5" 小写、未标准化写法应被归一为 "P5"
    snapshot = resolve_travel_standard_snapshot(
        location="武汉",
        employee_grade="p5",
    )
    assert snapshot["employee_grade"] == "P5"
    assert snapshot["lodging"]["daily_cap"] == "480.00"


def test_resolve_travel_standard_snapshot_handles_unknown_grade():
    snapshot = resolve_travel_standard_snapshot(
        location="武汉",
        employee_grade="",
    )
    # 无职级时无法匹配住宿标准(需要职级档位);补助为占位说明,不计入 matched_any
    assert snapshot["lodging"] is None
    assert snapshot["matched_any"] is False
    assert snapshot["allowance"] is not None  # 仍返回占位说明


def test_execute_travel_standard_query_returns_succeeded_with_answer_markdown():
    request = _build_request(location="武汉", employee_grade="P5")
    response = execute_travel_standard_query(
        executor=None,
        request=request,
        current_user=_build_current_user("P5"),
        trace=[],
    )
    assert response.status == "succeeded"
    assert response.action_type == "execute_travel_standard_query"
    assert "差旅标准查询结果" in response.message
    assert "480.00" in response.message  # 住宿标准
    assert response.result_payload["matched"] is True
    assert response.result_payload["standards"]["lodging"]["daily_cap"] == "480.00"


def test_execute_travel_standard_query_falls_back_to_current_user_grade_when_field_missing():
    request = _build_request(location="武汉", employee_grade="")
    response = execute_travel_standard_query(
        executor=None,
        request=request,
        current_user=_build_current_user("P5"),
        trace=[],
    )
    assert response.status == "succeeded"
    # 应回退到 current_user.grade = P5
    assert response.result_payload["standards"]["employee_grade"] == "P5"


def test_execute_travel_standard_query_returns_no_match_when_grade_and_city_unknown():
    request = _build_request(
        location="未知城市",
        employee_grade="",
        message="查差旅标准",
    )
    # ontology_fields 也没有 grade,current_user 也没有
    response = execute_travel_standard_query(
        executor=None,
        request=request,
        current_user=_build_current_user(""),
        trace=[],
    )
    assert response.status == "succeeded"
    assert response.result_payload["matched"] is False
    assert "未能匹配" in response.message


def test_build_travel_standard_query_steps_generates_single_executable_step():
    task = StewardTask(
        task_id="task_query_001",
        task_type="query_travel_standard",
        assigned_agent="policy_query_assistant",
        title="差旅标准查询",
        summary="查武汉住宿标准",
        ontology_fields={"location": "武汉", "employee_grade": "P5"},
    )
    steps = build_travel_standard_query_steps(task)
    assert len(steps) == 1
    assert steps[0].action_type == "execute_travel_standard_query"
    assert steps[0].requires_confirmation is False
    assert steps[0].payload["ontology_fields"]["location"] == "武汉"