document-management/backend/tests/dms/test_notifications.py

"""站内通知接口测试（契约 1.34）。"""

from __future__ import annotations

from datetime import datetime

import pytest
from sqlalchemy import delete, select

from dms.extensions import db
from dms.models.notification import Notification


def auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture()
def notification_app(b2_app):
    """预置通知数据：admin 3 条（2 未读 1 已读、2 待办），user 1 条未读。"""
    with b2_app.app_context():
        db.session.execute(delete(Notification))
        from dms.models import User

        admin = db.session.scalar(select(User).where(User.username == "admin"))
        user = db.session.scalar(select(User).where(User.username == "user"))
        assert admin is not None and user is not None
        rows = [
            Notification(
                user_id=admin.id,
                title="待审批：作战方案A",
                content="作战方案A等待您审批",
                notify_type="APPROVAL",
                action="PENDING",
                status="UNREAD",
                created_at=datetime(2026, 9, 1, 0, 0),
            ),
            Notification(
                user_id=admin.id,
                title="已处理：通信方案B",
                content="通信方案B审批",
                notify_type="APPROVAL",
                action="AGREE",
                status="READ",
                created_at=datetime(2026, 9, 2, 0, 0),
            ),
            Notification(
                user_id=admin.id,
                title="无待办：系统公告",
                content="系统维护通知",
                notify_type="APPROVAL",
                action="NONE",
                status="UNREAD",
                created_at=datetime(2026, 9, 3, 0, 0),
            ),
            Notification(
                user_id=user.id,
                title="用户待办",
                content="用户自己的待办",
                notify_type="APPROVAL",
                action="PENDING",
                status="UNREAD",
                created_at=datetime(2026, 9, 4, 0, 0),
            ),
        ]
        db.session.add_all(rows)
        db.session.commit()
        b2_app.config["NOTIFICATION_IDS"] = {row.title: row.id for row in rows}
    yield b2_app
    with b2_app.app_context():
        db.session.execute(delete(Notification))
        db.session.commit()


@pytest.fixture()
def notification_client(notification_app):
    return notification_app.test_client()


def test_notifications_require_valid_token(notification_client):
    response = notification_client.get("/api/v1/notifications")
    assert response.status_code == 401
    assert response.get_json()["code"] == "TOKEN_INVALID"


def test_list_returns_only_own_notifications_id_desc(
    notification_client, token_for
):
    response = notification_client.get(
        "/api/v1/notifications", headers=auth(token_for("admin"))
    )
    assert response.status_code == 200
    items = response.get_json()["data"]["items"]
    assert [item["title"] for item in items] == [
        "无待办：系统公告",
        "已处理：通信方案B",
        "待审批：作战方案A",
    ]
    first = items[0]
    assert first["id"].isdecimal()
    assert first["notifyType"] == "APPROVAL"
    assert first["status"] == "UNREAD"
    assert first["createdAt"].endswith("Z")


def test_unread_count_per_user(notification_client, token_for):
    admin = notification_client.get(
        "/api/v1/notifications/unread-count", headers=auth(token_for("admin"))
    )
    user = notification_client.get(
        "/api/v1/notifications/unread-count", headers=auth(token_for("user"))
    )
    assert admin.get_json()["data"] == {"count": 2}
    assert user.get_json()["data"] == {"count": 1}


def test_read_all_clears_own_unread_only(notification_client, notification_app, token_for):
    response = notification_client.post(
        "/api/v1/notifications/read-all", headers=auth(token_for("admin"))
    )
    assert response.status_code == 200
    assert response.get_json()["data"]["updated"] == 2

    admin = notification_client.get(
        "/api/v1/notifications/unread-count", headers=auth(token_for("admin"))
    )
    user = notification_client.get(
        "/api/v1/notifications/unread-count", headers=auth(token_for("user"))
    )
    assert admin.get_json()["data"]["count"] == 0
    assert user.get_json()["data"]["count"] == 1

    # 已处理记录的 action 不受全部已读影响
    with notification_app.app_context():
        handled = db.session.get(
            Notification, notification_app.config["NOTIFICATION_IDS"]["已处理：通信方案B"]
        )
        assert handled is not None and handled.action == "AGREE"


@pytest.mark.parametrize("action,expected", [("AGREE", "AGREE"), ("REJECT", "REJECT")])
def test_handle_pending_notification(
    notification_client, notification_app, token_for, action, expected
):
    notification_id = notification_app.config["NOTIFICATION_IDS"]["待审批：作战方案A"]
    response = notification_client.post(
        f"/api/v1/notifications/{notification_id}/handle",
        headers=auth(token_for("admin")),
        json={"action": action},
    )
    assert response.status_code == 200
    data = response.get_json()["data"]
    assert data["action"] == expected
    assert data["status"] == "READ"
    assert data["handledAt"] is not None


def test_handle_twice_returns_conflict(notification_client, notification_app, token_for):
    notification_id = notification_app.config["NOTIFICATION_IDS"]["待审批：作战方案A"]
    headers = auth(token_for("admin"))
    first = notification_client.post(
        f"/api/v1/notifications/{notification_id}/handle",
        headers=headers,
        json={"action": "AGREE"},
    )
    second = notification_client.post(
        f"/api/v1/notifications/{notification_id}/handle",
        headers=headers,
        json={"action": "REJECT"},
    )
    assert first.status_code == 200
    assert second.status_code == 409
    assert second.get_json()["code"] == "DATA_VERSION_CONFLICT"


def test_handle_none_action_returns_conflict(
    notification_client, notification_app, token_for
):
    notification_id = notification_app.config["NOTIFICATION_IDS"]["无待办：系统公告"]
    response = notification_client.post(
        f"/api/v1/notifications/{notification_id}/handle",
        headers=auth(token_for("admin")),
        json={"action": "AGREE"},
    )
    assert response.status_code == 409


def test_handle_other_users_notification_returns_404(
    notification_client, notification_app, token_for
):
    notification_id = notification_app.config["NOTIFICATION_IDS"]["用户待办"]
    response = notification_client.post(
        f"/api/v1/notifications/{notification_id}/handle",
        headers=auth(token_for("admin")),
        json={"action": "AGREE"},
    )
    assert response.status_code == 404
    assert response.get_json()["code"] == "RESOURCE_NOT_FOUND"


@pytest.mark.parametrize(
    "path,body",
    [
        ("/api/v1/notifications/0/handle", {"action": "AGREE"}),
        ("/api/v1/notifications/abc/handle", {"action": "AGREE"}),
        ("/api/v1/notifications/1/handle", {"action": "FOO"}),
        ("/api/v1/notifications/1/handle", {}),
        ("/api/v1/notifications/1/handle", "not-a-dict"),
    ],
)
def test_handle_invalid_arguments(notification_client, token_for, path, body):
    response = notification_client.post(
        path, headers=auth(token_for("admin")), json=body
    )
    assert response.status_code == 400
    assert response.get_json()["code"] == "INVALID_ARGUMENT"