zhifu-platform/tests/message_recovery_acceptance.py

import asyncio
import json
from pathlib import Path
from uuid import UUID, uuid4

import pytest
from sqlalchemy import func, select
from test_after_sales import scenario  # noqa: F401
from test_fulfillment import approve, connected, due, operation_for  # noqa: F401

from zhifu.core.config import Settings
from zhifu.modules.after_sales.models import BusinessEvent
from zhifu.modules.integrations.fulfillment import execute_due
from zhifu.modules.integrations.models import ConsumerReceipt, ExternalOperation
from zhifu.modules.integrations.worker import Broker, consume_one, publish_one
from zhifu.oms.models import Exchange, Order
from zhifu.operations.message_recovery import reconcile


async def test_lost_message_replay_and_duplicate_fulfillment(connected):  # noqa: F811
    sessions = connected["app"].state.sessions
    tenant = connected["tenant"].bytes
    ticket, event_id = await approve(connected)
    name = "recovery-" + uuid4().hex
    async with Broker(Settings().broker_url.get_secret_value(), name, name) as broker:
        try:
            pending = await reconcile(
                sessions, tenant, [event_id], operator="acceptance", reason="检查等待事件"
            )
            assert pending["items"][0]["disposition"] == "PENDING"
            assert not pending["items"][0]["requeued"]
            assert await publish_one(sessions, broker, event_id)
            # 仅清空本次创建的独立队列，实际制造发布已确认但消息不可再读取的场景。
            await broker.queue.purge()
            assert not await consume_one(sessions, broker, 5)
            report = await reconcile(sessions, tenant, [event_id])
            assert report["items"][0]["disposition"] == "SENT_WITHOUT_RECEIPT"
            with pytest.raises(ValueError, match="租户"):
                await reconcile(sessions, uuid4().bytes, [event_id], operator="test", reason="核对")
            with pytest.raises(ValueError, match="租户"):
                await reconcile(
                    sessions,
                    tenant,
                    [event_id, UUID(int=(1 << 128) - 1).bytes],
                    operator="acceptance",
                    reason="检查批量事务",
                )
            async with sessions() as session:
                assert (await session.get(BusinessEvent, event_id)).delivery_status == "SENT"
            reports = await asyncio.gather(
                *(
                    reconcile(
                        sessions, tenant, [event_id], operator="acceptance", reason="队列恢复演练"
                    )
                    for _ in range(2)
                )
            )
            assert sum(item["items"][0]["requeued"] for item in reports) == 1
            assert await publish_one(sessions, broker, event_id)
            assert await consume_one(sessions, broker, 5)
            operation = await operation_for(connected, ticket)
            assert await execute_due(sessions, 5, operation.id)
            async with sessions() as session:
                event = await session.get(BusinessEvent, event_id)
                await broker.publish_event(event)
                await broker.publish_event(event)
            assert await consume_one(sessions, broker, 5)
            assert await consume_one(sessions, broker, 5)
            skipped = await reconcile(
                sessions, tenant, [event_id], operator="acceptance", reason="再次核对"
            )
            assert skipped["items"][0]["disposition"] == "CONSUMED"
            assert not skipped["items"][0]["requeued"]
            async with sessions() as session:
                for model, predicate in (
                    (ConsumerReceipt, ConsumerReceipt.event_id == event_id),
                    (ExternalOperation, ExternalOperation.tenant_id == tenant),
                    (
                        BusinessEvent,
                        (BusinessEvent.tenant_id == tenant)
                        & (BusinessEvent.event_type == "fulfillment.replay_requested"),
                    ),
                ):
                    assert (
                        await session.scalar(
                            select(func.count()).select_from(model).where(predicate)
                        )
                        == 1
                    )
            async with connected["oms_sessions"]() as session:
                assert (
                    await session.scalar(
                        select(func.count())
                        .select_from(Exchange)
                        .where(Exchange.tenant_id == tenant)
                    )
                    == 1
                )
                order = await session.get(Order, connected["order_id"].bytes)
                assert order.available_quantity == 0
                assert order.source_version == 2
            response = await connected["warehouse"].post(
                f"/exchanges/{UUID(bytes=operation.id)}/complete"
            )
            assert response.status_code == 200
            await due(connected, operation.id)
            assert await execute_due(sessions, 5, operation.id)
            response = await connected["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")
            assert response.json()["status"] == "COMPLETED"
            await asyncio.to_thread(
                Path("docs/verification/message-recovery.json").write_text,
                json.dumps(
                    {
                        "queue_loss_recovered": True,
                        "concurrent_replay_changes": 1,
                        "consumer_receipts": 1,
                        "external_operations": 1,
                        "oms_exchanges": 1,
                        "ticket_status": "COMPLETED",
                        "foreign_tenant_rejected": True,
                        "invalid_batch_atomic": True,
                        "pending_event_skipped": True,
                    },
                    indent=2,
                ),
                encoding="utf-8",
            )
        finally:
            await broker.queue.delete(if_unused=False, if_empty=False)
            await broker.exchange.delete(if_unused=False)