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