zhifu-platform/tests/test_merchant.py

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

import pytest
from langchain_core.messages import HumanMessage, message_to_dict
from sqlalchemy import func, select

from tests.test_after_sales import scenario  # noqa: F401
from zhifu.modules.agent_runtime.config import AgentSettings
from zhifu.modules.agent_runtime.models import AgentRun, ModelAttempt
from zhifu.modules.agent_runtime.service import queue_run
from zhifu.modules.conversations.api import Message, accept_message
from zhifu.modules.conversations.models import Conversation
from zhifu.modules.identity.access_models import Principal, StoreGrant
from zhifu.modules.identity.auth import authenticate_hash
from zhifu.modules.merchant.models import StaffInvitation

pytestmark = pytest.mark.integration


async def grant_management(s):
    async with s["app"].state.sessions.begin() as session:
        for permission in ["stores:manage", "staff:manage", "channels:manage", "agent:publish"]:
            session.add(StoreGrant(tenant_id=s["tenant"].bytes, store_id=s["store"].bytes,
                                   principal_id=s["accounts"]["agent"].bytes,
                                   permission=permission))


async def test_staff_store_permissions_versions_and_sessions(scenario):  # noqa: F811
    s = scenario
    await grant_management(s)
    call = s["call"]
    base = f"/admin/v1/stores/{s['store']}"
    assert (await call("GET", base, "customer")).status_code == 403
    assert (await call("GET", base, "reviewer")).status_code == 404
    staff = await call("GET", base + "/staff", "agent")
    assert staff.status_code == 200, staff.text
    row = next(item for item in staff.json()["items"] if item["account"] == "reviewer")
    path = base + f"/staff/{row['id']}/status"
    change = {"expected_version": row["version"], "enabled": False}
    key = str(uuid4())
    disabled = await call("PUT", path, "agent", change, key)
    assert disabled.status_code == 200, disabled.text
    assert (await call("PUT", path, "agent", change, key)).json() == disabled.json()
    assert (await call("GET", "/openapi/v1/auth/me", "reviewer")).status_code == 401
    assert (await call("PUT", path, "agent", change)).status_code == 409
    assert (await call("PUT", path, "agent", {
        "expected_version": row["version"] + 1, "enabled": True
    })).status_code == 200
    # 启用后原会话保持撤销，员工需要重新验证身份。
    assert (await call("GET", "/openapi/v1/auth/me", "reviewer")).status_code == 401
    store = (await call("GET", base, "agent")).json()
    assert (await call("PUT", base + "/status", "agent", {
        "expected_version": store["version"], "enabled": False
    })).status_code == 200
    assert (await call("GET", base + "/exchange-rules", "agent")).status_code == 404
    assert (await call("GET", base, "agent")).status_code == 200
    assert (await call("PUT", base + "/status", "agent", {
        "expected_version": store["version"] + 1, "enabled": True
    })).status_code == 200


async def test_invitation_persistence_and_no_implicit_login(scenario):  # noqa: F811
    s = scenario
    await grant_management(s)
    base = f"/admin/v1/stores/{s['store']}"
    body = {"account": "invited.staff", "role": "staff"}
    key = str(uuid4())
    response = await s["call"]("POST", base + "/staff/invitations", "agent", body, key)
    assert response.status_code == 201, response.text
    assert response.json()["invited"] and not response.json()["enabled"]
    repeated = await s["call"]("POST", base + "/staff/invitations", "agent", body, key)
    assert repeated.json() == response.json()
    assert (await s["call"]("POST", base + "/staff/invitations", "agent", body)).status_code == 409
    async with s["app"].state.sessions.begin() as session:
        invitation = await session.scalar(select(StaffInvitation).where(
            StaffInvitation.tenant_id == s["tenant"].bytes,
            StaffInvitation.account == body["account"]))
        assert invitation.status == "PENDING"
        assert await session.scalar(select(Principal).where(
            Principal.tenant_id == s["tenant"].bytes,
            Principal.username == body["account"])) is None


async def test_rule_draft_publish_and_stale_action(scenario):  # noqa: F811
    s = scenario
    base = f"/admin/v1/stores/{s['store']}/exchange-rules"
    call = s["call"]
    state = (await call("GET", base, "agent")).json()
    assert state["current"]["version"] == 1 and state["draft"] is None
    body = {"expected_version": 0, "base_version": 1, "window_days": 14,
            "product_types": ["keyboard"], "citation": "签收十四天内功能故障可申请换货。"}
    saved = await call("PUT", base + "/draft", "agent", body)
    assert saved.status_code == 200, saved.text
    assert (await call("GET", base, "agent")).json()["draft"]["window_days"] == 14
    assert (await call("PUT", base + "/draft", "agent", body)).status_code == 409
    publish = {"expected_version": saved.json()["version"], "base_version": 1}
    key = str(uuid4())
    response = await call("POST", base + "/draft/publish", "agent", publish, key)
    assert response.status_code == 200, response.text
    assert response.json()["version"] == 2
    repeated = await call("POST", base + "/draft/publish", "agent", publish, key)
    assert repeated.json() == response.json()
    public = await call("GET", f"/openapi/v1/stores/{s['store']}/exchange-rules/current")
    assert public.json()["window_days"] == 14
    saved = await call("PUT", base + "/draft", "agent", {**body, "base_version": 2})
    assert saved.status_code == 200
    assert (await call("POST", base + "/draft/publish", "agent", publish)).status_code == 409


async def test_channel_validation_and_release_requirements(scenario):  # noqa: F811
    s = scenario
    await grant_management(s)
    base = f"/admin/v1/stores/{s['store']}"
    body = {"expected_version": 0, "name": "渠道配置验证", "scope_confirmed": True,
            "callback_url": "http://127.0.0.1/events"}
    assert (await s["call"]("PUT", base + "/channels", "agent", body)).status_code == 422
    assert (await s["call"]("PUT", base + "/channels", "agent", {
        **body, "callback_url": "https://127.0.0.1/events"
    })).status_code == 422
    channels = (await s["call"]("GET", base + "/channels", "agent")).json()
    assert channels["channel"] is None
    assert channels["store"] == (await s["call"]("GET", base, "agent")).json()
    candidate = await s["call"]("GET", base + "/agent-releases", "agent")
    assert candidate.status_code == 200, candidate.text
    assert candidate.json()["current"] is None and candidate.json()["errors"]
    release = await s["call"]("POST", base + "/agent-releases", "agent", {
        "expected_version": 0, "candidate_hash": candidate.json()["candidate_hash"],
        "regression_checked": True, "reason": "核对配置完整性"
    })
    assert release.status_code == 409 and release.json()["code"] == "RELEASE_NOT_READY"


async def persist_cancelled_snapshot(s, conversation_id=None, key=None):
    # 队列记录与取消状态在同一事务提交，工作进程无法看到可执行任务。
    token = s["headers"]["customer"]["Authorization"].removeprefix("Bearer ")
    async with s["app"].state.sessions.begin() as session:
        auth = await authenticate_hash(
            session, hashlib.sha256(token.encode()).digest(), str(uuid4())
        )
        conversation = await session.get(
            Conversation, UUID(conversation_id).bytes, with_for_update=True
        ) if conversation_id else None
        request_key = key or str(uuid4())
        if conversation:
            event = await accept_message(
                session, auth, conversation,
                Message(text="查询当前店铺规则",
                        expected_control_version=conversation.control_version),
                request_key,
            )
            run = await session.get(AgentRun, UUID(event["payload"]["run_id"]).bytes)
            conversation.active_run_id = None
        else:
            run = await queue_run(
                session, auth, s["store"].bytes, request_key,
                hashlib.sha256(request_key.encode()).digest(),
                [message_to_dict(HumanMessage(content="查询当前店铺规则"))],
            )
        run.status = "CANCELLED"
        return run.id


async def test_agent_release_success_and_execution_snapshots(scenario):  # noqa: F811
    s = scenario
    settings = AgentSettings()
    if not os.getenv(settings.credential_env):
        pytest.skip("发布成功验收需要配置真实模型凭据")
    await grant_management(s)
    async with s["app"].state.sessions.begin() as session:
        session.add(StoreGrant(
            tenant_id=s["tenant"].bytes, store_id=s["store"].bytes,
            principal_id=s["accounts"]["agent"].bytes, permission="knowledge:write",
        ))
    call = s["call"]
    base = f"/admin/v1/stores/{s['store']}"
    document_id = uuid4()
    document_path = base + f"/knowledge/{document_id}"
    release_path = base + "/agent-releases"

    async def create_conversation():
        response = await call(
            "POST", "/openapi/v1/conversations",
            payload={"store_id": str(s["store"])},
        )
        assert response.status_code == 201, response.text
        return response.json()["conversation_id"]

    try:
        draft = await call("PUT", document_path, "agent", {
            "expected_version": 0, "title": "键盘售后服务说明",
            "text": "键盘出现功能故障时，请查询订单和当前换货规则，由客户确认后交主管审批。",
        })
        assert draft.status_code == 200, draft.text
        published = await call("POST", document_path + "/publish", "agent", {
            "expected_version": draft.json()["version"],
        })
        assert published.status_code == 200, published.text
        assert published.json()["status"] == "PUBLISHED"
        first_candidate = (await call("GET", release_path, "agent")).json()
        assert first_candidate["errors"] == []
        body = {
            "expected_version": 0, "candidate_hash": first_candidate["candidate_hash"],
            "regression_checked": True, "reason": "核对发布条件、规则与知识版本",
        }
        key = str(uuid4())
        first_response = await call("POST", release_path, "agent", body, key)
        assert first_response.status_code == 201, first_response.text
        first = first_response.json()
        assert first["version"] == 1
        assert first["manifest"]["knowledge"][0]["document_id"] == str(document_id)
        assert first["manifest"]["knowledge"][0]["version"] == published.json()["version"]
        repeated = await call("POST", release_path, "agent", body, key)
        assert repeated.status_code == 201 and repeated.json() == first
        conflicting = await call(
            "POST", release_path, "agent", {**body, "reason": "其他发布说明"}, key
        )
        assert conflicting.status_code == 409
        assert conflicting.json()["code"] == "IDEMPOTENCY_CONFLICT"
        stale = await call("POST", release_path, "agent", body)
        assert stale.status_code == 409 and stale.json()["code"] == "VERSION_CONFLICT"
        old_conversation = await create_conversation()
        async with s["app"].state.sessions.begin() as session:
            old = await session.get(Conversation, UUID(old_conversation).bytes)
            assert old.agent_release_version is None
            # 保留升级前会话可能存在的历史列值，验证它不再参与后续运行的版本选择。
            old.agent_release_version = 1
        first_message_key = str(uuid4())
        first_run_id = await persist_cancelled_snapshot(
            s, old_conversation, first_message_key
        )

        revised = await call("PUT", document_path, "agent", {
            "expected_version": published.json()["version"], "title": "键盘售后服务说明",
            "text": "键盘功能故障需要核对订单及有效换货规则；运输中请保护键盘并保留包装。",
        })
        assert revised.status_code == 200, revised.text
        republished = await call("POST", document_path + "/publish", "agent", {
            "expected_version": revised.json()["version"],
        })
        assert republished.status_code == 200, republished.text
        second_candidate = (await call("GET", release_path, "agent")).json()
        assert second_candidate["errors"] == []
        assert second_candidate["candidate_hash"] != first_candidate["candidate_hash"]
        changed = await call("POST", release_path, "agent", {**body, "expected_version": 1})
        assert changed.status_code == 409 and changed.json()["code"] == "VERSION_CONFLICT"
        second_response = await call("POST", release_path, "agent", {
            **body, "expected_version": 1,
            "candidate_hash": second_candidate["candidate_hash"],
        })
        assert second_response.status_code == 201, second_response.text
        second = second_response.json()
        assert second["version"] == 2
        assert second["manifest"]["knowledge"][0]["version"] == republished.json()["version"]
        # 同一消息重试继续返回原运行，不因中途发布而重复创建运行或替换快照。
        assert await persist_cancelled_snapshot(
            s, old_conversation, first_message_key
        ) == first_run_id
        new_conversation = await create_conversation()
        old_run_id = await persist_cancelled_snapshot(s, old_conversation)
        new_run_id = await persist_cancelled_snapshot(s, new_conversation)
        independent_run_id = await persist_cancelled_snapshot(s)
        async with s["app"].state.sessions.begin() as session:
            old = await session.get(Conversation, UUID(old_conversation).bytes)
            new = await session.get(Conversation, UUID(new_conversation).bytes)
            assert old.agent_release_version == 1 and new.agent_release_version is None
            for run_id, expected in [
                (first_run_id, first), (old_run_id, second),
                (new_run_id, second), (independent_run_id, second),
            ]:
                stored = await session.get(AgentRun, run_id)
                assert stored.release_snapshot == expected["manifest"]
                assert stored.graph_version == expected["manifest"]["graph_version"]
                assert stored.model_name == settings.model_name
                assert stored.model_url == settings.model_url
                assert stored.model_credential_env == settings.credential_env
                assert stored.status == "CANCELLED" and stored.model_calls == 0
            continued = await session.get(AgentRun, old_run_id)
            assert continued.conversation_id == old.id
            assert [item["data"]["content"] for item in continued.messages] == [
                "查询当前店铺规则", "查询当前店铺规则",
            ]
            assert old.sequence == 2 and old.active_run_id is None
            attempts = await session.scalar(select(func.count()).select_from(ModelAttempt).where(
                ModelAttempt.tenant_id == s["tenant"].bytes,
            ))
            assert attempts == 0
        current = (await call("GET", release_path, "agent")).json()["current"]
        assert current == second
    finally:
        # 只清理本次随机租户与资料标识对应的向量，MySQL 由 scenario 清理。
        knowledge = s["app"].state.knowledge
        await asyncio.to_thread(
            knowledge.milvus.delete, knowledge.settings.collection,
            filter=f'tenant == "{s["tenant"].hex}" and document == "{document_id.hex}"',
            timeout=knowledge.settings.timeout,
        )