zhifu-platform/tests/test_after_sales.py

import asyncio
from argparse import Namespace
from datetime import UTC, datetime, timedelta
from ipaddress import IPv6Address
from uuid import uuid4

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import delete, func, select

from zhifu.app import create_app
from zhifu.manage import run as manage
from zhifu.modules.after_sales.models import (
    Approval,
    BusinessEvent,
    CommandReceipt,
    Confirmation,
    ExchangeRule,
    OrderProjection,
    ProposalVersion,
    Ticket,
)
from zhifu.modules.agent_runtime.models import AgentRun, ModelAttempt, RunInputEvent
from zhifu.modules.channels.models import ChannelInbox, ChannelKey, ChannelNonce
from zhifu.modules.conversations.models import Conversation, ConversationEvent
from zhifu.modules.identity.access_models import AccessSession, Principal, StoreGrant
from zhifu.modules.identity.auth import password_hasher
from zhifu.modules.identity.models import Store, Tenant
from zhifu.modules.integrations.models import (
    Connector,
    ConsumerReceipt,
    ExternalOperation,
    Reservation,
)
from zhifu.modules.knowledge.models import KnowledgeDocument
from zhifu.modules.memory.models import MemoryJob, MemoryProfile
from zhifu.modules.memory.service import erase, snapshot
from zhifu.modules.merchant.models import (
    AgentRelease,
    ChannelApplication,
    RuleDraft,
    StaffInvitation,
)
from zhifu.modules.notifications.models import Notification
from zhifu.modules.platform.models import ChannelDelivery
from zhifu.modules.storefront.models import (
    CatalogCategory,
    CatalogProduct,
    ChannelIdentity,
    IdentityChallenge,
)

pytestmark = pytest.mark.integration


@pytest_asyncio.fixture
async def scenario():
    app = create_app()
    tenant_ids = [uuid4(), uuid4()]
    stores = [uuid4(), uuid4()]
    accounts = {
        name: uuid4() for name in ("customer", "other", "agent", "reviewer", "connector", "foreign")
    }
    password = "integration-" + str(uuid4())
    encoded = password_hasher.hash(password)
    permissions = {
        "customer": [
            "orders:read",
            "tickets:read",
            "after_sales:create",
            "after_sales:confirm",
            "after_sales:cancel",
        ],
        "other": ["orders:read", "tickets:read", "after_sales:confirm"],
        "agent": [
            "orders:read",
            "tickets:read",
            "after_sales:create",
            "after_sales:cancel",
            "approvals:decide",
            "rules:write",
        ],
        "reviewer": ["tickets:read", "approvals:decide"],
        "connector": ["orders:sync"],
        "foreign": ["tickets:read", "orders:read", "after_sales:create"],
    }
    async with app.router.lifespan_context(app):
        async with app.state.sessions.begin() as session:
            for tenant_id, store_id in zip(tenant_ids, stores, strict=True):
                session.add(Tenant(id=tenant_id.bytes, name="集成验收"))
                await session.flush()
                session.add(Store(id=store_id.bytes, tenant_id=tenant_id.bytes, name="集成店铺"))
            await session.flush()
            for name, actor_id in accounts.items():
                index = 1 if name == "foreign" else 0
                kind = "SERVICE" if name == "connector" else "STAFF"
                if name in {"customer", "other", "foreign"}:
                    kind = "CUSTOMER"
                session.add(
                    Principal(
                        id=actor_id.bytes,
                        tenant_id=tenant_ids[index].bytes,
                        username=name,
                        password_hash=encoded,
                        kind=kind,
                    )
                )
                await session.flush()
                for permission in permissions[name]:
                    session.add(
                        StoreGrant(
                            tenant_id=tenant_ids[index].bytes,
                            principal_id=actor_id.bytes,
                            store_id=stores[index].bytes,
                            permission=permission,
                        )
                    )
        try:
            transport = ASGITransport(app=app, client=(str(IPv6Address(tenant_ids[0].int)), 12345))
            async with AsyncClient(transport=transport, base_url="http://test") as client:
                headers = {}
                for name in accounts:
                    index = 1 if name == "foreign" else 0
                    response = await client.post(
                        "/openapi/v1/auth/sessions",
                        json={
                            "tenant_id": str(tenant_ids[index]),
                            "username": name,
                            "password": password,
                        },
                    )
                    assert response.status_code == 200, response.text
                    headers[name] = {"Authorization": "Bearer " + response.json()["access_token"]}
                order_id = uuid4()

                async def call(method, path, actor="customer", payload=None, key=None):
                    return await client.request(
                        method,
                        path,
                        headers={**headers[actor], "Idempotency-Key": key or str(uuid4())},
                        json=payload,
                    )

                source = {
                    "store_id": str(stores[0]),
                    "subject_id": str(accounts["customer"]),
                    "source_system": "oms",
                    "source_version": 1,
                    "item_id": "keyboard-1",
                    "product_type": "keyboard",
                    "available_quantity": 1,
                    "signed_at": datetime.now(UTC).isoformat(),
                }
                response = await call("PUT", f"/internal/v1/orders/{order_id}", "connector", source)
                assert response.status_code == 200, response.text
                response = await call(
                    "POST",
                    f"/admin/v1/stores/{stores[0]}/exchange-rules",
                    "agent",
                    {
                        "expected_version": 0,
                        "window_days": 7,
                        "product_types": ["keyboard"],
                        "citation": "商家换货规则第 1 条：签收后七天内受理功能故障换货。",
                    },
                )
                assert response.status_code == 201, response.text
                yield {
                    "app": app,
                    "call": call,
                    "client": client,
                    "headers": headers,
                    "tenant": tenant_ids[0],
                    "store": stores[0],
                    "accounts": accounts,
                    "order_id": order_id,
                    "source": source,
                    "proposal": {
                        "order_id": str(order_id),
                        "item_id": "keyboard-1",
                        "quantity": 1,
                        "reason": "按键失灵",
                    },
                }
        finally:
            async with app.state.sessions() as session:
                profiles = (
                    await session.scalars(
                        select(MemoryProfile).where(
                            MemoryProfile.tenant_id.in_([value.bytes for value in tenant_ids])
                        )
                    )
                ).all()
            for profile in profiles:
                await erase(profile.tenant_id, profile.subject_id, snapshot(profile))
            async with app.state.sessions.begin() as session:
                for model in (
                    ChannelDelivery,
                    ChannelInbox,
                    ChannelNonce,
                    ChannelKey,
                    ChannelIdentity,
                    IdentityChallenge,
                    CatalogProduct,
                    CatalogCategory,
                    AgentRelease,
                    RuleDraft,
                    StaffInvitation,
                    ChannelApplication,
                    Notification,
                    MemoryJob,
                    MemoryProfile,
                    KnowledgeDocument,
                    ModelAttempt,
                    RunInputEvent,
                    AgentRun,
                    ConversationEvent,
                    Conversation,
                    ConsumerReceipt,
                    Reservation,
                    ExternalOperation,
                    Connector,
                    BusinessEvent,
                    CommandReceipt,
                    Approval,
                    Confirmation,
                    ProposalVersion,
                    Ticket,
                    ExchangeRule,
                    OrderProjection,
                    StoreGrant,
                    AccessSession,
                    Principal,
                    Store,
                ):
                    await session.execute(
                        delete(model).where(
                            model.tenant_id.in_([value.bytes for value in tenant_ids])
                        )
                    )
                await session.execute(
                    delete(Tenant).where(Tenant.id.in_([value.bytes for value in tenant_ids]))
                )


async def takeover_customer_conversation(scenario):
    # 通过真实授权与会话接口建立客服接管条件，调用方明确选择此业务准备步骤。
    async with scenario["app"].state.sessions.begin() as session:
        for permission in ("conversations:read", "conversations:manage", "conversations:reply"):
            identity = (
                scenario["tenant"].bytes, scenario["accounts"]["agent"].bytes,
                scenario["store"].bytes, permission,
            )
            if await session.get(StoreGrant, identity) is None:
                session.add(StoreGrant(
                    tenant_id=identity[0], principal_id=identity[1],
                    store_id=identity[2], permission=permission,
                ))
    response = await scenario["call"](
        "POST", "/openapi/v1/conversations", payload={"store_id": str(scenario["store"])}
    )
    assert response.status_code == 201, response.text
    conversation_id = response.json()["conversation_id"]
    response = await scenario["call"](
        "POST", f"/openapi/v1/conversations/{conversation_id}/takeover", "agent",
        {"expected_control_version": 1},
    )
    assert response.status_code == 200, response.text
    return {
        "conversation_id": conversation_id,
        "expected_control_version": response.json()["payload"]["control_version"],
    }


async def make_proposal(scenario, actor="customer", control=None):
    response = await scenario["call"](
        "POST", "/openapi/v1/after-sales/proposals", actor,
        {**scenario["proposal"], **(control or {})},
    )
    assert response.status_code == 201, response.text
    return response.json()


async def confirm_proposal(scenario, proposal):
    response = await scenario["call"](
        "POST",
        f"/openapi/v1/after-sales/proposals/{proposal['proposal_id']}/confirm",
        payload={
            "proposal_version": proposal["proposal_version"],
            "expected_ticket_version": proposal["version"],
        },
    )
    assert response.status_code == 200, response.text
    return response.json()


async def test_confirmation_approval_and_isolation(scenario):
    call = scenario["call"]
    control = await takeover_customer_conversation(scenario)
    proposal = await make_proposal(scenario, "agent", control)
    path = f"/openapi/v1/after-sales/proposals/{proposal['proposal_id']}/confirm"
    confirm = {"proposal_version": 1, "expected_ticket_version": 1}
    assert (await call("POST", path, "agent", confirm)).status_code == 403
    assert (await call("POST", path, "other", confirm)).status_code == 404
    assert (
        await call("GET", f"/openapi/v1/tickets/{proposal['ticket_id']}", "foreign")
    ).status_code == 404
    proposal = await confirm_proposal(scenario, proposal)
    path = f"/openapi/v1/approvals/{proposal['approval_id']}/decisions"
    decision = {
        "expected_version": 2,
        "proposal_version": 1,
        "decision": "APPROVED",
        "reason": "审核通过",
    }
    assert (await call("POST", path, "agent", decision)).status_code == 403
    response = await call("POST", path, "reviewer", decision)
    assert response.status_code == 200, response.text
    assert response.json()["status"] == "APPROVED"
    async with scenario["app"].state.sessions() as session:
        count = await session.scalar(
            select(func.count())
            .select_from(BusinessEvent)
            .where(
                BusinessEvent.tenant_id == scenario["tenant"].bytes,
                BusinessEvent.event_type == "after_sale.approved",
            )
        )
        assert count == 1


async def test_concurrent_idempotency_and_conflict(scenario):
    call = scenario["call"]
    key = str(uuid4())
    responses = await asyncio.gather(
        *[
            call("POST", "/openapi/v1/after-sales/proposals", payload=scenario["proposal"], key=key)
            for _ in range(4)
        ]
    )
    assert all(response.status_code == 201 for response in responses)
    assert len({r.json()["ticket_id"] for r in responses}) == 1
    response = await call(
        "POST",
        "/openapi/v1/after-sales/proposals",
        payload={**scenario["proposal"], "reason": "新的故障描述"},
        key=key,
    )
    assert response.status_code == 409
    assert response.json()["code"] == "IDEMPOTENCY_CONFLICT"
    ticket = responses[0].json()
    confirm_key = str(uuid4())
    replies = await asyncio.gather(
        *[
            call(
                "POST",
                f"/openapi/v1/after-sales/proposals/{ticket['proposal_id']}/confirm",
                payload={"proposal_version": 1, "expected_ticket_version": 1},
                key=confirm_key,
            )
            for _ in range(3)
        ]
    )
    assert all(r.status_code == 200 for r in replies)
    assert len({r.json()["approval_id"] for r in replies}) == 1


async def test_revision_invalidates_previous_approval(scenario):
    call = scenario["call"]
    proposal = await confirm_proposal(scenario, await make_proposal(scenario))
    response = await call(
        "POST",
        f"/openapi/v1/after-sales/proposals/{proposal['proposal_id']}/revisions",
        payload={"expected_version": 2, "quantity": 1, "reason": "补充故障情况"},
    )
    assert response.status_code == 200, response.text
    revised = response.json()
    assert revised["proposal_version"] == 2
    response = await call(
        "POST",
        f"/openapi/v1/approvals/{proposal['approval_id']}/decisions",
        "reviewer",
        {"expected_version": 3, "proposal_version": 1, "decision": "APPROVED", "reason": "审核"},
    )
    assert response.status_code == 409
    assert response.json()["code"] == "PROPOSAL_VERSION_CONFLICT"
    current = await confirm_proposal(scenario, revised)
    assert current["approval_id"] != proposal["approval_id"]


async def test_permission_revocation_and_logout(scenario):
    proposal = await make_proposal(scenario)
    async with scenario["app"].state.sessions.begin() as session:
        actor = await session.get(
            Principal, scenario["accounts"]["customer"].bytes, with_for_update=True
        )
        await session.execute(
            delete(StoreGrant).where(
                StoreGrant.principal_id == actor.id, StoreGrant.permission == "tickets:read"
            )
        )
    response = await scenario["call"]("GET", f"/openapi/v1/tickets/{proposal['ticket_id']}")
    assert response.status_code == 404
    assert (
        await scenario["call"]("DELETE", "/openapi/v1/auth/sessions/current")
    ).status_code == 204
    assert (await scenario["call"]("GET", "/openapi/v1/auth/me")).status_code == 401


async def test_rule_revocation(scenario):
    proposal = await make_proposal(scenario)
    response = await scenario["call"](
        "POST",
        f"/admin/v1/stores/{scenario['store']}/exchange-rules/revoke",
        "agent",
        {"expected_version": 1, "reason": "规则撤销"},
    )
    assert response.status_code == 200
    response = await scenario["call"](
        "POST",
        f"/openapi/v1/after-sales/proposals/{proposal['proposal_id']}/confirm",
        payload={"expected_ticket_version": 1, "proposal_version": 1},
    )
    assert response.status_code == 422
    assert response.json()["code"] == "RULE_NOT_SATISFIED"


async def test_tenant_injection_and_cancellation(scenario):
    response = await scenario["call"](
        "POST",
        "/openapi/v1/after-sales/proposals",
        payload={**scenario["proposal"], "tenant_id": str(uuid4())},
    )
    assert response.status_code == 422
    proposal = await make_proposal(scenario)
    response = await scenario["call"](
        "POST",
        f"/openapi/v1/tickets/{proposal['ticket_id']}/cancel",
        payload={"expected_version": 1, "reason": "客户取消"},
    )
    assert response.status_code == 200
    assert response.json()["status"] == "CANCELLED"
    response = await scenario["call"](
        "POST",
        f"/openapi/v1/after-sales/proposals/{proposal['proposal_id']}/confirm",
        payload={"proposal_version": 1, "expected_ticket_version": 2},
    )
    assert response.status_code == 409


async def test_order_version_requires_reconfirmation(scenario):
    proposal = await make_proposal(scenario)
    response = await scenario["call"](
        "PUT",
        f"/internal/v1/orders/{scenario['order_id']}",
        "connector",
        {**scenario["source"], "source_version": 2},
    )
    assert response.status_code == 200
    response = await scenario["call"](
        "POST",
        f"/openapi/v1/after-sales/proposals/{proposal['proposal_id']}/confirm",
        payload={"proposal_version": 1, "expected_ticket_version": 1},
    )
    assert response.status_code == 409
    assert response.json()["code"] == "ORDER_VERSION_CONFLICT"


async def test_expired_token_and_login_throttling(scenario):
    async with scenario["app"].state.sessions.begin() as session:
        access = await session.scalar(
            select(AccessSession).where(
                AccessSession.principal_id == scenario["accounts"]["customer"].bytes
            )
        )
        access.expires_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(seconds=1)
    assert (await scenario["call"]("GET", "/openapi/v1/auth/me")).status_code == 401
    invalid_password = "invalid-password-" + str(uuid4())
    for _ in range(10):
        response = await scenario["client"].post(
            "/openapi/v1/auth/sessions",
            json={
                "tenant_id": str(scenario["tenant"]),
                "username": "missing",
                "password": invalid_password,
            },
        )
        assert response.status_code == 401
        assert invalid_password not in response.text
    response = await scenario["client"].post(
        "/openapi/v1/auth/sessions",
        json={
            "tenant_id": str(scenario["tenant"]),
            "username": "missing",
            "password": invalid_password,
        },
    )
    assert response.status_code == 429
    assert response.headers["Retry-After"] == "60"


async def test_approval_cancellation_race(scenario):
    ticket = await confirm_proposal(scenario, await make_proposal(scenario))
    responses = await asyncio.gather(
        scenario["call"](
            "POST",
            f"/openapi/v1/approvals/{ticket['approval_id']}/decisions",
            "reviewer",
            {
                "expected_version": 2,
                "proposal_version": 1,
                "decision": "APPROVED",
                "reason": "审核",
            },
        ),
        scenario["call"](
            "POST",
            f"/openapi/v1/tickets/{ticket['ticket_id']}/cancel",
            payload={"expected_version": 2, "reason": "取消"},
        ),
    )
    assert sorted(response.status_code for response in responses) == [200, 409]


async def test_revision_submitter_cannot_approve(scenario):
    ticket = await make_proposal(scenario)
    control = await takeover_customer_conversation(scenario)
    response = await scenario["call"](
        "POST",
        f"/openapi/v1/after-sales/proposals/{ticket['proposal_id']}/revisions",
        "agent",
        {"expected_version": 1, "quantity": 1, "reason": "客服补充申请", **control},
    )
    assert response.status_code == 200
    ticket = await confirm_proposal(scenario, response.json())
    response = await scenario["call"](
        "POST",
        f"/openapi/v1/approvals/{ticket['approval_id']}/decisions",
        "agent",
        {
            "expected_version": ticket["version"],
            "proposal_version": ticket["proposal_version"],
            "decision": "APPROVED",
            "reason": "审批",
        },
    )
    assert response.status_code == 403


async def test_store_and_list_isolation(scenario):
    ticket = await confirm_proposal(scenario, await make_proposal(scenario))
    path = f"/openapi/v1/orders?store_id={scenario['store']}"
    assert len((await scenario["call"]("GET", path)).json()["items"]) == 1
    assert (await scenario["call"]("GET", path, "other")).json()["items"] == []
    assert (await scenario["call"]("GET", path, "foreign")).status_code == 404
    response = await scenario["call"](
        "GET", f"/openapi/v1/approvals?store_id={scenario['store']}", "reviewer"
    )
    assert response.json()["items"][0]["approval_id"] == ticket["approval_id"]
    async with scenario["app"].state.sessions.begin() as session:
        store_id = uuid4()
        session.add(Store(id=store_id.bytes, tenant_id=scenario["tenant"].bytes, name="未授权店铺"))
    response = await scenario["call"]("GET", f"/openapi/v1/orders?store_id={store_id}", "agent")
    assert response.status_code == 404


async def test_order_sync_requires_service_and_immutable_version(scenario):
    path = f"/internal/v1/orders/{scenario['order_id']}"
    assert (await scenario["call"]("PUT", path, "customer", scenario["source"])).status_code == 403
    assert (await scenario["call"]("PUT", path, "connector", scenario["source"])).status_code == 200
    response = await scenario["call"](
        "PUT", path, "connector", {**scenario["source"], "available_quantity": 0}
    )
    assert response.status_code == 409


async def test_management_revocation_is_audited(scenario):
    ticket = await make_proposal(scenario)
    await manage(
        Namespace(
            action="revoke",
            tenant=scenario["tenant"],
            actor=scenario["accounts"]["agent"],
            store=scenario["store"],
            permission="tickets:read",
        )
    )
    response = await scenario["call"]("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}", "agent")
    assert response.status_code == 404
    async with scenario["app"].state.sessions() as session:
        event = await session.scalar(
            select(BusinessEvent).where(
                BusinessEvent.tenant_id == scenario["tenant"].bytes,
                BusinessEvent.event_type == "identity.revoke",
            )
        )
        assert event.payload["permission"] == "tickets:read"