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, 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):
        """有了日志记录后可查询"""
        from app.models import NotificationLog
        from datetime import datetime

        # 直接插入日志
        log = NotificationLog(
            alert_id=1, channel="wecom", recipient="测试群",
            title="测试通知", content="测试内容", status="sent", sent_at=datetime.now(),
        )
        db.add(log)
        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
        assert len(resp.json()["data"]) >= 1


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端点已移除"