zhifu-platform/tests/test_agent_order_eligibility.py

import json
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401
from test_agent_runtime import tool_run
from test_business_boundaries import order
from test_fulfillment import connected  # noqa: F401

from zhifu.modules.after_sales.models import Ticket
from zhifu.modules.agent_runtime.models import AgentRun
from zhifu.modules.agent_runtime.runtime import build_tool_graph
from zhifu.modules.agent_runtime.worker import claim


async def lookup(context, proposal, quantity=1):
    run = await tool_run(context, "get_order", {
        "order_id": proposal["order_id"], "quantity": quantity})
    sessions = context["app"].state.sessions
    run_id, token = await claim(sessions, 2, run_ids=[run.id])
    await build_tool_graph(sessions).ainvoke({"run_id": run_id, "token": token})
    async with sessions() as session:
        current = await session.get(AgentRun, run_id)
        result = json.loads(current.messages[-1]["data"]["content"])
        assert await session.scalar(select(func.count()).select_from(Ticket).where(
            Ticket.tenant_id == context["tenant"].bytes)) == 0
    return result


@pytest.mark.parametrize("condition,code", [
    ("valid", None), ("expired", "RULE_NOT_SATISFIED"),
    ("unsigned", "RULE_NOT_SATISFIED"), ("quantity", "QUANTITY_CONFLICT"),
    ("product", "RULE_NOT_SATISFIED"),
])
async def test_order_tool_uses_real_current_rule_and_business_eligibility(
    connected, condition, code,  # noqa: F811
):
    changes = {"signed_at": (datetime.now(UTC) - timedelta(days=3)).isoformat()}
    if condition == "expired":
        changes["signed_at"] = (datetime.now(UTC) - timedelta(days=8)).isoformat()
    if condition == "unsigned":
        changes["signed_at"] = None
    if condition == "product":
        changes["product_type"] = "excluded-product"
    proposal = await order(connected, **changes)
    result = await lookup(connected, proposal, 2 if condition == "quantity" else 1)
    assert result["product_type"] == changes.get("product_type", "keyboard")
    assert (result["signed_at"] is None) == (condition == "unsigned")
    assert result["exchange_rule"]["version"] == 1
    assert result["exchange_rule"]["window_days"] == 7
    assert result["exchange_rule"]["citation"]
    eligibility = result["exchange_eligibility"]
    assert eligibility["eligible"] == (code is None)
    assert eligibility["code"] == code
    assert eligibility["evaluated_at"].endswith("Z")
    if code:
        assert result["exchange_rule"]["citation"] in eligibility["message"]
        response = await connected["call"]("POST", "/openapi/v1/after-sales/proposals",
            payload={**proposal, "quantity": eligibility["quantity"]})
        assert response.status_code == eligibility["status"]
        assert response.json()["code"] == code


async def test_order_tool_reads_new_rule_version_and_creation_rechecks_revocation(connected):  # noqa: F811
    proposal = await order(connected,
        signed_at=(datetime.now(UTC) - timedelta(days=10)).isoformat())
    assert (await lookup(connected, proposal))["exchange_eligibility"]["eligible"] is False
    path = f"/admin/v1/stores/{connected['store']}/exchange-rules"
    response = await connected["call"]("POST", path, "agent", {
        "expected_version": 1, "window_days": 15, "product_types": ["keyboard"],
        "citation": "当前十五天内换货规则",
    })
    assert response.status_code == 201, response.text
    current = await lookup(connected, proposal)
    assert current["exchange_eligibility"]["eligible"] is True
    assert current["exchange_rule"]["version"] == 2
    # 查询成功后撤销规则，创建事务必须重新检查当前规则有效性。
    response = await connected["call"]("POST", path + "/revoke", "agent", {
        "expected_version": 2, "reason": "核对提交时规则有效性"})
    assert response.status_code == 200, response.text
    response = await connected["call"](
        "POST", "/openapi/v1/after-sales/proposals", payload=proposal)
    assert response.status_code == 422 and response.json()["code"] == "RULE_NOT_SATISFIED"