zhifu-platform/tests/test_channel_acceptance.py

import asyncio
from uuid import UUID, uuid4

import pytest
from langchain_core.messages import AIMessage, message_to_dict
from sqlalchemy import func, select
from test_after_sales import confirm_proposal, make_proposal, scenario  # noqa: F401
from test_channel_inbox import channel, envelope  # noqa: F401
from test_fulfillment import connected, due, operation_for  # noqa: F401
from test_platform_management import receiver  # noqa: F401
from test_storefront import challenge, proof

from zhifu.modules.after_sales.models import BusinessEvent, 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
from zhifu.modules.conversations.models import Conversation, ConversationEvent
from zhifu.modules.identity.access_models import Principal, StoreGrant
from zhifu.modules.identity.auth import password_hasher
from zhifu.modules.integrations.fulfillment import execute_due, prepare_event
from zhifu.modules.integrations.models import ConsumerReceipt, ExternalOperation, Reservation
from zhifu.modules.merchant.models import ChannelApplication
from zhifu.modules.platform.delivery import enqueue, execute_one
from zhifu.modules.platform.models import ChannelDelivery
from zhifu.oms.models import Exchange

pytestmark = pytest.mark.integration


async def service_account(context):
    actor_id = uuid4()
    password = uuid4().hex
    async with context["app"].state.sessions.begin() as session:
        session.add(Principal(
            id=actor_id.bytes, tenant_id=context["tenant"].bytes,
            username="channel-service-" + actor_id.hex,
            password_hash=password_hasher.hash(password), kind="SERVICE",
        ))
        await session.flush()
        for permission in ("identities:verify", "tickets:read"):
            session.add(StoreGrant(
                tenant_id=context["tenant"].bytes, store_id=context["store"].bytes,
                principal_id=actor_id.bytes, permission=permission,
            ))
    response = await context["client"].post("/openapi/v1/auth/sessions", json={
        "tenant_id": str(context["tenant"]), "username": "channel-service-" + actor_id.hex,
        "password": password,
    })
    assert response.status_code == 200, response.text
    return {"Authorization": "Bearer " + response.json()["access_token"]}


async def test_h5_webhook_business_query_and_independent_histories(channel):  # noqa: F811
    async with asyncio.timeout(40):
        call = channel["call"]
        ticket = await make_proposal(channel)
        # 两个真实 SERVICE 账户具有独立访问令牌，共用已授权的通用 webhook 绑定。
        second_service = await service_account(channel)
        requested = await challenge(channel)
        verify_path = (
            f"/openapi/v1/channel/identity-challenges/{requested['challenge_id']}/verify"
        )
        response = await channel["client"].post(
            verify_path, json=proof(channel), headers=second_service)
        assert response.status_code == 200 and response.json()["status"] == "VERIFIED"
        confirm = await call(
            "POST", f"/openapi/v1/me/identity-challenges/{requested['challenge_id']}/confirm")
        assert confirm.status_code == 200 and confirm.json()["status"] == "LINKED"
        path = (f"/openapi/v1/channel/identities/tickets?store_id={channel['store']}"
                "&external_subject=verified-channel-account")
        results = await asyncio.gather(
            call("GET", path, "connector"),
            channel["client"].get(path, headers=second_service),
            call("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}"),
        )
        assert all(item.status_code == 200 for item in results)
        assert results[0].json()["items"] == results[1].json()["items"] == [results[2].json()]
        assert (await call("GET", path, "other")).status_code == 403
        assert (await call("GET", path, "foreign")).status_code == 403
        assert (await call("GET", path.replace("verified-channel-account", "unlinked"),
                           "connector")).status_code == 404
        assert (await call("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}",
                           "other")).status_code == 404

        h5 = await call("POST", "/openapi/v1/conversations", payload={
            "store_id": str(channel["store"])})
        assert h5.status_code == 201
        h5_id = h5.json()["conversation_id"]
        sent = await call("POST", f"/openapi/v1/conversations/{h5_id}/messages", payload={
            "text": "H5 会话专属消息", "expected_control_version": 1})
        assert sent.status_code == 201, sent.text
        external = await channel["send"](envelope(payload={"text": "渠道会话专属消息"}))
        assert external.status_code == 202, external.text
        external_id = external.json()["conversation_id"]
        assert external_id != h5_id
        history = await call("GET", f"/openapi/v1/conversations/{h5_id}/messages")
        assert "H5 会话专属消息" in history.text and "渠道会话专属消息" not in history.text
        external_history = await call(
            "GET", f"/openapi/v1/conversations/{external_id}/messages")
        assert external_history.status_code == 404
        listed = await call("GET", f"/openapi/v1/conversations?store_id={channel['store']}")
        assert [item["conversation_id"] for item in listed.json()["items"]] == [h5_id]
        async with channel["app"].state.sessions() as session:
            conversations = (await session.scalars(select(Conversation).where(
                Conversation.tenant_id == channel["tenant"].bytes))).all()
            assert sorted(item.channel for item in conversations) == ["h5", "webhook"]
            events = (await session.scalars(select(ConversationEvent).where(
                ConversationEvent.conversation_id == UUID(external_id).bytes,
                ConversationEvent.event_type == "message.user"))).all()
            assert len(events) == 1 and events[0].payload["text"] == "渠道会话专属消息"


async def channel_proposal(context):
    response = await context["send"](envelope(payload={"text": "申请键盘换货一件"}))
    assert response.status_code == 202, response.text
    sessions = context["app"].state.sessions
    async with sessions.begin() as session:
        conversation = await session.get(
            Conversation, UUID(response.json()["conversation_id"]).bytes)
        run = await session.get(AgentRun, conversation.active_run_id)
        # 验收真实工具节点及全部权限检查，不调用生成模型；工具参数来自固定验收订单。
        run.messages = run.messages + [message_to_dict(AIMessage(content="", tool_calls=[{
            "name": "create_exchange_proposal", "id": uuid4().hex,
            "args": context["proposal"], "type": "tool_call",
        }]))]
        run.next_node = "tools"
    run_id, token = await claim(sessions, 2)
    result = await build_tool_graph(sessions).ainvoke({"run_id": run_id, "token": token})
    assert result["next_node"] == "wait"
    async with sessions() as session:
        run = await session.get(AgentRun, run_id)
        assert run.model_calls == 0
        assert run.waiting["expected_input_type"] == "proposal_confirmation"
        proposal_id = run.waiting["proposal_id"]
    response = await context["call"]("GET", f"/openapi/v1/tickets/{proposal_id}")
    assert response.status_code == 200
    return response.json()


async def approve_ticket(context, ticket):
    response = await context["call"]("POST",
        f"/openapi/v1/approvals/{ticket['approval_id']}/decisions", "reviewer", {
            "proposal_version": ticket["proposal_version"], "expected_version": ticket["version"],
            "decision": "APPROVED", "reason": "核对客户确认及当前规则",
        })
    assert response.status_code == 200, response.text
    async with context["app"].state.sessions() as session:
        event = await session.scalar(select(BusinessEvent).where(
            BusinessEvent.aggregate_id == UUID(ticket["ticket_id"]).bytes,
            BusinessEvent.event_type == "after_sale.approved"))
        return event.id


async def test_h5_channel_compete_quantity_and_duplicate_consumption(channel, connected):  # noqa: F811
    async with asyncio.timeout(60):
        sessions = connected["app"].state.sessions
        h5_ticket = await make_proposal(connected)
        channel_ticket = await channel_proposal(channel)
        confirmed_h5 = await confirm_proposal(connected, h5_ticket)
        confirmation = envelope("proposal_confirmation", {
            "proposal_id": channel_ticket["proposal_id"],
            "proposal_version": channel_ticket["proposal_version"],
            "expected_ticket_version": channel_ticket["version"],
        })
        receipts = await asyncio.gather(*[channel["send"](confirmation) for _ in range(3)])
        assert all(item.status_code == 202 for item in receipts)
        assert len({item.json()["inbox_id"] for item in receipts}) == 1
        confirmed_channel = receipts[0].json()["result"]
        events = await asyncio.gather(
            approve_ticket(connected, confirmed_h5), approve_ticket(connected, confirmed_channel))
        # 两个入口在同一屏障后竞争数量，每个审批事件并发重投两次。
        barrier = asyncio.Barrier(4)

        async def consume(event):
            await barrier.wait()
            await prepare_event(sessions, event, 5)

        await asyncio.gather(*[consume(event) for event in events for _ in range(2)])
        async with sessions() as session:
            operations = (await session.scalars(select(ExternalOperation).where(
                ExternalOperation.tenant_id == channel["tenant"].bytes))).all()
            assert len(operations) == 1
            operation = operations[0]
            reservations = (await session.scalars(select(Reservation).where(
                Reservation.tenant_id == channel["tenant"].bytes))).all()
            assert sum(item.quantity for item in reservations) == 1
            tickets = (await session.scalars(select(Ticket).where(
                Ticket.tenant_id == channel["tenant"].bytes))).all()
            assert sorted(item.status for item in tickets) == ["FAILED", "PROCESSING"]
            assert await session.scalar(select(func.count()).select_from(ConsumerReceipt).where(
                ConsumerReceipt.event_id.in_(events))) == 2
        assert await execute_due(sessions, 5, operation.id)
        async with connected["oms_sessions"]() as session:
            exchanges = (await session.scalars(select(Exchange).where(
                Exchange.tenant_id == channel["tenant"].bytes))).all()
            assert len(exchanges) == 1 and exchanges[0].quantity == 1
        # 已批准事件晚到时继续复用消费收据，不重新建立操作或增加占用。
        await asyncio.gather(*[prepare_event(sessions, event, 5) for event in reversed(events)])
        async with sessions() as session:
            assert await session.scalar(select(func.count()).select_from(ExternalOperation).where(
                ExternalOperation.tenant_id == channel["tenant"].bytes)) == 1


async def test_cancelled_ticket_rejects_late_approval_event(channel, connected):  # noqa: F811
    async with asyncio.timeout(40):
        ticket = await confirm_proposal(connected, await channel_proposal(channel))
        event = await approve_ticket(connected, ticket)
        current = await connected["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")
        cancelled = await connected["call"](
            "POST", f"/openapi/v1/tickets/{ticket['ticket_id']}/cancel",
            payload={"expected_version": current.json()["version"], "reason": "客户取消申请"})
        assert cancelled.status_code == 200, cancelled.text
        sessions = connected["app"].state.sessions
        await asyncio.gather(*[prepare_event(sessions, event, 5) for _ in range(3)])
        assert await operation_for(connected, ticket) is None
        latest = await connected["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")
        assert latest.json()["status"] == "CANCELLED"
        async with connected["oms_sessions"]() as session:
            assert await session.scalar(select(func.count()).select_from(Exchange).where(
                Exchange.tenant_id == connected["tenant"].bytes)) == 0


async def test_channel_full_oms_flow_and_ordered_tls_callbacks(
    channel, connected, receiver,  # noqa: F811
):
    async with asyncio.timeout(60):
        sessions = connected["app"].state.sessions
        async with sessions.begin() as session:
            application = await session.get(ChannelApplication, (
                connected["tenant"].bytes, connected["store"].bytes))
            application.callback_url = receiver["url"]
            application.credential_env = receiver["credential_env"]
        ticket = await channel_proposal(channel)
        response = await channel["send"](envelope("proposal_confirmation", {
            "proposal_id": ticket["proposal_id"], "proposal_version": ticket["proposal_version"],
            "expected_ticket_version": ticket["version"],
        }))
        assert response.status_code == 202
        event = await approve_ticket(connected, response.json()["result"])
        await prepare_event(sessions, event, 5)
        operation = await operation_for(connected, ticket)
        assert await execute_due(sessions, 5, operation.id)
        completed = await connected["warehouse"].post(
            f"/exchanges/{UUID(bytes=operation.id)}/complete")
        assert completed.status_code == 200
        await due(connected, operation.id)
        assert await execute_due(sessions, 5, operation.id)
        query = (f"/openapi/v1/channel/identities/tickets?store_id={channel['store']}"
                 "&external_subject=verified-channel-account")
        queried = await connected["call"]("GET", query, "connector")
        assert queried.status_code == 200
        assert queried.json()["items"][0]["status"] == "COMPLETED"
        assert queried.json()["items"][0]["ticket_id"] == ticket["ticket_id"]
        assert await enqueue(sessions, tenant_id=connected["tenant"].bytes) > 0
        assert await enqueue(sessions, tenant_id=connected["tenant"].bytes) == 0
        async with sessions() as session:
            deliveries = (await session.scalars(select(ChannelDelivery).join(
                BusinessEvent, BusinessEvent.id == ChannelDelivery.event_id).where(
                    ChannelDelivery.tenant_id == connected["tenant"].bytes,
                    BusinessEvent.aggregate_id == UUID(ticket["ticket_id"]).bytes,
                ).order_by(BusinessEvent.aggregate_version))).all()
        assert len(deliveries) >= 5
        assert await execute_one(sessions, deliveries[-1].id) is False
        receiver["accepted"].set()
        for delivery in deliveries:
            assert await execute_one(sessions, delivery.id)
            assert await execute_one(sessions, delivery.id) is False
        events = receiver["received"]
        assert len(events) == len(deliveries)
        versions = [item["aggregate_version"] for item in events]
        assert versions == sorted(set(versions))
        assert {item["aggregate_id"] for item in events} == {ticket["ticket_id"]}
        assert events[-1]["payload"]["operation_status"] == "COMPLETED"
        async with sessions() as session:
            for delivery in deliveries:
                current = await session.get(ChannelDelivery, delivery.id)
                assert current.status == "SENT" and current.attempts == 1
        async with connected["oms_sessions"]() as session:
            assert await session.scalar(select(func.count()).select_from(Exchange).where(
                Exchange.tenant_id == connected["tenant"].bytes)) == 1