zhifu-platform/tests/test_merchant_integration.py

from uuid import UUID

import pytest
from test_after_sales import scenario, takeover_customer_conversation  # noqa: F401
from test_knowledge_memory import document

from zhifu.modules.knowledge.models import KnowledgeDocument

pytestmark = pytest.mark.integration


async def test_staff_proposal_requires_current_control(scenario):  # noqa: F811
    call = scenario["call"]
    path = "/openapi/v1/after-sales/proposals"
    response = await call("POST", path, "agent", scenario["proposal"])
    assert response.status_code == 403
    control = await takeover_customer_conversation(scenario)
    stale = {**scenario["proposal"], **control, "expected_control_version": 1}
    response = await call("POST", path, "agent", stale)
    assert response.status_code == 409
    assert response.json()["code"] == "CONTROL_CONFLICT"
    response = await call("POST", path, "agent", {**scenario["proposal"], **control})
    assert response.status_code == 201, response.text
    ticket = response.json()
    conversation_path = f"/openapi/v1/conversations/{control['conversation_id']}"
    response = await call(
        "POST", conversation_path + "/release", "agent",
        {"expected_control_version": control["expected_control_version"]},
    )
    assert response.status_code == 200, response.text
    response = await call(
        "POST", f"{path}/{ticket['proposal_id']}/revisions", "agent",
        {"expected_version": ticket["version"], "quantity": 1, "reason": "释放后编辑", **control},
    )
    assert response.status_code == 403
    response = await call("GET", f"/openapi/v1/tickets/{ticket['ticket_id']}")
    assert response.json()["version"] == ticket["version"]


async def test_knowledge_original_text_round_trip(scenario):  # noqa: F811
    path = await document(scenario)
    # 长正文包含重复词句与换行，验证编辑读取完整原文且不会引入检索重叠内容。
    original = ("键盘资料正文，保留编辑时的换行和标点。\n" * 60) + "末尾说明。"
    payload = {"expected_version": 1, "title": "完整正文验收", "text": original}
    response = await scenario["call"]("PUT", path, "agent", payload)
    assert response.status_code == 200, response.text
    assert len(response.json()["chunks"]) > 1
    assert response.json()["text"] == original
    response = await scenario["call"]("GET", path, "agent")
    assert response.json()["text"] == original
    response = await scenario["call"]("PUT", path, "agent", payload)
    assert response.json()["version"] == 2
    assert response.json()["text"] == original
    async with scenario["app"].state.sessions.begin() as session:
        row = await session.get(
            KnowledgeDocument, (scenario["tenant"].bytes, UUID(path.rsplit("/", 1)[1]).bytes)
        )
        row.source_text = None
    response = await scenario["call"]("GET", path, "agent")
    assert response.json()["text"] is None
    assert response.json()["chunks"]