84 lines
3.2 KiB
Python
84 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from app.services import debug_event_service
|
|
|
|
|
|
class DebugEventServiceTests(unittest.TestCase):
|
|
def test_ingest_debug_event_requires_configured_shared_token(self) -> None:
|
|
with patch.object(debug_event_service.settings, "sync_shared_token", ""):
|
|
ok, message, data = debug_event_service.ingest_debug_event(
|
|
{
|
|
"source_region": "mainland",
|
|
"node_code": "mainland-controller-01",
|
|
"service": "worker-event",
|
|
"event_type": "worker_log",
|
|
"message": "开始检测域名: a.com",
|
|
"payload": {"job_id": 1, "job_code": "sync-overseas-1"},
|
|
},
|
|
shared_token=None,
|
|
)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertIn("token 未配置", message)
|
|
self.assertTrue(data["configuration_required"])
|
|
|
|
def test_resolve_target_job_for_debug_event_rejects_missing_identity(self) -> None:
|
|
job, reason = debug_event_service._resolve_target_job_for_debug_event({"domain": "a.com"})
|
|
|
|
self.assertIsNone(job)
|
|
self.assertEqual("missing_job_identity", reason)
|
|
|
|
@patch("app.services.debug_event_service._load_detect_job_summary_by_job_code")
|
|
@patch("app.services.detect_job_service.get_detect_job_summary")
|
|
@patch("app.services.detect_job_service.get_active_detect_job_summary")
|
|
def test_resolve_target_job_for_debug_event_prefers_payload_job_id_over_current_active_job(
|
|
self,
|
|
mock_get_active_detect_job_summary,
|
|
mock_get_detect_job_summary,
|
|
mock_load_detect_job_summary_by_job_code,
|
|
) -> None:
|
|
mock_get_active_detect_job_summary.return_value = {
|
|
"job_id": 12,
|
|
"job_code": "sync-overseas-12",
|
|
"current_cycle_token": "cycle-12",
|
|
}
|
|
mock_get_detect_job_summary.return_value = {
|
|
"job_id": 11,
|
|
"job_code": "sync-overseas-11",
|
|
"current_cycle_token": "cycle-11",
|
|
}
|
|
|
|
job, reason = debug_event_service._resolve_target_job_for_debug_event(
|
|
{"job_id": 11, "job_code": "sync-overseas-11", "cycle_token": "cycle-11"}
|
|
)
|
|
|
|
self.assertEqual("matched", reason)
|
|
self.assertEqual(11, job["job_id"])
|
|
mock_get_detect_job_summary.assert_called_once_with(11, event_limit=1)
|
|
mock_load_detect_job_summary_by_job_code.assert_not_called()
|
|
|
|
@patch("app.services.detect_job_service.get_active_detect_job_summary")
|
|
def test_resolve_target_job_for_debug_event_rejects_cycle_mismatch(
|
|
self,
|
|
mock_get_active_detect_job_summary,
|
|
) -> None:
|
|
mock_get_active_detect_job_summary.return_value = {
|
|
"job_id": 12,
|
|
"job_code": "sync-overseas-12",
|
|
"current_cycle_token": "cycle-current",
|
|
}
|
|
|
|
job, reason = debug_event_service._resolve_target_job_for_debug_event(
|
|
{"job_id": 12, "job_code": "sync-overseas-12", "cycle_token": "cycle-old"}
|
|
)
|
|
|
|
self.assertIsNone(job)
|
|
self.assertEqual("cycle_mismatch", reason)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|