zhifu-platform/tests/test_fulfillment.py

import asyncio
import hashlib
import os
import secrets
from argparse import Namespace
from datetime import timedelta
from uuid import UUID, uuid4

import httpx
import pytest
import pytest_asyncio
from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import async_sessionmaker
from test_after_sales import confirm_proposal, make_proposal, scenario  # noqa: F401

from zhifu.core.config import Settings
from zhifu.modules.after_sales.models import BusinessEvent, Ticket
from zhifu.modules.identity.access_models import StoreGrant
from zhifu.modules.identity.auth import now
from zhifu.modules.integrations.client import OmsClient
from zhifu.modules.integrations.fulfillment import execute_due, prepare_event, record_result
from zhifu.modules.integrations.manage import main as configure_connector
from zhifu.modules.integrations.models import (
    ConsumerReceipt,
    ExternalOperation,
    Reservation,
)
from zhifu.modules.integrations.worker import Broker, consume_one, publish_one
from zhifu.oms.database import oms_engine
from zhifu.oms.manage import main as register_credential
from zhifu.oms.models import Credential, Exchange, Order

pytestmark = pytest.mark.integration


@pytest_asyncio.fixture
async def connected(scenario):  # noqa: F811
    engine = oms_engine()
    sessions = async_sessionmaker(engine, expire_on_commit=False)
    token = secrets.token_urlsafe(32)
    warehouse_token = secrets.token_urlsafe(32)
    variable = "ZHIFU_OMS_" + uuid4().hex.upper()
    os.environ[variable] = token
    await register_credential(
        Namespace(
            tenant=scenario["tenant"],
            store=scenario["store"],
            credential_env=variable,
            fulfillment=False,
        )
    )
    async with sessions.begin() as session:
        session.add(
            Credential(
                token_hash=hashlib.sha256(warehouse_token.encode()).digest(),
                tenant_id=scenario["tenant"].bytes,
                store_id=scenario["store"].bytes,
                can_fulfill=True,
            )
        )
    await configure_connector(
        Namespace(
            tenant=scenario["tenant"],
            store=scenario["store"],
            source_system="oms",
            base_url="http://127.0.0.1:8100",
            credential_env=variable,
        )
    )
    try:
        async with httpx.AsyncClient(
            base_url="http://127.0.0.1:8100",
            headers={"Authorization": "Bearer " + warehouse_token},
            timeout=5,
        ) as warehouse:
            body = {
                name: value
                for name, value in scenario["source"].items()
                if name not in {"store_id", "source_system"}
            }
            response = await warehouse.put(f"/orders/{scenario['order_id']}", json=body)
            assert response.status_code == 200, response.text
            yield {
                **scenario,
                "warehouse": warehouse,
                "oms_sessions": sessions,
                "variable": variable,
            }
    finally:
        async with sessions.begin() as session:
            for model in (Exchange, Order, Credential):
                await session.execute(
                    delete(model).where(model.tenant_id == scenario["tenant"].bytes)
                )
        await engine.dispose()
        del os.environ[variable]


async def approve(connected):
    ticket = await confirm_proposal(connected, await make_proposal(connected))
    response = await connected["call"](
        "POST",
        f"/openapi/v1/approvals/{ticket['approval_id']}/decisions",
        "reviewer",
        {
            "proposal_version": 1,
            "expected_version": 2,
            "decision": "APPROVED",
            "reason": "审核通过",
        },
    )
    assert response.status_code == 200, response.text
    async with connected["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 ticket, event.id


async def operation_for(connected, ticket):
    async with connected["app"].state.sessions() as session:
        return await session.scalar(
            select(ExternalOperation).where(
                ExternalOperation.ticket_id == UUID(ticket["ticket_id"]).bytes
            )
        )


async def due(connected, operation_id):
    async with connected["app"].state.sessions.begin() as session:
        operation = await session.get(ExternalOperation, operation_id, with_for_update=True)
        operation.next_attempt_at = now() - timedelta(seconds=1)


async def test_broker_delivery_and_fulfillment(connected):
    ticket, event_id = await approve(connected)
    sessions = connected["app"].state.sessions
    queue_name = "verify-" + uuid4().hex
    async with Broker(Settings().broker_url.get_secret_value(), queue_name, queue_name) as broker:
        try:
            async with sessions() as session:
                event = await session.get(BusinessEvent, event_id)
                await broker.publish_event(event)
            assert await publish_one(sessions, broker, event_id)
            assert await consume_one(sessions, broker, 5)
            assert await consume_one(sessions, broker, 5)
            async with sessions() as session:
                assert (
                    await session.scalar(
                        select(func.count())
                        .select_from(ConsumerReceipt)
                        .where(ConsumerReceipt.event_id == event_id)
                    )
                    == 1
                )
                assert (await session.get(BusinessEvent, event_id)).delivery_status == "SENT"
            operation = await operation_for(connected, ticket)
            assert await execute_due(sessions, 5, operation.id)
            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)
            result = await connected["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")
            assert result.json()["status"] == "COMPLETED"
        finally:
            await broker.queue.delete(if_unused=False, if_empty=False)
            await broker.exchange.delete(if_unused=False)


async def test_two_proposals_cannot_reserve_same_quantity(connected):
    one, event_one = await approve(connected)
    two, event_two = await approve(connected)
    sessions = connected["app"].state.sessions
    await asyncio.gather(
        prepare_event(sessions, event_one, 5), prepare_event(sessions, event_two, 5)
    )
    async with sessions() as session:
        reservations = (
            await session.scalars(
                select(Reservation).where(Reservation.tenant_id == connected["tenant"].bytes)
            )
        ).all()
        assert sum(r.quantity for r in reservations) == 1
        tickets = (
            await session.scalars(
                select(Ticket).where(
                    Ticket.id.in_([UUID(one["ticket_id"]).bytes, UUID(two["ticket_id"]).bytes])
                )
            )
        ).all()
        assert sorted(t.status for t in tickets) == ["FAILED", "PROCESSING"]


async def test_external_condition_failure_releases_reservation(connected):
    ticket, event_id = await approve(connected)
    sessions = connected["app"].state.sessions
    await prepare_event(sessions, event_id, 5)
    operation = await operation_for(connected, ticket)
    async with connected["oms_sessions"].begin() as session:
        order = await session.get(Order, connected["order_id"].bytes, with_for_update=True)
        order.source_version += 1
    assert await execute_due(sessions, 5, operation.id)
    async with sessions() as session:
        reservation = await session.get(Reservation, operation.id)
        assert reservation.status == "RELEASED"
        assert (await session.get(Ticket, operation.ticket_id)).status == "FAILED"


async def test_expired_execution_cannot_overwrite_recovered_result(connected):
    ticket, event_id = await approve(connected)
    sessions = connected["app"].state.sessions
    await prepare_event(sessions, event_id, 5)
    operation = await operation_for(connected, ticket)
    async with sessions.begin() as session:
        current = await session.get(ExternalOperation, operation.id, with_for_update=True)
        current.lease_token = 1
        current.lease_until = now() - timedelta(seconds=1)
    client = OmsClient(operation.base_url, operation.credential_env, 5)
    try:
        result = await client.process(operation.id, operation.request, query_first=False)
    finally:
        await client.close()
    await record_result(sessions, operation.id, 1, result)
    async with sessions() as session:
        assert (await session.get(ExternalOperation, operation.id)).status == "PREPARED"
    assert await execute_due(sessions, 5, operation.id)
    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
        )


async def test_prepared_operation_blocks_cancellation(connected):
    ticket, event_id = await approve(connected)
    await prepare_event(connected["app"].state.sessions, event_id, 5)
    response = await connected["call"](
        "POST",
        f"/openapi/v1/tickets/{ticket['ticket_id']}/cancel",
        payload={"expected_version": 4, "reason": "取消"},
    )
    assert response.status_code == 409
    assert response.json()["code"] == "OPERATION_IN_PROGRESS"


async def test_lost_response_keeps_reservation_and_recovers(connected):
    ticket, event_id = await approve(connected)
    sessions = connected["app"].state.sessions
    await prepare_event(sessions, event_id, 5)
    operation = await operation_for(connected, ticket)
    async with httpx.AsyncClient(base_url="http://127.0.0.1:18474") as proxy:
        name = "oms-" + uuid4().hex
        response = await proxy.post(
            "/proxies",
            json={
                "name": name,
                "listen": "0.0.0.0:18100",
                "upstream": "host.docker.internal:8100",
            },
        )
        assert response.status_code == 201, response.text
        try:
            response = await proxy.post(
                f"/proxies/{name}/toxics",
                json={
                    "name": "response-delay",
                    "type": "latency",
                    "stream": "downstream",
                    "attributes": {"latency": 2000, "jitter": 0},
                },
            )
            assert response.status_code == 200, response.text
            async with sessions.begin() as session:
                current = await session.get(ExternalOperation, operation.id)
                current.base_url = "http://127.0.0.1:18100"
            assert await execute_due(sessions, 0.3, operation.id)
            async with sessions() as session:
                assert (await session.get(ExternalOperation, operation.id)).status == "UNKNOWN"
                assert (await session.get(Reservation, operation.id)).status == "HELD"
            async with connected["oms_sessions"]() as session:
                assert (await session.get(Exchange, operation.id)).status == "ACCEPTED"
            response = await proxy.delete(f"/proxies/{name}/toxics/response-delay")
            assert response.status_code == 204
            await due(connected, operation.id)
            assert await execute_due(sessions, 5, operation.id)
            async with sessions() as session:
                assert (await session.get(ExternalOperation, operation.id)).status == "ACCEPTED"
                assert (await session.get(Reservation, operation.id)).status == "CONSUMED"
        finally:
            await proxy.delete(f"/proxies/{name}")


async def test_revoked_approval_permission_prevents_external_operation(connected):
    ticket, event_id = await approve(connected)
    sessions = connected["app"].state.sessions
    async with sessions.begin() as session:
        grant = await session.get(
            StoreGrant,
            (
                connected["tenant"].bytes,
                connected["accounts"]["reviewer"].bytes,
                connected["store"].bytes,
                "approvals:decide",
            ),
            with_for_update=True,
        )
        await session.delete(grant)
    await prepare_event(sessions, event_id, 5)
    assert await operation_for(connected, ticket) is None
    response = await connected["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")
    assert response.json()["status"] == "NEEDS_REVISION"


async def test_broker_redelivers_unacknowledged_message(connected):
    ticket, event_id = await approve(connected)
    sessions = connected["app"].state.sessions
    queue_name = "verify-" + uuid4().hex
    url = Settings().broker_url.get_secret_value()
    async with Broker(url, queue_name, queue_name) as broker:
        assert await publish_one(sessions, broker, event_id)
        message = await broker.queue.get()
        assert message is not None
    async with Broker(url, queue_name, queue_name) as broker:
        try:
            assert await consume_one(sessions, broker, 5)
            assert await operation_for(connected, ticket) is not None
        finally:
            await broker.queue.delete(if_unused=False, if_empty=False)
            await broker.exchange.delete(if_unused=False)


async def test_platform_oms_credential_cannot_complete_exchange(connected):
    ticket, event_id = await approve(connected)
    sessions = connected["app"].state.sessions
    await prepare_event(sessions, event_id, 5)
    operation = await operation_for(connected, ticket)
    assert await execute_due(sessions, 5, operation.id)
    async with httpx.AsyncClient(
        base_url="http://127.0.0.1:8100",
        headers={"Authorization": "Bearer " + os.environ[connected["variable"]]},
    ) as client:
        response = await client.post(f"/exchanges/{UUID(bytes=operation.id)}/complete")
        assert response.status_code == 403