330 lines
14 KiB
Python
330 lines
14 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from app.sync_agent import (
|
|
_append_detect_result_projection_snapshot,
|
|
_build_aligned_queue_health_snapshot,
|
|
_emit_runtime_debug_snapshots,
|
|
_maybe_trigger_overlap_start,
|
|
_run_sync_tick_once,
|
|
_emit_structured_tick,
|
|
_emit_sync_result_breakdown,
|
|
_filter_runtime_events_for_job,
|
|
_run_pipeline_stage_processor,
|
|
_select_projection_job_snapshots,
|
|
)
|
|
|
|
|
|
class SyncAgentTests(unittest.TestCase):
|
|
@patch("app.sync_agent.append_detect_result_projection_if_changed")
|
|
def test_append_detect_result_projection_snapshot_uses_active_job_summary(self, mock_append) -> None:
|
|
_append_detect_result_projection_snapshot(
|
|
{
|
|
"job_id": 1,
|
|
"job_code": "detect-20260419030000-abc123",
|
|
"status": "running",
|
|
"items_pending": 12,
|
|
"items_running": 3,
|
|
"items_completed": 8,
|
|
"items_blacklisted": 1,
|
|
"items_failed": 2,
|
|
}
|
|
)
|
|
|
|
mock_append.assert_called_once()
|
|
payload = mock_append.call_args.kwargs["detect"]
|
|
self.assertEqual("running", payload["phase_label"])
|
|
self.assertEqual("detect-20260419030000-abc123", payload["active_job"]["job_code"])
|
|
self.assertEqual(12, payload["progress"]["pending"])
|
|
self.assertEqual(3, payload["progress"]["running"])
|
|
self.assertEqual(8, payload["progress"]["completed"])
|
|
self.assertEqual(1, payload["progress"]["blacklisted"])
|
|
self.assertEqual(2, payload["progress"]["failed"])
|
|
|
|
@patch("app.sync_agent.get_latest_unprojected_detect_job_summary")
|
|
@patch("app.sync_agent.get_active_detect_job_summary")
|
|
def test_select_projection_job_snapshots_returns_active_and_latest_finished(
|
|
self,
|
|
mock_get_active_detect_job_summary,
|
|
mock_get_latest_unprojected_detect_job_summary,
|
|
) -> None:
|
|
mock_get_active_detect_job_summary.return_value = {"job_id": 101, "job_code": "running-job"}
|
|
mock_get_latest_unprojected_detect_job_summary.return_value = {"job_id": 102, "job_code": "finished-job"}
|
|
|
|
snapshots = _select_projection_job_snapshots()
|
|
|
|
self.assertEqual(["running-job", "finished-job"], [item["job_code"] for item in snapshots])
|
|
|
|
@patch("app.sync_agent.get_latest_unprojected_detect_job_summary")
|
|
@patch("app.sync_agent.get_active_detect_job_summary")
|
|
def test_select_projection_job_snapshots_deduplicates_same_job(
|
|
self,
|
|
mock_get_active_detect_job_summary,
|
|
mock_get_latest_unprojected_detect_job_summary,
|
|
) -> None:
|
|
mock_get_active_detect_job_summary.return_value = {"job_id": 101, "job_code": "same-job"}
|
|
mock_get_latest_unprojected_detect_job_summary.return_value = {"job_id": 101, "job_code": "same-job"}
|
|
|
|
snapshots = _select_projection_job_snapshots()
|
|
|
|
self.assertEqual(1, len(snapshots))
|
|
self.assertEqual("same-job", snapshots[0]["job_code"])
|
|
|
|
def test_build_aligned_queue_health_snapshot_overrides_mismatched_job(self) -> None:
|
|
active_job = {
|
|
"job_id": 1891,
|
|
"job_code": "sync-overseas-31230",
|
|
"status": "running",
|
|
"progress_percent": 75.9,
|
|
"items_total": 1000,
|
|
"items_pending": 155,
|
|
"items_claimed": 45,
|
|
"items_running": 41,
|
|
"items_completed": 759,
|
|
"items_blacklisted": 0,
|
|
"items_failed": 0,
|
|
"node_stats": [
|
|
{
|
|
"node_code": "mainland-worker-01",
|
|
"items_total": 710,
|
|
"items_claimed": 45,
|
|
"items_running": 41,
|
|
"items_completed": 624,
|
|
}
|
|
],
|
|
}
|
|
queue_health = {
|
|
"job": {"job_code": "sync-overseas-31273"},
|
|
"queue": {"items_total": 1000, "pending": 838, "claimed": 162, "running": 0, "completed": 0},
|
|
"nodes": [{"node_code": "mainland-controller-01", "items_total": 162}],
|
|
}
|
|
|
|
data = _build_aligned_queue_health_snapshot(active_job, queue_health)
|
|
|
|
self.assertEqual("sync-overseas-31230", data["job"]["job_code"])
|
|
self.assertEqual(155, data["queue"]["pending"])
|
|
self.assertEqual(45, data["queue"]["claimed"])
|
|
self.assertEqual(41, data["queue"]["running"])
|
|
self.assertEqual(759, data["queue"]["completed"])
|
|
self.assertEqual("mainland-worker-01", data["nodes"][0]["node_code"])
|
|
self.assertEqual("unassigned", data["nodes"][-1]["node_code"])
|
|
|
|
@patch("app.sync_agent.push_debug_event")
|
|
def test_emit_sync_result_breakdown_marks_idle_projection_as_idle(self, mock_push_debug_event) -> None:
|
|
_emit_sync_result_breakdown(
|
|
{
|
|
"results": [
|
|
{
|
|
"sync_type": "detect_result_projection",
|
|
"ok": False,
|
|
"message": "当前没有可推送的detect_result_projection",
|
|
"data": {},
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
mock_push_debug_event.assert_called_once()
|
|
self.assertEqual("detect_result_projection_sync_idle", mock_push_debug_event.call_args.kwargs["event_type"])
|
|
self.assertEqual("info", mock_push_debug_event.call_args.kwargs["level"])
|
|
|
|
@patch("app.sync_agent.push_debug_event")
|
|
def test_emit_structured_tick_marks_idle_sync_push_as_idle(self, mock_push_debug_event) -> None:
|
|
_emit_structured_tick(
|
|
base_event_type="sync_push",
|
|
ok=False,
|
|
message="当前没有需要立即推送的同步投影",
|
|
data={"results": []},
|
|
)
|
|
|
|
mock_push_debug_event.assert_called_once()
|
|
self.assertEqual("sync_push_idle", mock_push_debug_event.call_args.kwargs["event_type"])
|
|
self.assertEqual("info", mock_push_debug_event.call_args.kwargs["level"])
|
|
|
|
def test_filter_runtime_events_for_job_discards_mismatched_jobs(self) -> None:
|
|
events = _filter_runtime_events_for_job(
|
|
[
|
|
{
|
|
"job_id": 2137,
|
|
"node_code": "mainland-controller-01",
|
|
"event_type": "job_created",
|
|
"message": "new job",
|
|
"payload": {"source_record_id": 35461},
|
|
"created_at": "2026-04-22 01:21:29",
|
|
},
|
|
{
|
|
"job_id": 1902,
|
|
"node_code": "mainland-controller-01",
|
|
"event_type": "worker_log",
|
|
"message": "claim batch",
|
|
"payload": {"job_code": "sync-overseas-31437"},
|
|
"created_at": "2026-04-22 01:19:25",
|
|
},
|
|
],
|
|
job_code="sync-overseas-31437",
|
|
job_id=1902,
|
|
limit=8,
|
|
)
|
|
|
|
self.assertEqual(1, len(events))
|
|
self.assertEqual("worker_log", events[0]["event_type"])
|
|
|
|
@patch("app.sync_agent.push_debug_event")
|
|
@patch("app.sync_agent.process_detect_pipeline_now")
|
|
def test_run_pipeline_stage_processor_uses_high_default_limit(
|
|
self,
|
|
mock_process_detect_pipeline_now,
|
|
mock_push_debug_event,
|
|
) -> None:
|
|
mock_process_detect_pipeline_now.return_value = (
|
|
True,
|
|
"pipeline ok",
|
|
{"processed_items": 512},
|
|
)
|
|
|
|
ok, message, data = _run_pipeline_stage_processor()
|
|
|
|
self.assertTrue(ok)
|
|
self.assertEqual("pipeline ok", message)
|
|
self.assertEqual({"processed_items": 512}, data)
|
|
mock_process_detect_pipeline_now.assert_called_once_with(limit=5000)
|
|
self.assertEqual("pipeline_tick_success", mock_push_debug_event.call_args.kwargs["event_type"])
|
|
|
|
@patch("app.sync_agent._load_local_detect_backlog_snapshot", return_value={"pending_total": 12})
|
|
@patch("app.sync_agent.list_recent_detect_run_events", return_value=[])
|
|
@patch("app.sync_agent.get_detect_queue_health", return_value={"queue": {"overdue_leases": 0}})
|
|
@patch("app.sync_agent.push_debug_event")
|
|
@patch("app.sync_agent._append_detect_result_projection_snapshot")
|
|
@patch("app.sync_agent._select_projection_job_snapshots", return_value=[{"job_id": 1, "job_code": "finished-job"}])
|
|
@patch(
|
|
"app.sync_agent.get_active_detect_job_summary",
|
|
return_value={
|
|
"job_id": 2,
|
|
"job_code": "running-job",
|
|
"status": "running",
|
|
"items_total": 100,
|
|
"items_pending": 10,
|
|
"items_claimed": 5,
|
|
"items_running": 7,
|
|
"items_completed": 70,
|
|
"items_failed": 8,
|
|
"progress_percent": 70.0,
|
|
"node_stats": [{"node_code": "mainland-controller-01", "items_total": 90}],
|
|
},
|
|
)
|
|
def test_emit_runtime_debug_snapshots_emits_projection_and_active_job(
|
|
self,
|
|
mock_get_active_detect_job_summary,
|
|
mock_select_projection_job_snapshots,
|
|
mock_append_detect_result_projection_snapshot,
|
|
mock_push_debug_event,
|
|
mock_get_detect_queue_health,
|
|
mock_list_recent_detect_run_events,
|
|
mock_load_local_detect_backlog_snapshot,
|
|
) -> None:
|
|
_emit_runtime_debug_snapshots()
|
|
|
|
mock_get_active_detect_job_summary.assert_called_once_with(event_limit=10)
|
|
mock_select_projection_job_snapshots.assert_called_once_with()
|
|
mock_append_detect_result_projection_snapshot.assert_called_once()
|
|
self.assertTrue(mock_push_debug_event.called)
|
|
self.assertEqual("active_job_snapshot", mock_push_debug_event.call_args_list[0].kwargs["event_type"])
|
|
|
|
@patch("app.sync_agent._emit_runtime_debug_snapshots")
|
|
@patch("app.sync_agent._run_pipeline_stage_processor", return_value=(True, "pipeline ok", {"stage": "pipeline"}))
|
|
@patch("app.sync_agent.pull_detect_task_batch_now", return_value=(True, "pull ok", {"stage": "pull"}))
|
|
@patch("app.sync_agent.push_runtime_projection_now", return_value=(True, "sync ok", {"stage": "sync"}))
|
|
def test_run_sync_tick_once_prioritizes_sync_before_pipeline(
|
|
self,
|
|
mock_push_runtime_projection_now,
|
|
mock_pull_detect_task_batch_now,
|
|
mock_run_pipeline_stage_processor,
|
|
mock_emit_runtime_debug_snapshots,
|
|
) -> None:
|
|
with patch("app.sync_agent._maybe_trigger_overlap_start", return_value=(False, "no overlap", {})) as mock_overlap:
|
|
tick = _run_sync_tick_once()
|
|
|
|
self.assertTrue(tick["sync"]["ok"])
|
|
self.assertEqual("sync ok", tick["sync"]["message"])
|
|
mock_push_runtime_projection_now.assert_called_once_with()
|
|
mock_pull_detect_task_batch_now.assert_called_once_with()
|
|
mock_run_pipeline_stage_processor.assert_called_once_with()
|
|
mock_overlap.assert_called_once_with()
|
|
mock_emit_runtime_debug_snapshots.assert_called_once_with()
|
|
self.assertEqual("no overlap", tick["overlap"]["message"])
|
|
|
|
@patch("app.sync_agent.send_worker_command")
|
|
@patch("app.sync_agent._select_overlap_target_node_codes")
|
|
@patch("app.sync_agent._select_overlap_start_candidate")
|
|
@patch("app.sync_agent.push_debug_event")
|
|
def test_maybe_trigger_overlap_start_dispatches_pending_job(
|
|
self,
|
|
mock_push_debug_event,
|
|
mock_select_overlap_start_candidate,
|
|
mock_select_overlap_target_node_codes,
|
|
mock_send_worker_command,
|
|
) -> None:
|
|
mock_select_overlap_start_candidate.return_value = {
|
|
"job_id": 885,
|
|
"job_code": "sync-overseas-20592",
|
|
"task_mode": "domain_pipeline",
|
|
"items_pending": 64000,
|
|
"items_claimed": 0,
|
|
"items_running": 0,
|
|
"selection_reason": "overlap_tail_handoff",
|
|
}
|
|
mock_select_overlap_target_node_codes.return_value = [
|
|
"mainland-controller-01-a",
|
|
"mainland-controller-01-da",
|
|
]
|
|
mock_send_worker_command.return_value = (True, "已发送 Worker 控制指令: start_detection -> mainland-controller-01-a,mainland-controller-01-da")
|
|
|
|
with patch("app.sync_agent._LAST_OVERLAP_JOB_ID", 0), patch("app.sync_agent._LAST_OVERLAP_TRIGGERED_AT", 0.0):
|
|
ok, message, data = _maybe_trigger_overlap_start()
|
|
|
|
self.assertTrue(ok)
|
|
self.assertIn("mainland-controller-01-a", message)
|
|
self.assertEqual(885, data["job_id"])
|
|
mock_send_worker_command.assert_called_once_with(
|
|
"start_detection",
|
|
payload={
|
|
"job_id": 885,
|
|
"job_code": "sync-overseas-20592",
|
|
"target_job_id": 885,
|
|
"target_job_code": "sync-overseas-20592",
|
|
"task_mode": "domain_pipeline",
|
|
"source": "overlap-handoff",
|
|
"selection_reason": "overlap_tail_handoff",
|
|
"tail_handoff_candidate": True,
|
|
"target_node_codes": [
|
|
"mainland-controller-01-a",
|
|
"mainland-controller-01-da",
|
|
],
|
|
},
|
|
)
|
|
self.assertEqual("overlap_handoff_started", mock_push_debug_event.call_args.kwargs["event_type"])
|
|
|
|
@patch("app.sync_agent._select_overlap_start_candidate")
|
|
def test_maybe_trigger_overlap_start_honors_cooldown_for_same_job(
|
|
self,
|
|
mock_select_overlap_start_candidate,
|
|
) -> None:
|
|
mock_select_overlap_start_candidate.return_value = {
|
|
"job_id": 885,
|
|
"job_code": "sync-overseas-20592",
|
|
"task_mode": "domain_pipeline",
|
|
"items_pending": 64000,
|
|
"selection_reason": "overlap_tail_handoff",
|
|
}
|
|
|
|
with patch("app.sync_agent._LAST_OVERLAP_JOB_ID", 885), patch("app.sync_agent._LAST_OVERLAP_TRIGGERED_AT", __import__('time').time()):
|
|
ok, message, data = _maybe_trigger_overlap_start()
|
|
|
|
self.assertFalse(ok)
|
|
self.assertIn("冷却中", message)
|
|
self.assertEqual(885, data["job_id"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|