cma-management/backend/tests/test_notifications.py

"""通知配置模块测试 — 渠道CRUD + 推送历史"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from tests.conftest import create_test_user, create_test_kpi, get_token_for_user, auth_header


class TestNotificationChannels:
    """通知渠道CRUD测试"""

    BASE = "/api/cma/notifications"

    def test_list_channels_empty(self, client: TestClient, db: Session):
        """空列表"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.get(f"{self.BASE}/channels", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["data"] == []

    def test_create_wecom_channel(self, client: TestClient, db: Session):
        """创建企微机器人渠道"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.post(
            f"{self.BASE}/channels",
            headers=auth_header(token),
            json={
                "name": "告警群",
                "channel_type": "wecom",
                "config": {"webhook_url": "https://qyapi.weixin.qq.com/webhook/abc"},
                "enabled": True,
            },
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["name"] == "告警群"
        assert data["channel_type"] == "wecom"
        assert data["enabled"] is True

    def test_create_mail_channel(self, client: TestClient, db: Session):
        """创建邮件渠道"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.post(
            f"{self.BASE}/channels",
            headers=auth_header(token),
            json={
                "name": "管理邮箱",
                "channel_type": "mail",
                "config": {
                    "host": "smtp.qq.com",
                    "port": 465,
                    "user": "admin@example.com",
                    "password": "pass",
                    "to": ["admin@example.com"],
                },
            },
        )
        assert resp.status_code == 200
        assert resp.json()["channel_type"] == "mail"

    def test_update_channel(self, client: TestClient, db: Session):
        """更新渠道"""
        create_test_user(db)
        token = get_token_for_user(client)

        create_resp = client.post(
            f"{self.BASE}/channels",
            headers=auth_header(token),
            json={"name": "旧名称", "channel_type": "wecom", "config": {}},
        )
        ch_id = create_resp.json()["id"]

        resp = client.put(
            f"{self.BASE}/channels/{ch_id}",
            headers=auth_header(token),
            json={"name": "新名称", "enabled": False},
        )
        assert resp.status_code == 200
        assert resp.json()["name"] == "新名称"
        assert resp.json()["enabled"] is False

    def test_delete_channel(self, client: TestClient, db: Session):
        """删除渠道"""
        create_test_user(db)
        token = get_token_for_user(client)

        create_resp = client.post(
            f"{self.BASE}/channels",
            headers=auth_header(token),
            json={"name": "待删除", "channel_type": "wecom", "config": {}},
        )
        ch_id = create_resp.json()["id"]

        resp = client.delete(
            f"{self.BASE}/channels/{ch_id}",
            headers=auth_header(token),
        )
        assert resp.status_code == 200
        assert resp.json()["message"] == "已删除"

        # 验证已删除
        get_resp = client.get(f"{self.BASE}/channels", headers=auth_header(token))
        assert len(get_resp.json()["data"]) == 0

    def test_delete_channel_not_found(self, client: TestClient, db: Session):
        """删除不存在的渠道"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.delete(f"{self.BASE}/channels/99999", headers=auth_header(token))
        assert resp.status_code == 404

    def test_test_channel_network(self, client: TestClient, db: Session):
        """测试推送（网络可能不通，但不应500）"""
        create_test_user(db)
        token = get_token_for_user(client)

        create_resp = client.post(
            f"{self.BASE}/channels",
            headers=auth_header(token),
            json={"name": "测试", "channel_type": "wecom", "config": {"webhook_url": "http://invalid"}, "enabled": True},
        )
        ch_id = create_resp.json()["id"]

        # 网络不通，但应该返回结果而不是500
        resp = client.post(
            f"{self.BASE}/channels/{ch_id}/test",
            headers=auth_header(token),
        )
        assert resp.status_code == 200
        assert "results" in resp.json()


class TestNotificationLogs:
    """通知历史测试"""

    def test_logs_empty(self, client: TestClient, db: Session):
        """空日志"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.get("/api/cma/notifications/logs", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["total"] == 0

    def test_logs_with_data(self, client: TestClient, db: Session):
        """有了日志记录后可查询

        2026-09-22（dama-workbench-entity-isolation-fullstack-20260922 §四 B2）行为变更：
        `notification_logs` 无 entity_id 列 → 改为**经父键 join 归属**
        （`alert_id → kpi_alerts.id → kpi_alerts.entity_id`）。
        故日志必须挂到**本账套真实存在**的 KPIAlert 上才可见；
        `alert_id IS NULL` 或指向不存在父键的孤儿行**不返回**（宁少勿错）。
        """
        from app.models import NotificationLog, KPIAlert
        from datetime import datetime

        kpi = create_test_kpi(db, kpi_code="NOTIF_LOG_KPI", entity_id=1)
        alert = KPIAlert(entity_id=1, kpi_id=kpi.id, alert_level="yellow",
                         alert_message="通知日志用例预警", status="pending")
        db.add(alert)
        db.commit()
        db.refresh(alert)

        db.add_all([
            NotificationLog(
                alert_id=alert.id, channel="wecom", recipient="测试群",
                title="测试通知", content="测试内容", status="sent", sent_at=datetime.now(),
            ),
            # 孤儿行：无父键可归属 → 一律不返回
            NotificationLog(alert_id=None, channel="wecom", title="无归属通知", status="sent"),
        ])
        db.commit()

        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.get("/api/cma/notifications/logs", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["total"] == 1, resp.json()
        assert len(resp.json()["data"]) == 1
        assert resp.json()["data"][0]["title"] == "测试通知"


class TestAlertPush:
    """手动推送测试"""

    def test_push_alerts_no_channels(self, client: TestClient, db: Session):
        """推送端点已移除，预期404"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.post("/api/cma/notifications/alerts/push", headers=auth_header(token))
        assert resp.status_code == 404, "push端点已移除"