e0b69fa2b9
## T1: P0 测试补全（+67 个测试）
- test_utils_llm.py: extract_reply_text / extract_content_from_llm_response / parse_json_from_llm_text 各边界
- test_file_repository.py: 分类 CRUD / 树形结构 / 级联删除 / 文件创建/查询/删除/物理文件清理
- test_report.py: generate_report / generate_compare_report / render_markdown / render_json
- test_llm_score.py: OpenAI 格式 / Anthropic content-block 格式 / JSON 回退解析 / 异常降级
## T2: P1 测试补全（+28 个测试）
- test_scenarios.py: 模板列表/字段完整性/规则类型有效性 + YAML/JSON 加载/校验
- test_webhook.py: 未配置不发送 / 正确 payload / secret header / 异常静默忽略
- test_reports_api.py: GET /reports/{id} / /html / /json / /markdown / /compare 集成测试
## UTC 时区根本修复
- storage/db.py: 新增 iso_utc() 函数，确保所有 datetime 序列化输出带 Z 后缀
- runs.py / files.py / report.py: 6 处 .isoformat() → iso_utc()
- 前端 toDate() 兜底仍保留（向下兼容），但后端不再输出无时区时间戳
Co-Authored-By: Claude <noreply@anthropic.com>
85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
"""Unit tests for webhook notification utility."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
async def test_webhook_not_called_when_url_not_configured():
|
|
with patch("agenteval.utils.webhook.get_settings") as mock_settings:
|
|
mock_settings.return_value.webhook_url = None
|
|
mock_settings.return_value.webhook_secret = None
|
|
|
|
with patch("agenteval.utils.webhook.httpx.AsyncClient") as MockClient:
|
|
from agenteval.utils.webhook import send_run_webhook
|
|
await send_run_webhook("run-1", "completed", {"pass_rate": 1.0})
|
|
MockClient.assert_not_called()
|
|
|
|
|
|
async def test_webhook_posts_correct_payload():
|
|
captured = {}
|
|
|
|
async def fake_post(url, *, json=None, headers=None, **kwargs):
|
|
captured["url"] = url
|
|
captured["json"] = json
|
|
captured["headers"] = headers
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
return resp
|
|
|
|
with patch("agenteval.utils.webhook.get_settings") as mock_settings:
|
|
mock_settings.return_value.webhook_url = "http://hook.example.com/notify"
|
|
mock_settings.return_value.webhook_secret = "secret123"
|
|
|
|
with patch("agenteval.utils.webhook.httpx.AsyncClient") as MockClient:
|
|
instance = MockClient.return_value.__aenter__.return_value
|
|
instance.post = AsyncMock(side_effect=fake_post)
|
|
|
|
from agenteval.utils.webhook import send_run_webhook
|
|
await send_run_webhook("run-42", "completed", {"pass_rate": 0.9})
|
|
|
|
assert captured["url"] == "http://hook.example.com/notify"
|
|
assert captured["json"]["run_id"] == "run-42"
|
|
assert captured["json"]["status"] == "completed"
|
|
assert captured["json"]["event"] == "run_completed"
|
|
assert captured["json"]["summary"]["pass_rate"] == 0.9
|
|
assert captured["headers"]["X-Webhook-Secret"] == "secret123"
|
|
assert "/api/reports/run-42" in captured["json"]["report_url"]
|
|
|
|
|
|
async def test_webhook_no_secret_header():
|
|
captured_headers = {}
|
|
|
|
async def fake_post(url, *, json=None, headers=None, **kwargs):
|
|
captured_headers.update(headers or {})
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
return resp
|
|
|
|
with patch("agenteval.utils.webhook.get_settings") as mock_settings:
|
|
mock_settings.return_value.webhook_url = "http://hook.example.com/notify"
|
|
mock_settings.return_value.webhook_secret = None
|
|
|
|
with patch("agenteval.utils.webhook.httpx.AsyncClient") as MockClient:
|
|
instance = MockClient.return_value.__aenter__.return_value
|
|
instance.post = AsyncMock(side_effect=fake_post)
|
|
|
|
from agenteval.utils.webhook import send_run_webhook
|
|
await send_run_webhook("run-1", "completed", {})
|
|
|
|
assert "X-Webhook-Secret" not in captured_headers
|
|
|
|
|
|
async def test_webhook_silently_ignores_errors():
|
|
with patch("agenteval.utils.webhook.get_settings") as mock_settings:
|
|
mock_settings.return_value.webhook_url = "http://unreachable.example.com"
|
|
mock_settings.return_value.webhook_secret = None
|
|
|
|
with patch("agenteval.utils.webhook.httpx.AsyncClient") as MockClient:
|
|
instance = MockClient.return_value.__aenter__.return_value
|
|
instance.post = AsyncMock(side_effect=Exception("connection refused"))
|
|
|
|
from agenteval.utils.webhook import send_run_webhook
|
|
# Must not raise — errors are silently logged
|
|
await send_run_webhook("run-1", "failed", {})
|