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, )