"""Comprehensive tests for NotificationChannel CRUD, permissions, config masking, and test-send.""" import pytest from httpx import AsyncClient from tests.conftest import auth_header BASE = "/api/v1/notifications" # ─── Helpers ──────────────────────────────────────────────────────────── DINGTALK_PAYLOAD = { "name": "钉钉告警通道", "channel_type": "dingtalk", "config": { "webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=abc123", "secret": "SEC_MY_DINGTALK_SECRET", }, "is_enabled": True, } FEISHU_PAYLOAD = { "name": "飞书通知通道", "channel_type": "webhook_feishu", "config": { "webhook_url": "https://open.feishu.cn/open-apis/bot/v2/hook/xxx", "secret": "FEISHU_SIGN_SECRET", }, } EMAIL_PAYLOAD = { "name": "邮件通知通道", "channel_type": "email", "config": { "smtp_host": "smtp.example.com", "smtp_port": 465, "smtp_user": "alert@example.com", "smtp_password": "super_secret_pass", "from_addr": "alert@example.com", "use_tls": True, }, } # ─── CREATE (POST /notifications) ────────────────────────────────────── @pytest.mark.asyncio async def test_create_dingtalk_channel(client: AsyncClient, admin_token: str): resp = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) assert resp.status_code == 201 data = resp.json() assert data["name"] == "钉钉告警通道" assert data["channel_type"] == "dingtalk" assert data["is_enabled"] is True assert "id" in data assert data["config"]["webhook_url"] == DINGTALK_PAYLOAD["config"]["webhook_url"] # secret must be masked assert data["config"]["secret"] == "******" @pytest.mark.asyncio async def test_create_feishu_channel(client: AsyncClient, admin_token: str): resp = await client.post(BASE, json=FEISHU_PAYLOAD, headers=auth_header(admin_token)) assert resp.status_code == 201 assert resp.json()["channel_type"] == "webhook_feishu" @pytest.mark.asyncio async def test_create_email_channel(client: AsyncClient, admin_token: str): resp = await client.post(BASE, json=EMAIL_PAYLOAD, headers=auth_header(admin_token)) assert resp.status_code == 201 data = resp.json() assert data["channel_type"] == "email" # smtp_password must be masked assert data["config"]["smtp_password"] == "******" # non-sensitive fields preserved assert data["config"]["smtp_host"] == "smtp.example.com" assert data["config"]["smtp_port"] == 465 @pytest.mark.asyncio async def test_create_channel_default_enabled(client: AsyncClient, admin_token: str): payload = { "name": "默认启用", "channel_type": "dingtalk", "config": {"webhook_url": "https://example.com/hook"}, } resp = await client.post(BASE, json=payload, headers=auth_header(admin_token)) assert resp.status_code == 201 assert resp.json()["is_enabled"] is True @pytest.mark.asyncio async def test_create_channel_disabled(client: AsyncClient, admin_token: str): payload = { "name": "已禁用通道", "channel_type": "dingtalk", "config": {"webhook_url": "https://example.com/hook"}, "is_enabled": False, } resp = await client.post(BASE, json=payload, headers=auth_header(admin_token)) assert resp.status_code == 201 assert resp.json()["is_enabled"] is False @pytest.mark.asyncio async def test_create_channel_invalid_type(client: AsyncClient, admin_token: str): payload = { "name": "坏通道", "channel_type": "slack", "config": {"webhook_url": "https://example.com"}, } resp = await client.post(BASE, json=payload, headers=auth_header(admin_token)) assert resp.status_code == 422 @pytest.mark.asyncio async def test_create_channel_empty_name(client: AsyncClient, admin_token: str): payload = { "name": "", "channel_type": "dingtalk", "config": {"webhook_url": "https://example.com"}, } resp = await client.post(BASE, json=payload, headers=auth_header(admin_token)) assert resp.status_code == 422 @pytest.mark.asyncio async def test_create_channel_name_too_long(client: AsyncClient, admin_token: str): payload = { "name": "x" * 101, "channel_type": "dingtalk", "config": {"webhook_url": "https://example.com"}, } resp = await client.post(BASE, json=payload, headers=auth_header(admin_token)) assert resp.status_code == 422 # ─── LIST (GET /notifications) ───────────────────────────────────────── @pytest.mark.asyncio async def test_list_channels_empty(client: AsyncClient, admin_token: str): resp = await client.get(BASE, headers=auth_header(admin_token)) assert resp.status_code == 200 assert resp.json() == [] @pytest.mark.asyncio async def test_list_channels_returns_all(client: AsyncClient, admin_token: str): for payload in [DINGTALK_PAYLOAD, FEISHU_PAYLOAD, EMAIL_PAYLOAD]: await client.post(BASE, json=payload, headers=auth_header(admin_token)) resp = await client.get(BASE, headers=auth_header(admin_token)) assert resp.status_code == 200 items = resp.json() assert len(items) == 3 types = {ch["channel_type"] for ch in items} assert types == {"dingtalk", "webhook_feishu", "email"} @pytest.mark.asyncio async def test_list_channels_config_masked(client: AsyncClient, admin_token: str): await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) resp = await client.get(BASE, headers=auth_header(admin_token)) items = resp.json() assert len(items) == 1 assert items[0]["config"]["secret"] == "******" # ─── UPDATE (PUT /notifications/{channel_id}) ────────────────────────── @pytest.mark.asyncio async def test_update_channel_name(client: AsyncClient, admin_token: str): create = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.put( f"{BASE}/{cid}", json={"name": "更新后的钉钉通道"}, headers=auth_header(admin_token), ) assert resp.status_code == 200 assert resp.json()["name"] == "更新后的钉钉通道" @pytest.mark.asyncio async def test_update_channel_toggle_enabled(client: AsyncClient, admin_token: str): create = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.put( f"{BASE}/{cid}", json={"is_enabled": False}, headers=auth_header(admin_token), ) assert resp.status_code == 200 assert resp.json()["is_enabled"] is False @pytest.mark.asyncio async def test_update_channel_config(client: AsyncClient, admin_token: str): create = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] new_config = {"webhook_url": "https://new-hook.example.com", "secret": "NEW_SECRET"} resp = await client.put( f"{BASE}/{cid}", json={"config": new_config}, headers=auth_header(admin_token), ) assert resp.status_code == 200 data = resp.json() assert data["config"]["webhook_url"] == "https://new-hook.example.com" # secret masked in response assert data["config"]["secret"] == "******" @pytest.mark.asyncio async def test_update_nonexistent_channel(client: AsyncClient, admin_token: str): resp = await client.put( f"{BASE}/99999", json={"name": "不存在"}, headers=auth_header(admin_token), ) assert resp.status_code == 404 @pytest.mark.asyncio async def test_update_channel_invalid_type(client: AsyncClient, admin_token: str): create = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.put( f"{BASE}/{cid}", json={"channel_type": "invalid_type"}, headers=auth_header(admin_token), ) assert resp.status_code == 422 # ─── DELETE (DELETE /notifications/{channel_id}) ─────────────────────── @pytest.mark.asyncio async def test_delete_channel(client: AsyncClient, admin_token: str): create = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.delete(f"{BASE}/{cid}", headers=auth_header(admin_token)) assert resp.status_code == 200 assert "message" in resp.json() # Verify it's gone from the list items = (await client.get(BASE, headers=auth_header(admin_token))).json() assert all(ch["id"] != cid for ch in items) @pytest.mark.asyncio async def test_delete_nonexistent_channel(client: AsyncClient, admin_token: str): resp = await client.delete(f"{BASE}/99999", headers=auth_header(admin_token)) assert resp.status_code == 404 # ─── TEST SEND (POST /notifications/{channel_id}/test) ───────────────── @pytest.mark.asyncio async def test_send_dingtalk_success(client: AsyncClient, admin_token: str): create = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.post(f"{BASE}/{cid}/test", headers=auth_header(admin_token)) assert resp.status_code == 200 data = resp.json() assert data["success"] is True assert "钉钉" in data["message"] @pytest.mark.asyncio async def test_send_dingtalk_missing_webhook(client: AsyncClient, admin_token: str): payload = { "name": "缺少webhook的钉钉", "channel_type": "dingtalk", "config": {"secret": "abc"}, } create = await client.post(BASE, json=payload, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.post(f"{BASE}/{cid}/test", headers=auth_header(admin_token)) assert resp.status_code == 200 data = resp.json() assert data["success"] is False assert "webhook_url" in data["message"] @pytest.mark.asyncio async def test_send_feishu_success(client: AsyncClient, admin_token: str): create = await client.post(BASE, json=FEISHU_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.post(f"{BASE}/{cid}/test", headers=auth_header(admin_token)) assert resp.status_code == 200 data = resp.json() assert data["success"] is True assert "飞书" in data["message"] @pytest.mark.asyncio async def test_send_feishu_missing_webhook(client: AsyncClient, admin_token: str): payload = { "name": "缺少webhook的飞书", "channel_type": "webhook_feishu", "config": {}, } create = await client.post(BASE, json=payload, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.post(f"{BASE}/{cid}/test", headers=auth_header(admin_token)) assert resp.status_code == 200 assert resp.json()["success"] is False assert "webhook_url" in resp.json()["message"] @pytest.mark.asyncio async def test_send_email_success(client: AsyncClient, admin_token: str): create = await client.post(BASE, json=EMAIL_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.post(f"{BASE}/{cid}/test", headers=auth_header(admin_token)) assert resp.status_code == 200 data = resp.json() assert data["success"] is True assert "邮件" in data["message"] @pytest.mark.asyncio async def test_send_email_missing_smtp_config(client: AsyncClient, admin_token: str): payload = { "name": "不完整的邮件", "channel_type": "email", "config": {"smtp_host": "smtp.example.com"}, # missing smtp_port, smtp_user, from_addr } create = await client.post(BASE, json=payload, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.post(f"{BASE}/{cid}/test", headers=auth_header(admin_token)) assert resp.status_code == 200 data = resp.json() assert data["success"] is False assert "smtp_port" in data["message"] assert "smtp_user" in data["message"] assert "from_addr" in data["message"] @pytest.mark.asyncio async def test_send_test_nonexistent_channel(client: AsyncClient, admin_token: str): resp = await client.post(f"{BASE}/99999/test", headers=auth_header(admin_token)) assert resp.status_code == 404 # ─── PERMISSION CHECKS (all admin-only, others get 403) ──────────────── @pytest.mark.asyncio async def test_permission_dba_list_forbidden(client: AsyncClient, dba_token: str): resp = await client.get(BASE, headers=auth_header(dba_token)) assert resp.status_code == 403 @pytest.mark.asyncio async def test_permission_developer_list_forbidden(client: AsyncClient, developer_token: str): resp = await client.get(BASE, headers=auth_header(developer_token)) assert resp.status_code == 403 @pytest.mark.asyncio async def test_permission_guest_list_forbidden(client: AsyncClient, guest_token: str): resp = await client.get(BASE, headers=auth_header(guest_token)) assert resp.status_code == 403 @pytest.mark.asyncio async def test_permission_dba_create_forbidden(client: AsyncClient, dba_token: str): resp = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(dba_token)) assert resp.status_code == 403 @pytest.mark.asyncio async def test_permission_developer_update_forbidden(client: AsyncClient, developer_token: str): resp = await client.put(f"{BASE}/1", json={"name": "x"}, headers=auth_header(developer_token)) assert resp.status_code == 403 @pytest.mark.asyncio async def test_permission_guest_delete_forbidden(client: AsyncClient, guest_token: str): resp = await client.delete(f"{BASE}/1", headers=auth_header(guest_token)) assert resp.status_code == 403 @pytest.mark.asyncio async def test_permission_dba_test_send_forbidden(client: AsyncClient, dba_token: str): resp = await client.post(f"{BASE}/1/test", headers=auth_header(dba_token)) assert resp.status_code == 403 # ─── EDGE CASES ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_no_auth_returns_401(client: AsyncClient): """All endpoints require authentication — without a token expect 401.""" for method, url in [ ("get", BASE), ("post", BASE), ("put", f"{BASE}/1"), ("delete", f"{BASE}/1"), ("post", f"{BASE}/1/test"), ]: resp = await getattr(client, method)(url) assert resp.status_code == 401, f"{method.upper()} {url} should be 401, got {resp.status_code}" @pytest.mark.asyncio async def test_config_without_sensitive_keys(client: AsyncClient, admin_token: str): """Config with no sensitive keys should pass through unchanged.""" payload = { "name": "无敏感配置", "channel_type": "dingtalk", "config": {"webhook_url": "https://example.com/hook", "extra": "value"}, } resp = await client.post(BASE, json=payload, headers=auth_header(admin_token)) assert resp.status_code == 201 config = resp.json()["config"] assert config["webhook_url"] == "https://example.com/hook" assert config["extra"] == "value" assert "secret" not in config @pytest.mark.asyncio async def test_update_partial_preserves_others(client: AsyncClient, admin_token: str): """Updating only name should not change config or channel_type.""" create = await client.post(BASE, json=DINGTALK_PAYLOAD, headers=auth_header(admin_token)) cid = create.json()["id"] resp = await client.put( f"{BASE}/{cid}", json={"name": "只改名字"}, headers=auth_header(admin_token), ) assert resp.status_code == 200 data = resp.json() assert data["name"] == "只改名字" assert data["channel_type"] == "dingtalk" assert data["config"]["webhook_url"] == DINGTALK_PAYLOAD["config"]["webhook_url"]