zhifu-platform/tests/test_business_boundaries.py

from datetime import UTC, datetime, timedelta
from uuid import UUID, uuid4

import pytest
import pytest_asyncio
from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401
from test_fulfillment import connected  # noqa: F401

from zhifu.core.errors import BusinessError
from zhifu.modules.after_sales.models import ExchangeRule, OrderProjection, Ticket
from zhifu.modules.after_sales.service import AfterSalesService
from zhifu.modules.integrations.client import OmsClient
from zhifu.modules.integrations.models import ExternalOperation, Reservation
from zhifu.oms.models import Exchange

pytestmark = pytest.mark.integration


@pytest.fixture(scope="module")
def baseline():
    return datetime.now(UTC)


@pytest_asyncio.fixture(params=["A", "B"])
async def merchant(connected, request):  # noqa: F811
    # 两个商家分别使用独立随机租户与真实 OMS 凭据，规则值来自需求 R1～R5。
    name = request.param
    days = 7 if name == "A" else 15
    citation = f"商家 {name} R1～R5：签收后 {days} 天内受理功能故障，必须客户确认及主管审批。"
    response = await connected["call"](
        "POST", f"/admin/v1/stores/{connected['store']}/exchange-rules", "agent",
        {"expected_version": 1, "window_days": days,
         "product_types": ["keyboard"] if name == "A" else ["keyboard", "mouse"],
         "citation": citation},
    )
    assert response.status_code == 201, response.text
    return {**connected, "merchant": name, "days": days, "citation": citation}


async def order(context, **changes):
    identifier = uuid4()
    source = {**context["source"], **changes}
    response = await context["warehouse"].put(
        f"/orders/{identifier}",
        json={key: value for key, value in source.items()
              if key not in {"store_id", "source_system"}},
    )
    assert response.status_code == 200, response.text
    response = await context["warehouse"].get(f"/orders/{identifier}")
    assert response.status_code == 200, response.text
    if source["signed_at"] is None:
        assert response.json()["signed_at"] is None
    else:
        assert datetime.fromisoformat(response.json()["signed_at"]) == datetime.fromisoformat(
            source["signed_at"])
    response = await context["call"](
        "PUT", f"/internal/v1/orders/{identifier}", "connector", source,
    )
    assert response.status_code == 200, response.text
    return {**context["proposal"], "order_id": str(identifier), "item_id": source["item_id"]}


async def assert_no_execution(context, *, tickets=0, rejected_exchanges=0):
    async with context["app"].state.sessions() as session:
        for model, expected in ((Ticket, tickets), (ExternalOperation, 0), (Reservation, 0)):
            count = await session.scalar(select(func.count()).select_from(model).where(
                model.tenant_id == context["tenant"].bytes))
            assert count == expected, model.__name__
    async with context["oms_sessions"]() as session:
        assert await session.scalar(select(func.count()).select_from(Exchange).where(
            Exchange.tenant_id == context["tenant"].bytes)) == rejected_exchanges
        assert await session.scalar(select(func.count()).select_from(Exchange).where(
            Exchange.tenant_id == context["tenant"].bytes,
            Exchange.status.in_(["ACCEPTED", "COMPLETED"]))) == 0


async def test_inclusive_time_boundary_uses_persisted_rule_and_order(merchant):
    baseline = datetime.now(UTC).replace(tzinfo=None)
    proposal = await order(
        merchant, signed_at=(baseline - timedelta(days=merchant["days"])).replace(
            tzinfo=UTC).isoformat(),
    )
    async with merchant["app"].state.sessions() as session:
        current = await session.get(OrderProjection, UUID(proposal["order_id"]).bytes)
        rule = await session.get(ExchangeRule, (
            merchant["tenant"].bytes, merchant["store"].bytes, 2))
        service = AfterSalesService(session, None)
        # 直接调用生产规则函数的明确时间参数，精确验证微秒边界，不更改系统时钟。
        service.evaluate(current, rule, 1, evaluated_at=baseline)
        service.evaluate(current, rule, 1, evaluated_at=baseline - timedelta(microseconds=1))
        with pytest.raises(BusinessError) as error:
            service.evaluate(current, rule, 1, evaluated_at=baseline + timedelta(microseconds=1))
        assert error.value.code == "RULE_NOT_SATISFIED"
        assert merchant["citation"] in error.value.message
    await assert_no_execution(merchant)


async def test_same_ten_day_order_obeys_each_merchant_window(merchant, baseline):
    proposal = await order(
        merchant, signed_at=(baseline - timedelta(days=10)).isoformat())
    response = await merchant["call"](
        "POST", "/openapi/v1/after-sales/proposals", payload=proposal)
    accepted = merchant["merchant"] == "B"
    assert response.status_code == (201 if accepted else 422), response.text
    if accepted:
        assert response.json()["citation"] == merchant["citation"]
        assert response.json()["rule_version"] == 2
        assert response.json()["status"] == "DRAFT"
    else:
        assert response.json()["code"] == "RULE_NOT_SATISFIED"
        assert merchant["citation"] in response.json()["message"]
    await assert_no_execution(merchant, tickets=int(accepted))


async def test_mouse_product_scope_obeys_each_merchant_rule(merchant):
    proposal = await order(merchant, item_id="mouse-1", product_type="mouse")
    response = await merchant["call"](
        "POST", "/openapi/v1/after-sales/proposals", payload=proposal)
    accepted = merchant["merchant"] == "B"
    assert response.status_code == (201 if accepted else 422), response.text
    if accepted:
        assert response.json()["citation"] == merchant["citation"]
    else:
        assert merchant["citation"] in response.json()["message"]
    await assert_no_execution(merchant, tickets=int(accepted))


async def test_rejection_matrix_preserves_quantity_and_oms(merchant):
    cases = [
        ({"signed_at": None}, {}, 422, "RULE_NOT_SATISFIED"),
        ({"signed_at": (datetime.now(UTC) + timedelta(days=1)).isoformat()}, {},
         422, "RULE_NOT_SATISFIED"),
        ({"signed_at": (datetime.now(UTC) - timedelta(
            days=merchant["days"], seconds=1)).isoformat()}, {}, 422, "RULE_NOT_SATISFIED"),
        ({"available_quantity": 0}, {}, 409, "QUANTITY_CONFLICT"),
        ({}, {"quantity": 2}, 409, "QUANTITY_CONFLICT"),
        ({"product_type": "headset"}, {}, 422, "RULE_NOT_SATISFIED"),
        ({}, {"item_id": "unrelated-item"}, 404, "RESOURCE_NOT_FOUND"),
        ({}, {"quantity": 0}, 422, None),
        ({}, {"reason": "  \n  "}, 422, None),
        ({}, {"reason": "故障" * 1001}, 422, None),
        ({}, {"evaluated_at": "2026-01-01T00:00:00Z"}, 422, None),
    ]
    for source_changes, changes, status, code in cases:
        proposal = await order(merchant, **source_changes)
        response = await merchant["call"](
            "POST", "/openapi/v1/after-sales/proposals", payload={**proposal, **changes})
        assert response.status_code == status, response.text
        if code:
            assert response.json()["code"] == code
            if code in {"RULE_NOT_SATISFIED", "QUANTITY_CONFLICT"}:
                assert merchant["citation"] in response.json()["message"]
        async with merchant["app"].state.sessions() as session:
            current = await session.get(OrderProjection, UUID(proposal["order_id"]).bytes)
            assert current.available_quantity == source_changes.get("available_quantity", 1)
    await assert_no_execution(merchant)


async def test_fault_description_and_rule_reference_are_saved(merchant):
    response = await merchant["call"](
        "POST", "/openapi/v1/after-sales/proposals",
        payload={**merchant["proposal"], "reason": "  键盘接通电源后多个按键无法输入  "})
    assert response.status_code == 201, response.text
    proposal = response.json()
    response = await merchant["call"]("GET", f"/openapi/v1/tickets/{proposal['ticket_id']}")
    assert response.status_code == 200, response.text
    assert response.json()["reason"] == "键盘接通电源后多个按键无法输入"
    assert response.json()["citation"] == merchant["citation"]
    assert response.json()["rule_version"] == 2
    await assert_no_execution(merchant, tickets=1)


async def test_oms_rejects_unsigned_order_and_preserves_replay(connected):  # noqa: F811
    for signed_at in (None, (datetime.now(UTC) + timedelta(days=1)).isoformat()):
        proposal = await order(connected, signed_at=signed_at)
        client = OmsClient("http://127.0.0.1:8100", connected["variable"], 5)
        try:
            snapshot = await client.order(UUID(proposal["order_id"]).bytes)
            assert (snapshot.signed_at is None) == (signed_at is None)
            operation = uuid4()
            payload = {**proposal, "subject_id": str(connected["accounts"]["customer"]),
                       "expected_order_version": 1}
            result = await client.process(operation.bytes, payload)
            assert result.status == "FAILED" and result.reason == "ORDER_NOT_SIGNED"
            repeated = await client.process(operation.bytes, payload)
            assert repeated == result
            current = await client.order(UUID(proposal["order_id"]).bytes)
            assert current.available_quantity == 1 and current.source_version == 1
        finally:
            await client.close()
    await assert_no_execution(connected, rejected_exchanges=2)