zhifu-platform/tests/test_frontend_delivery_presentation.py

import asyncio
import json
import os
from uuid import UUID, uuid4

import pytest
from anyio import Path
from sqlalchemy import select
from test_after_sales import confirm_proposal, make_proposal, scenario  # noqa: F401
from test_platform_management import operator, receiver  # noqa: F401

from zhifu.modules.after_sales.models import BusinessEvent
from zhifu.modules.identity.auth import now
from zhifu.modules.merchant.models import ChannelApplication
from zhifu.modules.platform.delivery import enqueue, execute_one
from zhifu.modules.platform.models import ChannelDelivery

pytestmark = pytest.mark.integration


async def test_real_delivery_details_in_browser(scenario, operator, receiver):  # noqa: F811
    # 真实方案创建与本人确认分别产生业务事件，fixture 在浏览器检查后清理独立商家。
    proposal = await make_proposal(scenario)
    await confirm_proposal(scenario, proposal)
    sessions = scenario["app"].state.sessions
    receiver["accepted"].set()
    async with sessions.begin() as session:
        session.add(ChannelApplication(
            tenant_id=scenario["tenant"].bytes, store_id=scenario["store"].bytes,
            name="浏览器投递详情验收", callback_url=receiver["url"],
            credential_env=receiver["credential_env"], enabled=True, version=1,
            scopes=["tickets:results"], updated_at=now(),
        ))
    await enqueue(sessions, tenant_id=scenario["tenant"].bytes)
    async with sessions() as session:
        deliveries = (await session.scalars(select(ChannelDelivery).join(
            BusinessEvent, BusinessEvent.id == ChannelDelivery.event_id,
        ).where(
            ChannelDelivery.tenant_id == scenario["tenant"].bytes,
        ).order_by(BusinessEvent.aggregate_version))).all()
    assert len(deliveries) >= 2
    # 常驻投递进程可能先领取本轮事件；凭据只提供给当前接收器，必要时通过真实重试接口恢复。
    async with asyncio.timeout(20):
        for delivery in deliveries:
            for _ in range(3):
                await execute_one(sessions, delivery.id)
                async with sessions() as session:
                    result = await session.get(ChannelDelivery, delivery.id)
                if result.status == "SENT":
                    assert result.response_status == 204
                    break
                assert result.status == "FAILED", {
                    "status": result.status, "error": result.last_error,
                    "response_status": result.response_status,
                }
                response = await operator["call"](
                    "POST", f"/deliveries/{UUID(bytes=delivery.id)}/retry",
                    {"expected_version": result.version},
                )
                assert response.status_code == 200, response.text
            else:
                raise AssertionError("本轮真实投递未在三次受理内完成")
    snapshot = (await operator["call"]("GET", "/state")).json()
    ids = {str(UUID(bytes=item.id)) for item in deliveries}
    visible = [item for item in snapshot["deliveries"] if item["id"] in ids]
    assert len(visible) == len(ids)
    assert {item["delivery_id"] for item in receiver["received"]} == ids
    folder = Path(".work/ui-presentation")
    await folder.mkdir(parents=True, exist_ok=True)
    case_path = folder / f"delivery-case-{uuid4().hex}.json"
    await case_path.write_text(json.dumps({
        "username": operator["username"], "password": operator["password"],
        "deliveries": visible,
    }, ensure_ascii=False), encoding="utf-8")
    try:
        environment = {**os.environ, "DELIVERY_CASE_FILE": str(await case_path.resolve())}
        process = await asyncio.create_subprocess_exec(
            "node", "frontend/tests/platform-delivery-presentation.mjs",
            env=environment, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        try:
            async with asyncio.timeout(60):
                stdout, stderr = await process.communicate()
        finally:
            if process.returncode is None:
                process.terminate()
                await process.wait()
        assert process.returncode == 0, stderr.decode("utf-8", errors="replace")
        report = json.loads(stdout.decode("utf-8"))
        assert report["records"] == len(ids)
        print(json.dumps(report, ensure_ascii=False))
    finally:
        await case_path.unlink(missing_ok=True)