861 lines
32 KiB
Python
861 lines
32 KiB
Python
import json
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import patch
|
|
|
|
from app.services.sync_record_service import (
|
|
_build_runtime_projection_payload,
|
|
_collect_recent_domain_events,
|
|
_pick_latest_projection_row,
|
|
append_runtime_projection_if_changed,
|
|
get_detect_result_sync_batches,
|
|
)
|
|
|
|
|
|
class _FakeCursor:
|
|
def __init__(self, fetchone_values):
|
|
self.fetchone_values = list(fetchone_values or [])
|
|
self.executed = []
|
|
|
|
def execute(self, sql, params=None):
|
|
self.executed.append((sql, params))
|
|
|
|
def fetchone(self):
|
|
if self.fetchone_values:
|
|
return self.fetchone_values.pop(0)
|
|
return None
|
|
|
|
def fetchall(self):
|
|
if self.fetchone_values:
|
|
value = self.fetchone_values.pop(0)
|
|
return list(value or [])
|
|
return []
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
|
|
class _FakeConnection:
|
|
def __init__(self, fetchone_values):
|
|
self.cursor_obj = _FakeCursor(fetchone_values)
|
|
self.committed = False
|
|
|
|
def cursor(self):
|
|
return self.cursor_obj
|
|
|
|
def commit(self):
|
|
self.committed = True
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
|
|
class SyncRecordServiceTests(unittest.TestCase):
|
|
def test_collect_recent_domain_events_filters_and_keeps_latest_slice(self) -> None:
|
|
active_job = {
|
|
"current_cycle_events": [
|
|
{
|
|
"node_code": "mainland-worker-01",
|
|
"event_type": "job_dispatch_sent",
|
|
"level": "info",
|
|
"message": "dispatch",
|
|
"created_at": "2026-04-19 01:00:05",
|
|
"payload": {"cycle_token": "cycle-1"},
|
|
},
|
|
{
|
|
"node_code": "mainland-worker-01",
|
|
"event_type": "domain_completed",
|
|
"level": "info",
|
|
"message": "完成 a.com",
|
|
"created_at": "2026-04-19 01:00:04",
|
|
"payload": {"domain": "a.com", "cycle_token": "cycle-1"},
|
|
},
|
|
{
|
|
"node_code": "mainland-worker-01",
|
|
"event_type": "domain_started",
|
|
"level": "info",
|
|
"message": "开始 b.com",
|
|
"created_at": "2026-04-19 01:00:03",
|
|
"payload": {"domain": "b.com", "cycle_token": "cycle-1"},
|
|
},
|
|
{
|
|
"node_code": "mainland-worker-01",
|
|
"event_type": "domain_failed",
|
|
"level": "warning",
|
|
"message": "失败 c.com",
|
|
"created_at": "2026-04-19 01:00:02",
|
|
"payload": {"domain": "c.com", "cycle_token": "cycle-1"},
|
|
},
|
|
]
|
|
}
|
|
|
|
events = _collect_recent_domain_events(active_job, limit=2)
|
|
|
|
self.assertEqual(2, len(events))
|
|
self.assertEqual("domain_started", events[0]["event_type"])
|
|
self.assertEqual("b.com", events[0]["payload"]["domain"])
|
|
self.assertEqual("domain_completed", events[1]["event_type"])
|
|
self.assertEqual("a.com", events[1]["payload"]["domain"])
|
|
|
|
@patch("app.services.sync_record_service._resolve_local_ip", return_value="152.53.37.118")
|
|
@patch("app.services.sync_record_service.socket.gethostname", return_value="overseas-control-01")
|
|
@patch("app.services.sync_record_service.settings")
|
|
def test_build_runtime_projection_payload_zeros_execution_for_overseas_control(
|
|
self,
|
|
mock_settings,
|
|
_mock_hostname,
|
|
_mock_resolve_ip,
|
|
) -> None:
|
|
mock_settings.node_code = "overseas-control-01"
|
|
mock_settings.node_region = "overseas"
|
|
mock_settings.node_role = "control"
|
|
mock_settings.sync_source_region = "overseas"
|
|
mock_settings.sync_target_region = "mainland"
|
|
|
|
payload = _build_runtime_projection_payload(
|
|
detect={
|
|
"worker_online": False,
|
|
"worker_mode": "linux-systemd",
|
|
"active_thread_count": 0,
|
|
"max_thread_count": 0,
|
|
"phase_label": "当前节点不承载",
|
|
"phase_detail": "当前节点为海外控制面，仅承载 API 控制与同步接收，不执行本机检测任务。",
|
|
"proxy_runtime_label": "不适用",
|
|
"proxy_runtime_reason": "not_applicable",
|
|
"progress": {
|
|
"pending": 125,
|
|
"running": 293,
|
|
"completed": 71,
|
|
"blacklisted": 0,
|
|
"failed": 1,
|
|
},
|
|
"active_job": {
|
|
"job_id": 2027,
|
|
"job_code": "sync-overseas-33501",
|
|
"status": "running",
|
|
"progress_percent": 7.2,
|
|
"items_total": 1000,
|
|
"items_terminal": 72,
|
|
"items_pending": 125,
|
|
"items_running": 293,
|
|
"items_failed": 1,
|
|
"node_stats": [{"node_code": "mainland-controller-01", "items_running": 293}],
|
|
},
|
|
},
|
|
cluster={
|
|
"nodes_total": 3,
|
|
"nodes": [
|
|
{
|
|
"node_code": "overseas-control-01",
|
|
"current_load": 293,
|
|
"detect_participating": True,
|
|
}
|
|
],
|
|
"summary": {
|
|
"online_worker_nodes": 3,
|
|
"dedicated_online_worker_nodes": 1,
|
|
"online_control_nodes": 2,
|
|
"busy_nodes": ["overseas-control-01"],
|
|
"stale_nodes": [],
|
|
"offline_nodes": [],
|
|
},
|
|
},
|
|
source_region="overseas",
|
|
target_region="mainland",
|
|
)
|
|
|
|
projection = payload["projection"]
|
|
self.assertFalse(projection["worker_online"])
|
|
self.assertFalse(projection["detect_participating"])
|
|
self.assertEqual(0, projection["active_thread_count"])
|
|
self.assertEqual(0, projection["max_thread_count"])
|
|
self.assertEqual(0, projection["progress"]["running"])
|
|
self.assertEqual("", projection["active_job"]["job_code"])
|
|
self.assertEqual([], projection["active_job"]["node_stats"])
|
|
|
|
@patch("app.services.sync_record_service._resolve_local_ip", return_value="121.204.244.188")
|
|
@patch("app.services.sync_record_service.socket.gethostname", return_value="mainland-controller-01")
|
|
@patch("app.services.sync_record_service.settings")
|
|
def test_build_runtime_projection_payload_prefers_runtime_queue_nodes_for_multi_process_controller(
|
|
self,
|
|
mock_settings,
|
|
_mock_hostname,
|
|
_mock_resolve_ip,
|
|
) -> None:
|
|
mock_settings.node_code = "mainland-controller-01"
|
|
mock_settings.node_region = "mainland"
|
|
mock_settings.node_role = "control"
|
|
mock_settings.sync_source_region = "mainland"
|
|
mock_settings.sync_target_region = "overseas"
|
|
|
|
payload = _build_runtime_projection_payload(
|
|
detect={
|
|
"worker_online": True,
|
|
"worker_mode": "linux-systemd",
|
|
"active_thread_count": 1,
|
|
"max_thread_count": 1000,
|
|
"aggregate_max_thread_count": 80000,
|
|
"phase_label": "运行中",
|
|
"phase_detail": "80 实例运行",
|
|
"proxy_runtime_label": "正常",
|
|
"proxy_runtime_reason": "healthy",
|
|
"progress": {
|
|
"pending": 5001,
|
|
"running": 0,
|
|
"completed": 0,
|
|
"blacklisted": 0,
|
|
"failed": 0,
|
|
},
|
|
"backlog": {"pending_total": 1157325},
|
|
"queue_health": {
|
|
"queue": {
|
|
"items_total": 5292,
|
|
"pending": 4578,
|
|
"claimed": 0,
|
|
"display_claimed": 120,
|
|
"running": 1972,
|
|
"display_running": 1972,
|
|
"completed": 690,
|
|
"blacklisted": 0,
|
|
"failed": 24,
|
|
"terminal": 714,
|
|
},
|
|
"nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01-a",
|
|
"items_running": 1000,
|
|
"items_claimed": 0,
|
|
"display_running": 1000,
|
|
"active_threads": 1000,
|
|
"max_threads": 1000,
|
|
"region": "mainland",
|
|
"role": "control",
|
|
"status": "busy",
|
|
},
|
|
{
|
|
"node_code": "mainland-controller-01-b",
|
|
"items_running": 972,
|
|
"items_claimed": 120,
|
|
"display_running": 972,
|
|
"active_threads": 972,
|
|
"max_threads": 1000,
|
|
"region": "mainland",
|
|
"role": "control",
|
|
"status": "busy",
|
|
},
|
|
],
|
|
},
|
|
"active_job": {
|
|
"job_id": 55,
|
|
"job_code": "sync-overseas-55",
|
|
"status": "running",
|
|
"progress_percent": 13.49,
|
|
"items_total": 5292,
|
|
"items_terminal": 714,
|
|
"items_pending": 4578,
|
|
"items_running": 0,
|
|
"items_failed": 24,
|
|
"node_stats": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"items_running": 1,
|
|
"items_claimed": 0,
|
|
"items_total": 5292,
|
|
}
|
|
],
|
|
},
|
|
"dependency_alerts": [],
|
|
},
|
|
cluster={
|
|
"nodes_total": 80,
|
|
"nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"current_load": 1,
|
|
"detect_participating": True,
|
|
}
|
|
],
|
|
"summary": {
|
|
"online_worker_nodes": 65,
|
|
"dedicated_online_worker_nodes": 1,
|
|
"online_control_nodes": 65,
|
|
"busy_nodes": ["mainland-controller-01-a", "mainland-controller-01-b"],
|
|
"stale_nodes": [],
|
|
"offline_nodes": [],
|
|
},
|
|
},
|
|
source_region="mainland",
|
|
target_region="overseas",
|
|
)
|
|
|
|
projection = payload["projection"]
|
|
self.assertTrue(projection["detect_participating"])
|
|
self.assertEqual(1972, projection["active_thread_count"])
|
|
self.assertEqual(80000, projection["max_thread_count"])
|
|
self.assertEqual(1972, projection["progress"]["running"])
|
|
self.assertEqual(2, len(projection["active_job"]["node_stats"]))
|
|
self.assertEqual(2, len(projection["active_job"]["distributed_node_stats"]))
|
|
self.assertEqual(1, len(projection["cluster_nodes"]))
|
|
self.assertEqual("mainland-controller-01", projection["cluster_nodes"][0]["node_code"])
|
|
self.assertEqual(1972, projection["active_job"]["items_running"])
|
|
self.assertEqual(1972, projection["active_job"]["display_items_running"])
|
|
self.assertEqual(80000, projection["active_job"]["display_max_threads"])
|
|
|
|
@patch("app.services.sync_record_service._resolve_local_ip", return_value="121.204.244.188")
|
|
@patch("app.services.sync_record_service.socket.gethostname", return_value="mainland-controller-01")
|
|
@patch("app.services.sync_record_service.settings")
|
|
@patch("app.services.sync_record_service.get_db")
|
|
def test_append_runtime_projection_if_changed_writes_heartbeat_for_unchanged_projection_after_interval(
|
|
self,
|
|
mock_get_db,
|
|
mock_settings,
|
|
_mock_hostname,
|
|
_mock_resolve_ip,
|
|
) -> None:
|
|
mock_settings.node_code = "mainland-controller-01"
|
|
mock_settings.node_region = "mainland"
|
|
mock_settings.node_role = "control"
|
|
mock_settings.sync_source_region = "mainland"
|
|
mock_settings.sync_target_region = "overseas"
|
|
|
|
detect = {
|
|
"worker_online": True,
|
|
"worker_mode": "linux-systemd",
|
|
"active_thread_count": 3200,
|
|
"max_thread_count": 80000,
|
|
"phase_label": "运行中",
|
|
"phase_detail": "80 实例运行",
|
|
"proxy_runtime_label": "正常",
|
|
"proxy_runtime_reason": "healthy",
|
|
"progress": {
|
|
"pending": 1200000,
|
|
"running": 6400,
|
|
"completed": 50000,
|
|
"blacklisted": 1200,
|
|
"failed": 88,
|
|
},
|
|
"backlog": {"pending_total": 1200000},
|
|
"active_job": {
|
|
"job_id": 55,
|
|
"job_code": "sync-overseas-55",
|
|
"status": "running",
|
|
"progress_percent": 12.5,
|
|
"items_total": 1300000,
|
|
"items_terminal": 51288,
|
|
"items_pending": 1200000,
|
|
"items_running": 6400,
|
|
"items_failed": 88,
|
|
"node_stats": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"items_running": 6400,
|
|
"items_claimed": 7000,
|
|
"items_total": 1300000,
|
|
}
|
|
],
|
|
},
|
|
"dependency_alerts": [],
|
|
}
|
|
cluster = {
|
|
"nodes_total": 1,
|
|
"nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"current_load": 6400,
|
|
"detect_participating": True,
|
|
}
|
|
],
|
|
"summary": {
|
|
"online_worker_nodes": 1,
|
|
"dedicated_online_worker_nodes": 0,
|
|
"online_control_nodes": 1,
|
|
"busy_nodes": ["mainland-controller-01"],
|
|
"stale_nodes": [],
|
|
"offline_nodes": [],
|
|
},
|
|
}
|
|
|
|
previous_payload = _build_runtime_projection_payload(
|
|
detect=detect,
|
|
cluster=cluster,
|
|
source_region="mainland",
|
|
target_region="overseas",
|
|
)
|
|
fake_conn = _FakeConnection(
|
|
[
|
|
[(previous_payload, datetime.now() - timedelta(seconds=90))],
|
|
(321,),
|
|
]
|
|
)
|
|
mock_get_db.return_value = fake_conn
|
|
|
|
record_id = append_runtime_projection_if_changed(
|
|
detect=detect,
|
|
cluster=cluster,
|
|
source_region="mainland",
|
|
target_region="overseas",
|
|
)
|
|
|
|
self.assertEqual(321, record_id)
|
|
self.assertTrue(fake_conn.committed)
|
|
self.assertTrue(
|
|
any("INSERT INTO detect_sync_records" in sql for sql, _params in fake_conn.cursor_obj.executed)
|
|
)
|
|
|
|
@patch("app.services.sync_record_service._resolve_local_ip", return_value="121.204.244.188")
|
|
@patch("app.services.sync_record_service.socket.gethostname", return_value="mainland-controller-01")
|
|
@patch("app.services.sync_record_service.settings")
|
|
@patch("app.services.sync_record_service.get_db")
|
|
def test_append_runtime_projection_if_changed_writes_when_cluster_nodes_change_within_window(
|
|
self,
|
|
mock_get_db,
|
|
mock_settings,
|
|
_mock_hostname,
|
|
_mock_resolve_ip,
|
|
) -> None:
|
|
mock_settings.node_code = "mainland-controller-01"
|
|
mock_settings.node_region = "mainland"
|
|
mock_settings.node_role = "control"
|
|
mock_settings.sync_source_region = "mainland"
|
|
mock_settings.sync_target_region = "overseas"
|
|
|
|
previous_payload = {
|
|
"projection": {
|
|
"worker_online": True,
|
|
"worker_mode": "linux-systemd",
|
|
"phase_label": "运行中",
|
|
"phase_detail": "等待中",
|
|
"proxy_runtime_label": "正常",
|
|
"proxy_runtime_reason": "healthy",
|
|
"active_thread_count": 0,
|
|
"max_thread_count": 60000,
|
|
"progress": {"pending": 10000, "running": 0, "completed": 0, "blacklisted": 0, "failed": 0},
|
|
"active_job": {
|
|
"job_id": None,
|
|
"job_code": "",
|
|
"status": "",
|
|
"items_total": 0,
|
|
"items_running": 0,
|
|
"items_claimed": 0,
|
|
"display_items_running": 0,
|
|
"display_items_claimed": 0,
|
|
"display_max_threads": 0,
|
|
"node_stats": [],
|
|
"distributed_node_stats": [],
|
|
},
|
|
"cluster_summary": {
|
|
"nodes_total": 43,
|
|
"online_worker_nodes": 43,
|
|
"dedicated_online_worker_nodes": 42,
|
|
"online_control_nodes": 1,
|
|
"busy_nodes": [],
|
|
"stale_nodes": [],
|
|
"offline_nodes": [],
|
|
},
|
|
"cluster_nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"role": "control",
|
|
"status": "online",
|
|
"current_load": 0,
|
|
"active_threads": 0,
|
|
"max_threads": 60000,
|
|
"detect_participating": False,
|
|
}
|
|
],
|
|
"dependency_alerts": [],
|
|
}
|
|
}
|
|
fake_conn = _FakeConnection(
|
|
[
|
|
[(previous_payload, datetime.now() - timedelta(seconds=10))],
|
|
(911,),
|
|
]
|
|
)
|
|
mock_get_db.return_value = fake_conn
|
|
|
|
record_id = append_runtime_projection_if_changed(
|
|
detect={
|
|
"worker_online": True,
|
|
"worker_mode": "linux-systemd",
|
|
"phase_label": "运行中",
|
|
"phase_detail": "等待中",
|
|
"proxy_runtime_label": "正常",
|
|
"proxy_runtime_reason": "healthy",
|
|
"progress": {"pending": 10000, "running": 0, "completed": 0, "blacklisted": 0, "failed": 0},
|
|
"backlog": {},
|
|
"active_job": {},
|
|
"dependency_alerts": [],
|
|
},
|
|
cluster={
|
|
"nodes_total": 61,
|
|
"nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"role": "control",
|
|
"status": "online",
|
|
"current_load": 0,
|
|
"metadata": {"active_threads": 0, "max_threads": 60000},
|
|
},
|
|
{
|
|
"node_code": "mainland-controller-01-a",
|
|
"role": "worker",
|
|
"status": "online",
|
|
"current_load": 0,
|
|
"metadata": {"active_threads": 0, "max_threads": 1000},
|
|
},
|
|
],
|
|
"summary": {
|
|
"online_worker_nodes": 61,
|
|
"dedicated_online_worker_nodes": 60,
|
|
"online_control_nodes": 1,
|
|
"busy_nodes": [],
|
|
"stale_nodes": [],
|
|
"offline_nodes": [],
|
|
},
|
|
},
|
|
source_region="mainland",
|
|
target_region="overseas",
|
|
)
|
|
|
|
self.assertEqual(911, record_id)
|
|
self.assertTrue(fake_conn.committed)
|
|
self.assertTrue(
|
|
any("INSERT INTO detect_sync_records" in sql for sql, _params in fake_conn.cursor_obj.executed)
|
|
)
|
|
|
|
@patch("app.services.sync_record_service._resolve_local_ip", return_value="121.204.244.188")
|
|
@patch("app.services.sync_record_service.socket.gethostname", return_value="mainland-controller-01")
|
|
@patch("app.services.sync_record_service.settings")
|
|
@patch("app.services.sync_record_service.get_db")
|
|
def test_append_runtime_projection_if_changed_skips_unchanged_projection_within_heartbeat_window(
|
|
self,
|
|
mock_get_db,
|
|
mock_settings,
|
|
_mock_hostname,
|
|
_mock_resolve_ip,
|
|
) -> None:
|
|
mock_settings.node_code = "mainland-controller-01"
|
|
mock_settings.node_region = "mainland"
|
|
mock_settings.node_role = "control"
|
|
mock_settings.sync_source_region = "mainland"
|
|
mock_settings.sync_target_region = "overseas"
|
|
|
|
detect = {
|
|
"worker_online": True,
|
|
"worker_mode": "linux-systemd",
|
|
"active_thread_count": 3200,
|
|
"max_thread_count": 80000,
|
|
"phase_label": "运行中",
|
|
"phase_detail": "80 实例运行",
|
|
"proxy_runtime_label": "正常",
|
|
"proxy_runtime_reason": "healthy",
|
|
"progress": {
|
|
"pending": 1200000,
|
|
"running": 6400,
|
|
"completed": 50000,
|
|
"blacklisted": 1200,
|
|
"failed": 88,
|
|
},
|
|
"backlog": {"pending_total": 1200000},
|
|
"active_job": {
|
|
"job_id": 55,
|
|
"job_code": "sync-overseas-55",
|
|
"status": "running",
|
|
"progress_percent": 12.5,
|
|
"items_total": 1300000,
|
|
"items_terminal": 51288,
|
|
"items_pending": 1200000,
|
|
"items_running": 6400,
|
|
"items_failed": 88,
|
|
"node_stats": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"items_running": 6400,
|
|
"items_claimed": 7000,
|
|
"items_total": 1300000,
|
|
}
|
|
],
|
|
},
|
|
"dependency_alerts": [],
|
|
}
|
|
cluster = {
|
|
"nodes_total": 1,
|
|
"nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"current_load": 6400,
|
|
"detect_participating": True,
|
|
}
|
|
],
|
|
"summary": {
|
|
"online_worker_nodes": 1,
|
|
"dedicated_online_worker_nodes": 0,
|
|
"online_control_nodes": 1,
|
|
"busy_nodes": ["mainland-controller-01"],
|
|
"stale_nodes": [],
|
|
"offline_nodes": [],
|
|
},
|
|
}
|
|
|
|
previous_payload = _build_runtime_projection_payload(
|
|
detect=detect,
|
|
cluster=cluster,
|
|
source_region="mainland",
|
|
target_region="overseas",
|
|
)
|
|
fake_conn = _FakeConnection(
|
|
[
|
|
[(previous_payload, datetime.now() - timedelta(seconds=10))],
|
|
]
|
|
)
|
|
mock_get_db.return_value = fake_conn
|
|
|
|
record_id = append_runtime_projection_if_changed(
|
|
detect=detect,
|
|
cluster=cluster,
|
|
source_region="mainland",
|
|
target_region="overseas",
|
|
)
|
|
|
|
self.assertIsNone(record_id)
|
|
self.assertFalse(fake_conn.committed)
|
|
self.assertFalse(
|
|
any("INSERT INTO detect_sync_records" in sql for sql, _params in fake_conn.cursor_obj.executed)
|
|
)
|
|
|
|
@patch("app.services.sync_record_service._resolve_local_ip", return_value="121.204.244.188")
|
|
@patch("app.services.sync_record_service.socket.gethostname", return_value="mainland-controller-01")
|
|
@patch("app.services.sync_record_service.settings")
|
|
@patch("app.services.sync_record_service.get_db")
|
|
def test_append_runtime_projection_if_changed_writes_when_activity_signature_changes_within_window(
|
|
self,
|
|
mock_get_db,
|
|
mock_settings,
|
|
_mock_hostname,
|
|
_mock_resolve_ip,
|
|
) -> None:
|
|
mock_settings.node_code = "mainland-controller-01"
|
|
mock_settings.node_region = "mainland"
|
|
mock_settings.node_role = "control"
|
|
mock_settings.sync_source_region = "mainland"
|
|
mock_settings.sync_target_region = "overseas"
|
|
|
|
previous_detect = {
|
|
"worker_online": True,
|
|
"worker_mode": "linux-systemd",
|
|
"active_thread_count": 300,
|
|
"max_thread_count": 60000,
|
|
"phase_label": "运行中",
|
|
"phase_detail": "60 实例运行",
|
|
"proxy_runtime_label": "正常",
|
|
"proxy_runtime_reason": "healthy",
|
|
"progress": {
|
|
"pending": 5000,
|
|
"running": 600,
|
|
"completed": 1000,
|
|
"blacklisted": 0,
|
|
"failed": 10,
|
|
},
|
|
"queue_health": {
|
|
"queue": {
|
|
"items_total": 6610,
|
|
"pending": 5000,
|
|
"claimed": 10,
|
|
"display_claimed": 10,
|
|
"running": 600,
|
|
"display_running": 600,
|
|
"completed": 1000,
|
|
"blacklisted": 0,
|
|
"failed": 10,
|
|
"terminal": 1010,
|
|
"display_max_threads": 60000,
|
|
},
|
|
"nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01-a",
|
|
"items_running": 300,
|
|
"items_claimed": 10,
|
|
"display_running": 300,
|
|
"active_threads": 300,
|
|
"max_threads": 1000,
|
|
"region": "mainland",
|
|
"role": "control",
|
|
"status": "busy",
|
|
}
|
|
],
|
|
},
|
|
"active_job": {
|
|
"job_id": 867,
|
|
"job_code": "sync-overseas-19835",
|
|
"status": "running",
|
|
"progress_percent": 15.0,
|
|
"items_total": 6610,
|
|
"items_terminal": 1010,
|
|
"items_pending": 5000,
|
|
"items_claimed": 10,
|
|
"items_running": 600,
|
|
"items_failed": 10,
|
|
"node_stats": [
|
|
{
|
|
"node_code": "mainland-controller-01-a",
|
|
"items_running": 300,
|
|
"items_claimed": 10,
|
|
"items_total": 6610,
|
|
}
|
|
],
|
|
},
|
|
"dependency_alerts": [],
|
|
}
|
|
current_detect = {
|
|
**previous_detect,
|
|
"active_thread_count": 900,
|
|
"queue_health": {
|
|
"queue": {
|
|
"items_total": 6610,
|
|
"pending": 4300,
|
|
"claimed": 30,
|
|
"display_claimed": 30,
|
|
"running": 900,
|
|
"display_running": 900,
|
|
"completed": 1370,
|
|
"blacklisted": 0,
|
|
"failed": 10,
|
|
"terminal": 1380,
|
|
"display_max_threads": 60000,
|
|
},
|
|
"nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01-a",
|
|
"items_running": 420,
|
|
"items_claimed": 20,
|
|
"display_running": 420,
|
|
"active_threads": 420,
|
|
"max_threads": 1000,
|
|
"region": "mainland",
|
|
"role": "control",
|
|
"status": "busy",
|
|
},
|
|
{
|
|
"node_code": "mainland-controller-01-b",
|
|
"items_running": 480,
|
|
"items_claimed": 10,
|
|
"display_running": 480,
|
|
"active_threads": 480,
|
|
"max_threads": 1000,
|
|
"region": "mainland",
|
|
"role": "control",
|
|
"status": "busy",
|
|
},
|
|
],
|
|
},
|
|
"active_job": {
|
|
**previous_detect["active_job"],
|
|
"items_pending": 4300,
|
|
"items_claimed": 30,
|
|
"items_running": 900,
|
|
"node_stats": [
|
|
{
|
|
"node_code": "mainland-controller-01-a",
|
|
"items_running": 420,
|
|
"items_claimed": 20,
|
|
"items_total": 6610,
|
|
},
|
|
{
|
|
"node_code": "mainland-controller-01-b",
|
|
"items_running": 480,
|
|
"items_claimed": 10,
|
|
"items_total": 6610,
|
|
},
|
|
],
|
|
},
|
|
}
|
|
cluster = {
|
|
"nodes_total": 2,
|
|
"nodes": [
|
|
{
|
|
"node_code": "mainland-controller-01",
|
|
"current_load": 900,
|
|
"detect_participating": True,
|
|
}
|
|
],
|
|
"summary": {
|
|
"online_worker_nodes": 60,
|
|
"dedicated_online_worker_nodes": 0,
|
|
"online_control_nodes": 60,
|
|
"busy_nodes": ["mainland-controller-01-a", "mainland-controller-01-b"],
|
|
"stale_nodes": [],
|
|
"offline_nodes": [],
|
|
},
|
|
}
|
|
|
|
previous_payload = _build_runtime_projection_payload(
|
|
detect=previous_detect,
|
|
cluster=cluster,
|
|
source_region="mainland",
|
|
target_region="overseas",
|
|
)
|
|
fake_conn = _FakeConnection(
|
|
[
|
|
[(previous_payload, datetime.now() - timedelta(seconds=10))],
|
|
(654,),
|
|
]
|
|
)
|
|
mock_get_db.return_value = fake_conn
|
|
|
|
record_id = append_runtime_projection_if_changed(
|
|
detect=current_detect,
|
|
cluster=cluster,
|
|
source_region="mainland",
|
|
target_region="overseas",
|
|
)
|
|
|
|
self.assertEqual(654, record_id)
|
|
self.assertTrue(fake_conn.committed)
|
|
self.assertTrue(
|
|
any("INSERT INTO detect_sync_records" in sql for sql, _params in fake_conn.cursor_obj.executed)
|
|
)
|
|
|
|
def test_pick_latest_projection_row_skips_future_dated_rows(self) -> None:
|
|
rows = [
|
|
("future", datetime.now() + timedelta(hours=6)),
|
|
("recent", datetime.now() - timedelta(seconds=10)),
|
|
("older", datetime.now() - timedelta(minutes=2)),
|
|
]
|
|
|
|
selected = _pick_latest_projection_row(rows, created_at_index=1)
|
|
|
|
self.assertEqual("recent", selected[0])
|
|
|
|
@patch("app.services.sync_record_service.get_db")
|
|
@patch("app.services.sync_record_service.settings")
|
|
def test_get_detect_result_sync_batches_marks_overseas_control_as_not_applicable(
|
|
self,
|
|
mock_settings,
|
|
mock_get_db,
|
|
) -> None:
|
|
mock_settings.node_code = "overseas-control-01"
|
|
mock_settings.node_region = "overseas"
|
|
mock_settings.node_role = "control"
|
|
mock_settings.sync_source_region = "overseas"
|
|
mock_settings.sync_target_region = "mainland"
|
|
|
|
payload = get_detect_result_sync_batches(limit=5)
|
|
|
|
self.assertFalse(payload["applicable"])
|
|
self.assertFalse(payload["local_worker_expected"])
|
|
self.assertEqual(0, payload["jobs_total"])
|
|
self.assertEqual([], payload["batches"])
|
|
self.assertIn("不适用", payload["reason"])
|
|
mock_get_db.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|