118 lines
4.4 KiB
Python
118 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from monitors.yumi.tencent_cdn import get_cdn_overview_payload, submit_cdn_purge_payload
|
|
|
|
|
|
class CdnRefreshTests(unittest.TestCase):
|
|
@patch("monitors.yumi.tencent_cdn.CDN_PURGE_TASK_LIMIT", 5)
|
|
@patch("monitors.yumi.tencent_cdn.call_cdn")
|
|
@patch("monitors.yumi.tencent_cdn.cdn_configuration_error", return_value="")
|
|
def test_overview_combines_quota_and_recent_tasks(self, config_error_mock, call_cdn_mock) -> None:
|
|
del config_error_mock
|
|
|
|
def side_effect(method_name: str, payload: dict) -> dict:
|
|
if method_name == "DescribePurgeQuota":
|
|
self.assertEqual(payload, {})
|
|
return {
|
|
"UrlPurge": [
|
|
{"Area": "mainland", "Batch": 1000, "Total": 10000, "Available": 9999},
|
|
],
|
|
"PathPurge": [
|
|
{"Area": "overseas", "Batch": 500, "Total": 2000, "Available": 1800},
|
|
],
|
|
}
|
|
if method_name == "DescribePurgeTasks":
|
|
self.assertEqual(payload["Limit"], 5)
|
|
return {
|
|
"TotalCount": 1,
|
|
"PurgeLogs": [
|
|
{
|
|
"TaskId": "task-1",
|
|
"Url": "https://h5.haiyihy.com/index.html",
|
|
"Status": "done",
|
|
"PurgeType": "url",
|
|
"FlushType": "",
|
|
"CreateTime": "2026-04-23 12:00:00",
|
|
}
|
|
],
|
|
}
|
|
raise AssertionError(f"unexpected method: {method_name}")
|
|
|
|
call_cdn_mock.side_effect = side_effect
|
|
|
|
payload = get_cdn_overview_payload()
|
|
|
|
self.assertTrue(payload["ok"])
|
|
self.assertTrue(payload["configured"])
|
|
self.assertEqual(payload["quota"]["url"][0]["area"], "mainland")
|
|
self.assertEqual(payload["quota"]["path"][0]["available"], 1800)
|
|
self.assertEqual(payload["tasks"][0]["taskId"], "task-1")
|
|
self.assertEqual(payload["taskTotal"], 1)
|
|
|
|
@patch("monitors.yumi.tencent_cdn.CDN_ALLOWED_DOMAINS", ("h5.haiyihy.com",))
|
|
@patch("monitors.yumi.tencent_cdn.call_cdn", return_value={"TaskId": "task-2", "RequestId": "req-2"})
|
|
@patch("monitors.yumi.tencent_cdn.cdn_configuration_error", return_value="")
|
|
def test_submit_url_refresh_uses_allowlisted_domain(
|
|
self,
|
|
config_error_mock,
|
|
call_cdn_mock,
|
|
) -> None:
|
|
del config_error_mock
|
|
|
|
payload = submit_cdn_purge_payload(
|
|
{
|
|
"purgeType": "url",
|
|
"items": [
|
|
"https://h5.haiyihy.com/index.html",
|
|
"https://h5.haiyihy.com/index.html",
|
|
],
|
|
"area": "mainland",
|
|
"urlEncode": True,
|
|
}
|
|
)
|
|
|
|
self.assertTrue(payload["ok"])
|
|
self.assertEqual(payload["taskId"], "task-2")
|
|
self.assertEqual(payload["itemCount"], 1)
|
|
call_cdn_mock.assert_called_once_with(
|
|
"PurgeUrlsCache",
|
|
{
|
|
"UrlEncode": True,
|
|
"Area": "mainland",
|
|
"Urls": ["https://h5.haiyihy.com/index.html"],
|
|
},
|
|
)
|
|
|
|
@patch("monitors.yumi.tencent_cdn.CDN_ALLOWED_DOMAINS", ("h5.haiyihy.com",))
|
|
@patch("monitors.yumi.tencent_cdn.cdn_configuration_error", return_value="")
|
|
def test_submit_rejects_unknown_domain(self, config_error_mock) -> None:
|
|
del config_error_mock
|
|
|
|
with self.assertRaisesRegex(ValueError, "CDN_ALLOWED_DOMAINS"):
|
|
submit_cdn_purge_payload(
|
|
{
|
|
"purgeType": "url",
|
|
"items": ["https://static.example.com/app.js"],
|
|
}
|
|
)
|
|
|
|
@patch("monitors.yumi.tencent_cdn.CDN_ALLOWED_DOMAINS", ("h5.haiyihy.com",))
|
|
@patch("monitors.yumi.tencent_cdn.cdn_configuration_error", return_value="")
|
|
def test_path_refresh_requires_trailing_slash(self, config_error_mock) -> None:
|
|
del config_error_mock
|
|
|
|
with self.assertRaisesRegex(ValueError, "以 / 结尾"):
|
|
submit_cdn_purge_payload(
|
|
{
|
|
"purgeType": "path",
|
|
"items": ["https://h5.haiyihy.com/assets"],
|
|
}
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|