"""站内通知接口测试（契约 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"