zhifu-platform/tests/test_notifications.py

import pytest
from sqlalchemy import delete
from test_after_sales import scenario  # noqa: F401
from test_conversations import setup

from zhifu.modules.identity.access_models import StoreGrant

pytestmark = pytest.mark.integration


async def inbox(context, actor="agent"):
    response = await context["call"](
        "GET", f"/admin/v1/notifications?store_id={context['store']}&limit=100", actor
    )
    assert response.status_code == 200, response.text
    return response.json()


async def test_customer_message_recipient_and_read_snapshot(scenario):  # noqa: F811
    path = await setup(scenario)
    call = scenario["call"]
    takeover = await call("POST", path + "/takeover", "agent", {"expected_control_version": 1})
    assert takeover.status_code == 200, takeover.text
    message = {"text": "请人工核实商品信息", "expected_control_version": 2}
    first = await call("POST", path + "/messages", payload=message, key="notification-message")
    repeated = await call("POST", path + "/messages", payload=message, key="notification-message")
    assert first.status_code == 201 and first.json() == repeated.json()
    initial = await inbox(scenario)
    assert initial["unread_count"] == 1
    item = initial["items"][0]
    assert item["target_id"] == path.rsplit("/", 1)[1]
    assert message["text"] not in item["summary"]
    assert (await inbox(scenario, "reviewer"))["items"] == []
    unauthorized = await call("GET", f"/admin/v1/notifications/{item['id']}", "reviewer")
    assert unauthorized.status_code == 404
    second = await call("POST", path + "/messages", payload={**message, "text": "补充说明"})
    assert second.status_code == 201
    read = await call("POST", "/admin/v1/notifications/read-all", "agent", {
        "store_id": str(scenario["store"]), "through_id": initial["snapshot_id"],
    })
    assert read.json() == {"updated": 1}
    again = await call("POST", "/admin/v1/notifications/read", "agent", {"ids": [item["id"]]})
    assert again.json() == {"updated": 0}
    latest = await inbox(scenario)
    assert latest["unread_count"] == 1 and len(latest["items"]) == 2
    unread = next(row for row in latest["items"] if not row["read"])
    detail = await call("GET", f"/admin/v1/notifications/{unread['id']}", "agent")
    assert detail.status_code == 200 and detail.json()["read"]
    assert (await inbox(scenario))["unread_count"] == 0
    customer = await call("GET", f"/admin/v1/notifications?store_id={scenario['store']}")
    assert customer.status_code == 403


async def test_notifications_follow_current_permission_and_tenant(scenario):  # noqa: F811
    path = await setup(scenario)
    call = scenario["call"]
    assert (await call(
        "POST", path + "/takeover", "agent", {"expected_control_version": 1}
    )).status_code == 200
    assert (await call(
        "POST", path + "/messages", payload={"text": "请查看", "expected_control_version": 2}
    )).status_code == 201
    item = (await inbox(scenario))["items"][0]
    foreign = await call("GET", f"/admin/v1/notifications/{item['id']}", "foreign")
    assert foreign.status_code == 403
    async with scenario["app"].state.sessions.begin() as session:
        await session.execute(delete(StoreGrant).where(
            StoreGrant.tenant_id == scenario["tenant"].bytes,
            StoreGrant.principal_id == scenario["accounts"]["agent"].bytes,
            StoreGrant.permission == "conversations:read",
        ))
    assert (await inbox(scenario))["items"] == []
    assert (await call("GET", f"/admin/v1/notifications/{item['id']}", "agent")).status_code == 404


async def test_confirmed_proposal_creates_persistent_approval_notice(scenario):  # noqa: F811
    call = scenario["call"]
    created = await call("POST", "/openapi/v1/after-sales/proposals", payload=scenario["proposal"])
    assert created.status_code == 201, created.text
    ticket = created.json()
    response = await call(
        "POST", f"/openapi/v1/after-sales/proposals/{ticket['proposal_id']}/confirm",
        payload={"proposal_version": ticket["proposal_version"],
                 "expected_ticket_version": ticket["version"]}, key="confirm-notice",
    )
    assert response.status_code == 200, response.text
    notices = (await inbox(scenario, "reviewer"))["items"]
    assert len(notices) == 1 and notices[0]["target_id"] == ticket["ticket_id"]
    assert notices[0]["page"] == "approvals" and not notices[0]["read"]