import unittest from datetime import datetime, timedelta from unittest.mock import MagicMock, patch from psycopg2 import errors import app.services.detect_job_service as detect_job_service_module from app.services.detect_job_service import ( append_detect_job_event, _build_active_job_summary_from_runtime_snapshot, _build_step_payload, _classify_pipeline_item_outcome, _classify_runtime_debug_event, _build_runtime_snapshot_from_projection, _enrich_active_job_summary_with_runtime, _build_display_summary, _build_runtime_display_bucket, _load_latest_runtime_active_job_snapshot, get_active_detect_job_summary, normalize_detect_step_code, get_detect_queue_health, process_detect_pipeline_now, resolve_initial_domain_pipeline_item, resolve_domain_pipeline_step, resolve_detect_job_definition, ) class _SequenceCursor: def __init__(self, *, fetchone_results=None, fetchall_results=None) -> None: self.fetchone_results = list(fetchone_results or []) self.fetchall_results = list(fetchall_results or []) self.exec_calls = [] def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def execute(self, sql, params=None): self.exec_calls.append((sql, params)) def fetchone(self): if self.fetchone_results: return self.fetchone_results.pop(0) return None def fetchall(self): if self.fetchall_results: return self.fetchall_results.pop(0) return [] class _SequenceConn: def __init__(self, *, fetchone_results=None, fetchall_results=None) -> None: self.cursor_obj = _SequenceCursor(fetchone_results=fetchone_results, fetchall_results=fetchall_results) def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def cursor(self): return self.cursor_obj class DetectJobServiceTests(unittest.TestCase): def test_load_runtime_display_rows_excludes_disabled_managed_nodes(self) -> None: cursor = MagicMock() cursor.fetchall.return_value = [ ("mainland-controller-01", "mainland", "control", "busy", 1, {}, datetime.now()), ("mainland-worker-01", "mainland", "worker", "busy", 1, {}, datetime.now()), ] with patch("app.services.detect_job_service._load_disabled_managed_node_codes", return_value={"mainland-worker-01"}): rows = detect_job_service_module._load_runtime_display_rows(cursor) self.assertEqual(["mainland-controller-01"], [row[0] for row in rows]) def test_build_active_job_summary_from_runtime_snapshot_keeps_non_zero_display_when_queue_display_is_zero(self) -> None: with patch("app.services.detect_job_service.settings.node_region", "mainland"): summary = _build_active_job_summary_from_runtime_snapshot( { "job": { "job_id": 60, "job_code": "sync-overseas-255", "status": "running", "items_total": 8402, "items_pending": 7865, "items_claimed": 0, "items_running": 537, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "items_terminal": 0, "display_items_running": 0, "display_active_threads": 0, "display_max_threads": 0, "node_stats": [ { "node_code": "mainland-controller-01-a", "items_running": 0, "display_running": 0, "active_threads": 0, "max_threads": 0, } ], }, "queue_health": { "queue": { "items_total": 8402, "pending": 7865, "claimed": 0, "running": 537, "display_running": 0, "completed": 0, "blacklisted": 0, "failed": 0, "terminal": 0, }, "nodes": [ { "node_code": "mainland-controller-01-a", "items_running": 0, "display_running": 0, "active_threads": 0, "max_threads": 0, } ], "steps": [], "throughput": {}, }, "_snapshot_source": "runtime_ingest", } ) assert summary is not None self.assertEqual(537, summary["display_items_running"]) self.assertEqual(537, summary["display_active_threads"]) self.assertEqual(537, summary["display_current_load"]) def test_build_active_job_summary_from_runtime_snapshot_drops_stale_runtime_nodes_from_live_display(self) -> None: stale_heartbeat = (datetime.now() - timedelta(minutes=8)).isoformat(sep=" ", timespec="seconds") snapshot_created_at = datetime.now().isoformat(sep=" ", timespec="seconds") summary = _build_active_job_summary_from_runtime_snapshot( { "job": { "job_id": 61, "job_code": "sync-overseas-256", "status": "running", "items_total": 8402, "items_pending": 7865, "items_claimed": 0, "items_running": 537, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "items_terminal": 0, "display_items_running": 537, "display_active_threads": 537, "node_stats": [ { "node_code": "mainland-controller-01-a", "items_running": 537, "display_running": 537, "active_threads": 537, "max_threads": 1000, "status": "busy", "last_heartbeat_at": stale_heartbeat, } ], }, "queue_health": { "queue": { "items_total": 8402, "pending": 7865, "claimed": 0, "running": 537, "display_running": 537, "completed": 0, "blacklisted": 0, "failed": 0, "terminal": 0, }, "nodes": [ { "node_code": "mainland-controller-01-a", "items_running": 537, "display_running": 537, "active_threads": 537, "max_threads": 1000, "status": "busy", "last_heartbeat_at": stale_heartbeat, } ], "steps": [], "throughput": {}, }, "_snapshot_source": "runtime_ingest", "_created_at": snapshot_created_at, } ) assert summary is not None self.assertEqual(537, summary["items_running"]) self.assertEqual(0, summary["display_items_running"]) self.assertEqual(0, summary["display_active_threads"]) self.assertEqual([], summary["node_stats"]) def test_build_active_job_summary_from_runtime_snapshot_does_not_promote_unassigned_runtime_load(self) -> None: with patch("app.services.detect_job_service.settings.node_region", "mainland"): summary = _build_active_job_summary_from_runtime_snapshot( { "job": { "job_id": 62, "job_code": "sync-overseas-257", "status": "running", "items_total": 4000, "items_pending": 0, "items_claimed": 23, "items_running": 2100, "items_completed": 1117, "items_blacklisted": 0, "items_failed": 760, "items_terminal": 1877, "display_items_running": 0, "display_active_threads": 0, "display_max_threads": 0, "node_stats": [ { "node_code": "unassigned", "items_claimed": 23, "items_running": 1863, "display_running": 1863, "active_threads": 0, "max_threads": 0, } ], }, "queue_health": { "queue": { "items_total": 4000, "pending": 0, "claimed": 23, "running": 2100, "display_running": 2100, "completed": 1117, "blacklisted": 0, "failed": 760, "terminal": 1877, }, "nodes": [ { "node_code": "unassigned", "items_claimed": 23, "items_running": 1863, "display_running": 1863, "active_threads": 0, "max_threads": 0, } ], "steps": [], "throughput": {}, }, "_snapshot_source": "runtime_ingest", } ) assert summary is not None self.assertEqual(0, summary["display_items_running"]) self.assertEqual(0, summary["display_active_threads"]) self.assertEqual([], summary["display_active_node_codes"]) def test_append_detect_job_event_skips_missing_job_fk(self) -> None: class FakeCursor: def __init__(self) -> None: self.exec_calls = [] self.selects = 0 def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def execute(self, sql, params=None): self.exec_calls.append((sql, params)) if "INSERT INTO detect_run_events" in sql: raise AssertionError("should not insert detect_run_events when job row is missing") if "SELECT 1 FROM detect_jobs" in sql: self.selects += 1 def fetchone(self): return None class FakeConn: def __init__(self) -> None: self.cursor_obj = FakeCursor() self.commit_calls = 0 def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def cursor(self): return self.cursor_obj def commit(self): self.commit_calls += 1 fake_conn = FakeConn() with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch("app.services.detect_job_service.push_debug_event") as mock_push: append_detect_job_event( 5, event_type="job_dispatch_requested", message="控制面已发送检测启动请求", ) self.assertEqual(1, fake_conn.cursor_obj.selects) self.assertEqual(0, fake_conn.commit_calls) mock_push.assert_called_once() def test_process_detect_pipeline_now_retries_deadlock_once(self) -> None: calls = {"count": 0} def _run_pipeline(*, limit, job_id): calls["count"] += 1 if calls["count"] == 1: raise errors.DeadlockDetected() return { "processed_items": 7, "advanced_items": 3, "retried_items": 1, } with patch("app.services.detect_job_service.process_detect_pipeline", side_effect=_run_pipeline): with patch("app.services.detect_job_service.time.sleep") as mock_sleep: ok, message, data = process_detect_pipeline_now(limit=123, job_id=45) self.assertTrue(ok) self.assertIn("deadlock 自动重试 1 次后成功", message) self.assertEqual(1, data["retry_attempts"]) self.assertEqual(7, data["processed_items"]) self.assertEqual(2, calls["count"]) mock_sleep.assert_called_once() def test_build_runtime_snapshot_from_projection_prefers_distributed_projection_nodes(self) -> None: snapshot = _build_runtime_snapshot_from_projection( { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 1972, "max_thread_count": 2000, "progress": { "pending": 4578, "running": 1972, "completed": 690, "blacklisted": 0, "failed": 24, }, "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": 1972, "items_failed": 24, "display_items_claimed": 120, "display_items_running": 1972, "display_active_threads": 1972, "display_max_threads": 2000, "distributed_node_stats": [ { "node_code": "mainland-controller-01-a", "items_running": 1000, "display_running": 1000, "active_threads": 1000, "max_threads": 1000, "region": "mainland", "role": "control", "status": "busy", }, { "node_code": "mainland-controller-01-b", "items_running": 972, "display_running": 972, "active_threads": 972, "max_threads": 1000, "region": "mainland", "role": "control", "status": "busy", }, ], }, } ) queue = snapshot["queue_health"]["queue"] nodes = snapshot["queue_health"]["nodes"] self.assertEqual(1972, queue["running"]) self.assertEqual(1972, queue["display_running"]) self.assertEqual(3, len(nodes)) self.assertEqual("mainland-controller-01-a", nodes[0]["node_code"]) self.assertEqual(1000, nodes[0]["display_running"]) self.assertEqual(972, nodes[1]["active_threads"]) self.assertEqual("unassigned", nodes[2]["node_code"]) def test_enrich_active_job_summary_with_runtime_does_not_overwrite_non_zero_display_with_zero_snapshot(self) -> None: summary = _enrich_active_job_summary_with_runtime( { "job_id": 60, "job_code": "sync-overseas-255", "status": "running", "items_total": 8402, "items_pending": 7865, "items_claimed": 0, "items_running": 537, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "items_terminal": 0, "display_items_running": 537, "display_active_threads": 537, "display_current_load": 537, "display_max_threads": 80000, "recent_events": [], }, runtime_activity={}, runtime_snapshot={ "job": { "job_id": 60, "job_code": "sync-overseas-255", "status": "running", "node_stats": [ { "node_code": "mainland-controller-01-a", "items_running": 0, "display_running": 0, "active_threads": 0, "max_threads": 0, } ], }, "queue_health": { "queue": { "items_total": 8402, "pending": 7865, "claimed": 0, "running": 537, "display_running": 0, "completed": 0, "blacklisted": 0, "failed": 0, "terminal": 0, }, "nodes": [ { "node_code": "mainland-controller-01-a", "items_running": 0, "display_running": 0, "active_threads": 0, "max_threads": 0, } ], "steps": [], }, }, ) assert summary is not None self.assertEqual(537, summary["display_items_running"]) self.assertEqual(537, summary["display_active_threads"]) self.assertEqual(80000, summary["display_max_threads"]) def test_classify_runtime_debug_event_maps_single_step_finalized_completed(self) -> None: classified = _classify_runtime_debug_event( event_type="worker_log", message="检测步骤跟踪: domain=example.com | step=注册状态检测 | stage=single_step_finalized | elapsed_ms=2700 | ok=1 | detect_key=detect_register | final_status=completed | result_state=passed", payload={"job_code": "sync-overseas-27456", "node_code": "mainland-worker-01"}, ) self.assertEqual("sync-overseas-27456", classified["job_code"]) self.assertTrue(classified["terminal"]) self.assertEqual("completed", classified["terminal_status"]) self.assertEqual("detect_register", classified["step_code"]) def test_build_runtime_display_bucket_prefers_real_active_threads_over_raw_current_load(self) -> None: bucket = _build_runtime_display_bucket( ( "mainland-worker-01", "mainland", "worker", "busy", 753, { "job_items_total": 1000, "job_items_claimed": 0, "job_items_running": 0, "job_items_completed": 342, "active_threads": 61, "max_threads": 400, "detect_participating": True, }, None, ) ) self.assertIsNotNone(bucket) self.assertEqual(61, bucket["current_load"]) self.assertEqual(61, bucket["display_running"]) self.assertEqual(61, bucket["active_threads"]) def test_build_runtime_display_bucket_skips_local_overseas_control_plane_load(self) -> None: with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.settings.node_code", "overseas-control-01"): bucket = _build_runtime_display_bucket( ( "overseas-control-01", "overseas", "control", "busy", 371, { "active_threads": 371, "max_threads": 3200, "detect_participating": False, }, None, ) ) self.assertIsNone(bucket) def test_build_runtime_display_bucket_skips_any_control_node_without_worker_support(self) -> None: bucket = _build_runtime_display_bucket( ( "overseas-control-01", "overseas", "control", "busy", 378, { "active_threads": 378, "max_threads": 3200, "detect_participating": False, "worker_online": False, }, None, ) ) self.assertIsNone(bucket) def test_load_latest_runtime_active_job_snapshot_uses_runtime_ingest_when_debug_missing(self) -> None: ingest_payload = { "projection": { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 177, "max_thread_count": 2000, "progress": { "pending": 200, "running": 177, "completed": 440, "blacklisted": 0, "failed": 4, }, "backlog": { "pending_total": 200, "running_total": 177, }, "active_job": { "job_id": 551, "job_code": "sync-overseas-55", "status": "running", "items_total": 821, "items_terminal": 444, "items_failed": 4, "progress_percent": 54.08, "node_stats": [ { "node_code": "mainland-controller-01-a", "items_total": 400, "items_pending": 0, "items_claimed": 80, "items_running": 100, "items_completed": 220, "items_blacklisted": 0, "items_failed": 0, "active_threads": 100, "max_threads": 1000, } ], }, } } fake_conn = _SequenceConn( fetchall_results=[ [], [(ingest_payload, datetime(2026, 4, 22, 18, 5, 36))], ] ) with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value={}): snapshot = _load_latest_runtime_active_job_snapshot(15) self.assertEqual("runtime_ingest", snapshot["_snapshot_source"]) self.assertEqual("sync-overseas-55", snapshot["job"]["job_code"]) self.assertEqual(821, snapshot["queue_health"]["queue"]["items_total"]) self.assertEqual(177, snapshot["queue_health"]["queue"]["display_running"]) self.assertEqual(200, snapshot["backlog"]["pending_total"]) def test_load_latest_runtime_active_job_snapshot_prefers_fresher_runtime_ingest(self) -> None: debug_payload = { "job": { "job_id": 84, "job_code": "sync-overseas-366", "status": "running", }, "queue_health": { "queue": { "items_total": 5000, "running": 2, } }, } ingest_payload = { "projection": { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 177, "max_thread_count": 2000, "progress": { "pending": 200, "running": 177, "completed": 440, "blacklisted": 0, "failed": 4, }, "active_job": { "job_id": 551, "job_code": "sync-overseas-55", "status": "running", "items_total": 821, "items_terminal": 444, "items_failed": 4, }, } } fake_conn = _SequenceConn( fetchall_results=[ [(debug_payload, datetime(2026, 4, 22, 17, 37, 30))], [(ingest_payload, datetime(2026, 4, 22, 18, 5, 36))], ] ) with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value={}): snapshot = _load_latest_runtime_active_job_snapshot(15) self.assertEqual("runtime_ingest", snapshot["_snapshot_source"]) self.assertEqual("sync-overseas-55", snapshot["job"]["job_code"]) def test_load_latest_runtime_active_job_snapshot_prefers_richer_ingest_over_newer_debug_snapshot(self) -> None: debug_payload = { "job": { "job_id": 376, "job_code": "sync-overseas-376", "status": "running", "display_items_running": 138, "display_active_threads": 138, "node_stats": [ { "node_code": "mainland-controller-01", "items_running": 138, "display_running": 138, "active_threads": 138, "max_threads": 2000, } ], }, "queue_health": { "queue": { "items_total": 8402, "running": 138, "display_running": 138, }, "nodes": [ { "node_code": "mainland-controller-01", "items_running": 138, "display_running": 138, "active_threads": 138, "max_threads": 2000, } ], }, } ingest_payload = { "projection": { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 5909, "max_thread_count": 80000, "progress": { "pending": 2493, "running": 5909, "completed": 0, "blacklisted": 0, "failed": 0, }, "active_job": { "job_id": 551, "job_code": "sync-overseas-51", "status": "running", "items_total": 8402, "items_terminal": 0, "items_failed": 0, "display_items_running": 5909, "display_active_threads": 5909, "distributed_node_stats": [ { "node_code": "mainland-controller-01-a", "items_running": 1000, "display_running": 1000, "active_threads": 1000, "max_threads": 1000, }, { "node_code": "mainland-controller-01-b", "items_running": 1000, "display_running": 1000, "active_threads": 1000, "max_threads": 1000, }, { "node_code": "mainland-controller-01-c", "items_running": 3909, "display_running": 3909, "active_threads": 3909, "max_threads": 78000, }, ], }, } } fake_conn = _SequenceConn( fetchall_results=[ [(debug_payload, datetime(2026, 4, 23, 2, 45, 0))], [(ingest_payload, datetime(2026, 4, 23, 2, 38, 55))], ] ) with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value={}): snapshot = _load_latest_runtime_active_job_snapshot(15) self.assertEqual("runtime_ingest", snapshot["_snapshot_source"]) self.assertEqual("sync-overseas-51", snapshot["job"]["job_code"]) self.assertEqual(5909, snapshot["queue_health"]["queue"]["display_running"]) self.assertEqual( [ "mainland-controller-01-a", "mainland-controller-01-b", "mainland-controller-01-c", "unassigned", ], [item["node_code"] for item in snapshot["queue_health"]["nodes"]], ) def test_load_latest_runtime_active_job_snapshot_prefers_focus_job_match_over_newer_debug_snapshot(self) -> None: debug_payload = { "job": { "job_id": 376, "job_code": "sync-overseas-376", "status": "running", }, "queue_health": { "queue": { "items_total": 5000, "running": 282, } }, } ingest_payload = { "projection": { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 177, "max_thread_count": 2000, "progress": { "pending": 7800, "running": 177, "completed": 420, "blacklisted": 0, "failed": 5, }, "active_job": { "job_id": 551, "job_code": "sync-overseas-51", "status": "running", "items_total": 8402, "items_terminal": 425, "items_failed": 5, }, } } fake_conn = _SequenceConn( fetchall_results=[ [(debug_payload, datetime(2026, 4, 22, 23, 17, 47))], [ ({"projection": {"active_job": {"job_code": "sync-overseas-55"}}}, datetime(2026, 4, 22, 18, 6, 10)), (ingest_payload, datetime(2026, 4, 22, 18, 5, 36)), ], ] ) with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch( "app.services.detect_job_service._load_runtime_activity_snapshot", return_value={"focus_job_code": "sync-overseas-51"}, ): snapshot = _load_latest_runtime_active_job_snapshot(15) self.assertEqual("runtime_ingest", snapshot["_snapshot_source"]) self.assertEqual("sync-overseas-51", snapshot["job"]["job_code"]) def test_load_latest_runtime_ingest_active_job_snapshot_keeps_recent_focus_match_over_marginally_newer_non_focus_snapshot(self) -> None: rich_non_focus_payload = { "projection": { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 5909, "max_thread_count": 80000, "progress": { "pending": 2493, "running": 5909, "completed": 0, "blacklisted": 0, "failed": 0, }, "active_job": { "job_id": 198, "job_code": "sync-overseas-198", "status": "running", "items_total": 8402, "items_terminal": 0, "items_failed": 0, "display_items_running": 5909, "display_active_threads": 5909, "distributed_node_stats": [ { "node_code": "mainland-controller-01-a", "items_running": 1000, "display_running": 1000, "active_threads": 1000, "max_threads": 1000, }, { "node_code": "mainland-controller-01-b", "items_running": 4909, "display_running": 4909, "active_threads": 4909, "max_threads": 79000, }, ], }, } } focus_match_payload = { "projection": { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 177, "max_thread_count": 2000, "progress": { "pending": 7800, "running": 177, "completed": 420, "blacklisted": 0, "failed": 5, }, "active_job": { "job_id": 551, "job_code": "sync-overseas-51", "status": "running", "items_total": 8402, "items_terminal": 425, "items_failed": 5, "display_items_running": 177, "display_active_threads": 177, "distributed_node_stats": [ { "node_code": "mainland-controller-01", "items_running": 177, "display_running": 177, "active_threads": 177, "max_threads": 2000, } ], }, } } fake_conn = _SequenceConn( fetchall_results=[ [ (rich_non_focus_payload, datetime(2026, 4, 23, 2, 45, 0)), (focus_match_payload, datetime(2026, 4, 23, 2, 41, 0)), ] ] ) with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.get_db", return_value=fake_conn): snapshot = detect_job_service_module._load_latest_runtime_ingest_active_job_snapshot( 15, preferred_job_codes=["sync-overseas-51"], ) self.assertEqual("runtime_ingest", snapshot["_snapshot_source"]) self.assertEqual("sync-overseas-51", snapshot["job"]["job_code"]) def test_load_latest_runtime_active_job_snapshot_prefers_newer_richer_ingest_over_stale_focus_job_match(self) -> None: rich_ingest_payload = { "projection": { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 5909, "max_thread_count": 80000, "progress": { "pending": 2493, "running": 5909, "completed": 0, "blacklisted": 0, "failed": 0, }, "active_job": { "job_id": 198, "job_code": "sync-overseas-198", "status": "running", "items_total": 8402, "items_terminal": 0, "items_failed": 0, "display_items_running": 5909, "display_active_threads": 5909, "distributed_node_stats": [ { "node_code": "mainland-controller-01-a", "items_running": 1000, "display_running": 1000, "active_threads": 1000, "max_threads": 1000, }, { "node_code": "mainland-controller-01-b", "items_running": 1000, "display_running": 1000, "active_threads": 1000, "max_threads": 1000, }, { "node_code": "mainland-controller-01-c", "items_running": 3909, "display_running": 3909, "active_threads": 3909, "max_threads": 78000, }, ], }, } } stale_focus_ingest_payload = { "projection": { "node": { "node_code": "mainland-controller-01", "region": "mainland", "role": "control", }, "active_thread_count": 138, "max_thread_count": 2000, "progress": { "pending": 5001, "running": 0, "completed": 0, "blacklisted": 0, "failed": 0, }, "active_job": { "job_id": 51, "job_code": "sync-overseas-51", "status": "running", "items_total": 8402, "items_terminal": 3402, "items_failed": 0, "display_items_running": 138, "display_active_threads": 138, "distributed_node_stats": [ { "node_code": "mainland-controller-01", "items_running": 0, "display_running": 138, "active_threads": 138, "max_threads": 2000, } ], }, } } fake_conn = _SequenceConn( fetchall_results=[ [], [ (rich_ingest_payload, datetime(2026, 4, 23, 2, 38, 55)), (stale_focus_ingest_payload, datetime(2026, 4, 22, 18, 2, 4)), ], ] ) with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch( "app.services.detect_job_service._load_runtime_activity_snapshot", return_value={"focus_job_code": "sync-overseas-51", "job_codes": ["sync-overseas-51"]}, ): snapshot = _load_latest_runtime_active_job_snapshot(15) self.assertEqual("runtime_ingest", snapshot["_snapshot_source"]) self.assertEqual("sync-overseas-198", snapshot["job"]["job_code"]) self.assertEqual(5909, snapshot["queue_health"]["queue"]["display_running"]) def test_get_active_detect_job_summary_falls_back_to_runtime_snapshot(self) -> None: runtime_snapshot = { "job": { "job_id": 551, "job_code": "sync-overseas-55", "status": "running", "progress_percent": 54.08, "node_stats": [ { "node_code": "mainland-controller-01", "items_total": 821, "items_pending": 200, "items_claimed": 0, "items_running": 177, "items_completed": 440, "items_blacklisted": 0, "items_failed": 4, "active_threads": 177, "max_threads": 2000, } ], }, "queue_health": { "queue": { "items_total": 821, "pending": 200, "claimed": 0, "running": 177, "display_claimed": 0, "display_running": 177, "completed": 440, "blacklisted": 0, "failed": 4, "terminal": 444, }, "nodes": [ { "node_code": "mainland-controller-01", "items_total": 821, "items_pending": 200, "items_claimed": 0, "items_running": 177, "items_completed": 440, "items_blacklisted": 0, "items_failed": 4, "active_threads": 177, "max_threads": 2000, } ], "throughput": { "processed_recent": 0, "processed_per_minute": 0, "completed_recent": 0, "blacklisted_recent": 0, "failed_recent": 0, }, }, "_created_at": "2026-04-22 18:05:36", "_snapshot_source": "runtime_ingest", } fake_conn = _SequenceConn(fetchall_results=[[]]) with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value=runtime_snapshot): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value={}): summary = get_active_detect_job_summary(event_limit=10) self.assertIsNotNone(summary) self.assertEqual("sync-overseas-55", summary["job_code"]) self.assertEqual(821, summary["items_total"]) self.assertEqual(177, summary["display_items_running"]) def test_recycle_expired_detect_job_items_once_refreshes_touched_jobs(self) -> None: conn = MagicMock() main_cursor_cm = MagicMock() main_cursor = MagicMock() unlock_cursor_cm = MagicMock() unlock_cursor = MagicMock() main_cursor_cm.__enter__.return_value = main_cursor main_cursor_cm.__exit__.return_value = False unlock_cursor_cm.__enter__.return_value = unlock_cursor unlock_cursor_cm.__exit__.return_value = False conn.cursor.side_effect = [main_cursor_cm, unlock_cursor_cm] main_cursor.fetchone.side_effect = [ (True,), (2, [11, 12]), ("domain_pipeline", 1, 0, 0, 0, 0), ("single_step", 0, 0, 1, 0, 0), ] db_cm = MagicMock() db_cm.__enter__.return_value = conn db_cm.__exit__.return_value = False with patch("app.services.detect_job_service.get_db", return_value=db_cm): with patch("app.services.detect_job_service.push_debug_event") as mock_push: recycled = detect_job_service_module._recycle_expired_detect_job_items_once() self.assertEqual(2, recycled) self.assertTrue(any("UPDATE detect_job_items" in call.args[0] for call in main_cursor.execute.call_args_list)) self.assertTrue(any("FOR UPDATE SKIP LOCKED" in call.args[0] for call in main_cursor.execute.call_args_list)) self.assertTrue( any( "UPDATE detect_jobs" in call.args[0] and call.args[1] == (11,) for call in main_cursor.execute.call_args_list ) ) self.assertTrue( any( "UPDATE detect_jobs" in call.args[0] and call.args[1] == ("failed", 12) for call in main_cursor.execute.call_args_list ) ) unlock_cursor.execute.assert_called_once_with( "SELECT pg_advisory_unlock(%s)", (detect_job_service_module._DETECT_JOB_ITEM_RECYCLE_ADVISORY_LOCK_KEY,), ) self.assertGreaterEqual(conn.commit.call_count, 1) conn.rollback.assert_not_called() mock_push.assert_called_once() def test_recycle_expired_detect_job_items_once_skips_when_lock_is_busy(self) -> None: conn = MagicMock() main_cursor_cm = MagicMock() main_cursor = MagicMock() main_cursor_cm.__enter__.return_value = main_cursor main_cursor_cm.__exit__.return_value = False conn.cursor.return_value = main_cursor_cm main_cursor.fetchone.return_value = (False,) db_cm = MagicMock() db_cm.__enter__.return_value = conn db_cm.__exit__.return_value = False with patch("app.services.detect_job_service.get_db", return_value=db_cm): with patch("app.services.detect_job_service.push_debug_event") as mock_push: recycled = detect_job_service_module._recycle_expired_detect_job_items_once() self.assertEqual(0, recycled) conn.rollback.assert_called_once() conn.commit.assert_not_called() mock_push.assert_not_called() def test_get_active_detect_job_summary_attempts_expired_item_recycle_before_query(self) -> None: fake_conn = _SequenceConn(fetchall_results=[[]]) with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch("app.services.detect_job_service._maybe_recycle_expired_detect_job_items") as mock_recycle: with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value={}): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value={}): summary = get_active_detect_job_summary(event_limit=10) self.assertIsNone(summary) mock_recycle.assert_called_once_with() def test_get_active_detect_job_summary_prefers_runtime_focus_job_code(self) -> None: rows = [ ( 85, "sync-overseas-376", "manual", "domain_pipeline", "", "running", "tester", datetime(2026, 4, 22, 22, 0, 0), datetime(2026, 4, 22, 22, 0, 1), None, ), ( 11, "sync-overseas-51", "manual", "domain_pipeline", "", "running", "tester", datetime(2026, 4, 22, 17, 0, 0), datetime(2026, 4, 22, 17, 0, 1), None, ), ] fake_conn = _SequenceConn(fetchall_results=[rows]) selected_rows = [] def _fake_fetch(_cur, row, event_limit=20): selected_rows.append((row, event_limit)) return {"job_id": row[0], "job_code": row[1], "status": row[5]} with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch( "app.services.detect_job_service._load_runtime_activity_snapshot", return_value={"focus_job_code": "sync-overseas-51", "job_codes": ["sync-overseas-51"]}, ): with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value={}): with patch("app.services.detect_job_service._fetch_job_summary", side_effect=_fake_fetch): with patch( "app.services.detect_job_service._enrich_active_job_summary_with_runtime", side_effect=lambda summary, **kwargs: summary, ): summary = get_active_detect_job_summary(event_limit=10) self.assertEqual("sync-overseas-51", summary["job_code"]) self.assertEqual("sync-overseas-51", selected_rows[0][0][1]) self.assertEqual(10, selected_rows[0][1]) def test_get_active_detect_job_summary_prefers_runtime_snapshot_job_code_over_stale_runtime_focus(self) -> None: rows = [ ( 85, "sync-overseas-376", "manual", "domain_pipeline", "", "running", "tester", datetime(2026, 4, 22, 22, 0, 0), datetime(2026, 4, 22, 22, 0, 1), None, ), ( 11, "sync-overseas-51", "manual", "domain_pipeline", "", "running", "tester", datetime(2026, 4, 22, 17, 0, 0), datetime(2026, 4, 22, 17, 0, 1), None, ), ] fake_conn = _SequenceConn(fetchall_results=[rows]) selected_rows = [] def _fake_fetch(_cur, row, event_limit=20): selected_rows.append((row, event_limit)) return {"job_id": row[0], "job_code": row[1], "status": row[5]} with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch( "app.services.detect_job_service._load_runtime_activity_snapshot", return_value={"focus_job_code": "sync-overseas-51", "job_codes": ["sync-overseas-51"]}, ): with patch( "app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value={"job": {"job_code": "sync-overseas-376"}}, ): with patch("app.services.detect_job_service._fetch_job_summary", side_effect=_fake_fetch): with patch( "app.services.detect_job_service._enrich_active_job_summary_with_runtime", side_effect=lambda summary, **kwargs: summary, ): summary = get_active_detect_job_summary(event_limit=10) self.assertEqual("sync-overseas-376", summary["job_code"]) self.assertEqual("sync-overseas-376", selected_rows[0][0][1]) self.assertEqual(10, selected_rows[0][1]) def test_get_active_detect_job_summary_falls_back_when_detect_jobs_legacy_columns_are_missing(self) -> None: rows = [ ( 85, "sync-overseas-376", "manual", "domain_pipeline", "", "running", "tester", datetime(2026, 4, 22, 22, 0, 0), datetime(2026, 4, 22, 22, 0, 1), None, ), ] class LegacyCursor(_SequenceCursor): def execute(self, sql, params=None): self.exec_calls.append((sql, params)) normalized_sql = str(sql or "") if ( "FROM detect_jobs" in normalized_sql and "task_mode" in normalized_sql and " AS task_mode" not in normalized_sql ): raise errors.UndefinedColumn() class LegacyConn(_SequenceConn): def __init__(self) -> None: self.cursor_obj = LegacyCursor(fetchall_results=[rows]) fake_conn = LegacyConn() with patch.object(detect_job_service_module, "_DETECT_JOBS_LEGACY_SELECT_MODE", None): with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value={}): with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value={}): with patch( "app.services.detect_job_service._fetch_job_summary", side_effect=lambda _cur, row, event_limit=20: { "job_id": row[0], "job_code": row[1], "task_mode": row[3], "step_code": row[4], "status": row[5], }, ): with patch( "app.services.detect_job_service._enrich_active_job_summary_with_runtime", side_effect=lambda summary, **kwargs: summary, ): summary = get_active_detect_job_summary(event_limit=10) self.assertEqual("sync-overseas-376", summary["job_code"]) self.assertEqual("domain_pipeline", summary["task_mode"]) self.assertEqual("", summary["step_code"]) self.assertTrue(any(" AS task_mode" in call[0] for call in fake_conn.cursor_obj.exec_calls)) def test_build_display_summary_ignores_inflated_raw_current_load(self) -> None: summary = _build_display_summary( [ { "node_code": "mainland-controller-01", "items_claimed": 0, "items_running": 0, "display_running": 729, "current_load": 729, "active_threads": 729, "max_threads": 800, "items_completed": 196, "items_failed": 0, }, { "node_code": "mainland-worker-01", "items_claimed": 0, "items_running": 0, "display_running": 753, "current_load": 753, "active_threads": 61, "max_threads": 400, "items_completed": 146, "items_failed": 0, }, ] ) self.assertEqual(790, summary["display_running"]) self.assertEqual(790, summary["current_load"]) self.assertEqual(790, summary["active_threads"]) def test_enrich_active_job_summary_with_runtime_adds_runtime_job_code_and_recent_events(self) -> None: summary = _enrich_active_job_summary_with_runtime( { "job_id": 275, "job_code": "sync-overseas-5297", "processed_recent": 0, "processed_per_minute": 0, "completed_recent": 0, "failed_recent": 0, "blacklisted_recent": 0, "recent_domain_events": [], }, event_limit=20, window_minutes=15, runtime_activity={ "focus_job_code": "sync-overseas-9506", "job_codes": ["sync-overseas-9506", "sync-overseas-27456"], "processed_recent": 120, "completed_recent": 118, "failed_recent": 1, "blacklisted_recent": 1, }, runtime_snapshot={ "job": { "job_id": 1835, "job_code": "sync-overseas-9506", "progress_percent": 44.2, }, "queue_health": { "queue": { "items_total": 1000, "pending": 176, "claimed": 223, "running": 159, "completed": 442, "display_running": 220, } }, }, recent_domain_events=[ { "event_type": "domain_completed", "message": "域名检测完成: example.com", "created_at": "2026-04-21 04:10:00", } ], ) self.assertEqual("sync-overseas-9506", summary["runtime_job_code"]) self.assertEqual(["sync-overseas-9506", "sync-overseas-27456"], summary["runtime_job_codes"]) self.assertEqual(120, summary["processed_recent"]) self.assertEqual(8.0, summary["processed_per_minute"]) self.assertEqual(118, summary["completed_recent"]) self.assertEqual(1, summary["failed_recent"]) self.assertEqual(1, summary["blacklisted_recent"]) self.assertEqual(1, len(summary["recent_domain_events"])) self.assertEqual("sync-overseas-9506", summary["runtime_snapshot_job_code"]) self.assertEqual(1000, summary["runtime_snapshot_queue"]["items_total"]) self.assertEqual(1000, summary["items_total"]) self.assertEqual(176, summary["items_pending"]) self.assertEqual(223, summary["items_claimed"]) self.assertEqual(159, summary["items_running"]) self.assertEqual(442, summary["items_completed"]) self.assertEqual(220, summary["display_items_running"]) def test_enrich_active_job_summary_with_runtime_prefers_snapshot_identity_and_events(self) -> None: summary = _enrich_active_job_summary_with_runtime( { "job_id": 275, "job_code": "sync-overseas-5297", "status": "running", "recent_events": [ { "node_code": "mainland-worker-01", "event_type": "domain_started", "message": "开始检测域名: stale.com", "payload": {"job_code": "sync-overseas-5297"}, "created_at": "2026-04-21 11:58:00", } ], "current_cycle_events": [], "latest_event": None, "display_items_running": 0, "display_current_load": 0, "display_active_threads": 0, "display_max_threads": 0, "display_active_node_codes": [], }, event_limit=10, window_minutes=15, runtime_activity={ "focus_job_code": "sync-overseas-31437", "job_codes": ["sync-overseas-31437"], "processed_recent": 10, "completed_recent": 10, "failed_recent": 0, "blacklisted_recent": 0, }, runtime_snapshot={ "job": { "job_id": 1902, "job_code": "sync-overseas-31437", "status": "running", "progress_percent": 34.3, "node_stats": [ { "node_code": "mainland-controller-01", "items_claimed": 230, "items_running": 67, "items_completed": 343, "items_failed": 0, "active_threads": 170, "max_threads": 2000, }, { "node_code": "mainland-worker-01", "items_claimed": 75, "items_running": 0, "items_completed": 0, "items_failed": 0, "active_threads": 19, "max_threads": 1200, }, ], }, "queue_health": { "queue": { "items_total": 1000, "pending": 285, "claimed": 305, "running": 67, "completed": 343, "display_running": 189, } }, "recent_events": [ { "job_id": 2137, "node_code": "mainland-controller-01", "event_type": "job_created", "message": "同步拉取待检测批次 sync-overseas-35461，共 1000 个任务项", "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": "从任务队列获取到 125 个需要检测的域名", "payload": {"job_code": "sync-overseas-31437"}, "created_at": "2026-04-22 01:19:25", }, ], }, ) self.assertEqual(1902, summary["job_id"]) self.assertEqual("sync-overseas-31437", summary["job_code"]) self.assertEqual("sync-overseas-31437", summary["runtime_job_code"]) self.assertEqual(189, summary["display_items_running"]) self.assertEqual(189, summary["display_current_load"]) self.assertEqual(189, summary["display_active_threads"]) self.assertEqual(3200, summary["display_max_threads"]) self.assertEqual(["mainland-controller-01", "mainland-worker-01"], summary["display_active_node_codes"]) self.assertEqual(1, len(summary["recent_events"])) self.assertEqual("worker_log", summary["latest_event"]["event_type"]) self.assertEqual("sync-overseas-31437", summary["recent_events"][0]["payload"]["job_code"]) def test_enrich_active_job_summary_with_runtime_does_not_promote_unassigned_snapshot_running(self) -> None: summary = _enrich_active_job_summary_with_runtime( { "job_id": 384, "job_code": "sync-overseas-16654", "status": "running", "display_items_running": 331, "display_current_load": 331, "display_active_threads": 331, "display_max_threads": 1000, "display_active_node_codes": [], }, event_limit=10, window_minutes=15, runtime_activity={ "focus_job_code": "sync-overseas-16654", "job_codes": ["sync-overseas-16654"], "processed_recent": 0, "completed_recent": 0, "failed_recent": 0, "blacklisted_recent": 0, }, runtime_snapshot={ "_has_runtime_node_rows": True, "_dropped_runtime_node_count": 0, "job": { "job_id": 384, "job_code": "sync-overseas-16654", "status": "running", "node_stats": [ { "node_code": "unassigned", "items_claimed": 23, "items_running": 1863, "display_running": 1863, "active_threads": 0, "max_threads": 0, } ], }, "queue_health": { "queue": { "items_total": 4000, "pending": 0, "claimed": 23, "running": 2100, "completed": 1117, "blacklisted": 0, "failed": 760, "display_running": 2100, }, "nodes": [ { "node_code": "unassigned", "items_claimed": 23, "items_running": 1863, "display_running": 1863, "active_threads": 0, "max_threads": 0, } ], }, }, ) self.assertEqual(0, summary["display_items_running"]) self.assertEqual(0, summary["display_active_threads"]) self.assertEqual([], summary["display_active_node_codes"]) def test_get_detect_queue_health_preserves_runtime_node_throughput_after_snapshot_override(self) -> None: lease_row = (None, None, 0, 0) throughput_rows = [] step_throughput_rows = [] runtime_display_rows = [ ( "mainland-controller-01", "mainland", "control", "busy", 170, { "job_items_total": 265, "job_items_claimed": 85, "job_items_running": 0, "job_items_completed": 105, "active_threads": 170, "max_threads": 2000, "detect_participating": True, }, None, ), ( "mainland-worker-01", "mainland", "worker", "busy", 19, { "job_items_total": 139, "job_items_claimed": 64, "job_items_running": 0, "job_items_completed": 4, "active_threads": 19, "max_threads": 1200, "detect_participating": True, }, None, ), ( "overseas-control-01", "overseas", "control", "busy", 371, { "active_threads": 371, "max_threads": 3200, "detect_participating": False, }, None, ), ] conn = MagicMock() cursor_cm = MagicMock() cursor = MagicMock() conn.cursor.return_value = cursor_cm cursor_cm.__enter__.return_value = cursor db_cm = MagicMock() db_cm.__enter__.return_value = conn cursor.fetchone.return_value = lease_row cursor.fetchall.side_effect = [throughput_rows, step_throughput_rows, runtime_display_rows] active_job = { "job_id": 1937, "job_code": "sync-overseas-31987", "status": "running", "items_total": 1000, "items_pending": 596, "items_claimed": 149, "items_running": 146, "items_completed": 109, "items_blacklisted": 0, "items_failed": 0, "progress_percent": 10.9, "node_stats": [], "distributed_node_stats": [ { "node_code": "mainland-controller-01", "items_total": 265, "items_pending": 0, "items_claimed": 85, "items_running": 75, "items_completed": 105, "items_blacklisted": 0, "items_failed": 0, "metrics_source": "runtime", }, { "node_code": "mainland-worker-01", "items_total": 139, "items_pending": 0, "items_claimed": 64, "items_running": 71, "items_completed": 4, "items_blacklisted": 0, "items_failed": 0, "metrics_source": "runtime", }, ], "step_stats": [ { "step_code": "detect_register", "items_total": 1000, "items_pending": 449, "items_claimed": 131, "items_running": 273, "items_completed": 147, "items_blacklisted": 0, "items_failed": 0, } ], } runtime_activity = { "processed_recent": 133, "completed_recent": 113, "failed_recent": 20, "blacklisted_recent": 0, "focus_job_code": "sync-overseas-31563", "job_codes": ["sync-overseas-31563"], "step_code": "detect_register", "nodes": { "mainland-controller-01": { "node_code": "mainland-controller-01", "processed_recent": 74, "completed_recent": 54, "failed_recent": 20, "blacklisted_recent": 0, }, "mainland-worker-01": { "node_code": "mainland-worker-01", "processed_recent": 59, "completed_recent": 59, "failed_recent": 0, "blacklisted_recent": 0, }, }, } runtime_snapshot = { "job": { "job_id": 1937, "job_code": "sync-overseas-31987", "progress_percent": 10.9, }, "queue_health": { "queue": { "items_total": 1000, "pending": 596, "claimed": 149, "running": 146, "completed": 109, "blacklisted": 0, "failed": 0, "display_claimed": 149, "display_running": 403, }, "nodes": [ { "node_code": "mainland-controller-01", "items_total": 265, "items_pending": 0, "items_claimed": 85, "items_running": 75, "items_completed": 105, "items_blacklisted": 0, "items_failed": 0, "processed_recent": 0, "processed_per_minute": 0, "completed_recent": 0, "blacklisted_recent": 0, "failed_recent": 0, "metrics_source": "runtime", }, { "node_code": "mainland-worker-01", "items_total": 139, "items_pending": 0, "items_claimed": 64, "items_running": 71, "items_completed": 4, "items_blacklisted": 0, "items_failed": 0, "processed_recent": 0, "processed_per_minute": 0, "completed_recent": 0, "blacklisted_recent": 0, "failed_recent": 0, "metrics_source": "runtime", }, ], "steps": [], }, } with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.settings.node_code", "overseas-control-01"): with patch("app.services.detect_job_service.get_active_detect_job_summary", return_value=active_job): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value=runtime_activity): with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value=runtime_snapshot): with patch("app.services.detect_job_service._load_runtime_display_rows", return_value=runtime_display_rows): with patch("app.services.detect_job_service.get_db", return_value=db_cm): health = get_detect_queue_health(window_minutes=15) node_map = {item["node_code"]: item for item in health["nodes"]} self.assertEqual(74, node_map["mainland-controller-01"]["processed_recent"]) self.assertEqual(54, node_map["mainland-controller-01"]["completed_recent"]) self.assertEqual(20, node_map["mainland-controller-01"]["failed_recent"]) self.assertEqual(170, node_map["mainland-controller-01"]["items_running"]) self.assertEqual(59, node_map["mainland-worker-01"]["processed_recent"]) self.assertEqual(59, node_map["mainland-worker-01"]["completed_recent"]) self.assertEqual(19, node_map["mainland-worker-01"]["items_running"]) self.assertEqual(113, health["throughput"]["completed_recent"]) self.assertEqual(189, health["queue"]["display_running"]) self.assertNotIn("overseas-control-01", node_map) def test_get_detect_queue_health_preserves_snapshot_node_display_metrics_for_child_instances(self) -> None: lease_row = (None, None, 0, 0) throughput_rows = [] step_throughput_rows = [] runtime_display_rows = [] conn = MagicMock() cursor_cm = MagicMock() cursor = MagicMock() conn.cursor.return_value = cursor_cm cursor_cm.__enter__.return_value = cursor db_cm = MagicMock() db_cm.__enter__.return_value = conn cursor.fetchone.return_value = lease_row cursor.fetchall.side_effect = [throughput_rows, step_throughput_rows, runtime_display_rows] active_job = { "job_id": 198, "job_code": "sync-overseas-198", "status": "running", "items_total": 10344, "items_pending": 4945, "items_claimed": 0, "items_running": 5909, "items_completed": 5344, "items_blacklisted": 0, "items_failed": 42, "progress_percent": 52.07, "display_items_claimed": 13, "display_items_running": 11799, "display_active_threads": 11799, "display_max_threads": 75400, "node_stats": [], "distributed_node_stats": [ { "node_code": "mainland-controller-01-ae", "items_total": 947, "items_pending": 0, "items_claimed": 0, "items_running": 947, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, }, { "node_code": "mainland-controller-01-au", "items_total": 637, "items_pending": 0, "items_claimed": 0, "items_running": 637, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, }, ], "step_stats": [], } runtime_activity = { "processed_recent": 0, "completed_recent": 0, "failed_recent": 0, "blacklisted_recent": 0, "focus_job_code": "sync-overseas-198", "job_codes": ["sync-overseas-198"], "step_code": "detect_register", "nodes": {}, } runtime_snapshot = { "job": { "job_id": 198, "job_code": "sync-overseas-198", "progress_percent": 52.07, }, "queue_health": { "queue": { "items_total": 10344, "pending": 4945, "claimed": 0, "running": 5909, "completed": 5344, "blacklisted": 0, "failed": 42, "display_claimed": 13, "display_running": 5909, }, "nodes": [ { "node_code": "mainland-controller-01-ae", "items_total": 947, "items_pending": 0, "items_claimed": 0, "items_running": 947, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "display_running": 947, "current_load": 947, "active_threads": 947, "max_threads": 1000, }, { "node_code": "mainland-controller-01-au", "items_total": 637, "items_pending": 0, "items_claimed": 0, "items_running": 637, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "display_running": 637, "current_load": 637, "active_threads": 637, "max_threads": 1000, }, ], "steps": [], }, } with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.settings.node_code", "overseas-control-01"): with patch("app.services.detect_job_service.get_active_detect_job_summary", return_value=active_job): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value=runtime_activity): with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value=runtime_snapshot): with patch("app.services.detect_job_service.get_db", return_value=db_cm): with patch("app.services.detect_job_service._load_disabled_managed_node_codes", return_value=set()): health = get_detect_queue_health(window_minutes=15) node_map = {item["node_code"]: item for item in health["nodes"]} self.assertEqual(1584, health["queue"]["display_running"]) self.assertEqual(947, node_map["mainland-controller-01-ae"]["display_running"]) self.assertEqual(947, node_map["mainland-controller-01-ae"]["active_threads"]) self.assertEqual(1000, node_map["mainland-controller-01-ae"]["max_threads"]) self.assertEqual(637, node_map["mainland-controller-01-au"]["display_running"]) self.assertEqual(637, node_map["mainland-controller-01-au"]["active_threads"]) def test_get_detect_queue_health_drops_stale_runtime_nodes_from_display_running(self) -> None: lease_row = (None, None, 0, 0) throughput_rows = [] step_throughput_rows = [] runtime_display_rows = [] conn = MagicMock() cursor_cm = MagicMock() cursor = MagicMock() conn.cursor.return_value = cursor_cm cursor_cm.__enter__.return_value = cursor db_cm = MagicMock() db_cm.__enter__.return_value = conn cursor.fetchone.return_value = lease_row cursor.fetchall.side_effect = [throughput_rows, step_throughput_rows, runtime_display_rows] stale_heartbeat = (datetime.now() - timedelta(minutes=8)).isoformat(sep=" ", timespec="seconds") live_heartbeat = datetime.now().isoformat(sep=" ", timespec="seconds") active_job = { "job_id": 199, "job_code": "sync-overseas-199", "status": "running", "items_total": 2600, "items_pending": 500, "items_claimed": 0, "items_running": 2100, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "progress_percent": 80.77, "display_items_running": 2100, "display_active_threads": 2100, "display_max_threads": 2000, "node_stats": [], "distributed_node_stats": [ { "node_code": "mainland-controller-01-a", "items_total": 300, "items_pending": 0, "items_claimed": 0, "items_running": 300, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, }, { "node_code": "mainland-controller-01-b", "items_total": 1800, "items_pending": 0, "items_claimed": 0, "items_running": 1800, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, }, ], "step_stats": [], } runtime_activity = { "processed_recent": 0, "completed_recent": 0, "failed_recent": 0, "blacklisted_recent": 0, "focus_job_code": "sync-overseas-199", "job_codes": ["sync-overseas-199"], "step_code": "detect_register", "nodes": {}, } runtime_snapshot = { "_created_at": datetime.now().isoformat(sep=" ", timespec="seconds"), "job": { "job_id": 199, "job_code": "sync-overseas-199", "progress_percent": 80.77, }, "queue_health": { "queue": { "items_total": 2600, "pending": 500, "claimed": 0, "running": 2100, "completed": 0, "blacklisted": 0, "failed": 0, "display_claimed": 0, "display_running": 2100, }, "nodes": [ { "node_code": "mainland-controller-01-a", "items_total": 300, "items_pending": 0, "items_claimed": 0, "items_running": 300, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "display_running": 300, "current_load": 300, "active_threads": 300, "max_threads": 1000, "status": "busy", "last_heartbeat_at": live_heartbeat, }, { "node_code": "mainland-controller-01-b", "items_total": 1800, "items_pending": 0, "items_claimed": 0, "items_running": 1800, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "display_running": 1800, "current_load": 1800, "active_threads": 1800, "max_threads": 1000, "status": "stale", "last_heartbeat_at": stale_heartbeat, }, ], "steps": [], }, } with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.settings.node_code", "overseas-control-01"): with patch("app.services.detect_job_service.get_active_detect_job_summary", return_value=active_job): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value=runtime_activity): with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value=runtime_snapshot): with patch("app.services.detect_job_service._load_runtime_display_rows", return_value=runtime_display_rows): with patch("app.services.detect_job_service.get_db", return_value=db_cm): health = get_detect_queue_health(window_minutes=15) self.assertEqual(300, health["queue"]["running"]) self.assertEqual(300, health["queue"]["display_running"]) self.assertEqual(["mainland-controller-01-a"], [item["node_code"] for item in health["nodes"]]) def test_enrich_active_job_summary_drops_old_runtime_ingest_nodes_without_heartbeat(self) -> None: stale_snapshot_created_at = (datetime.now() - timedelta(minutes=12)).isoformat(sep=" ", timespec="seconds") summary = _enrich_active_job_summary_with_runtime( { "job_id": 384, "job_code": "sync-overseas-16654", "status": "running", "display_items_running": 0, "display_current_load": 0, "display_active_threads": 0, "display_max_threads": 0, "display_active_node_codes": [], }, event_limit=10, window_minutes=15, runtime_activity={ "focus_job_code": "sync-overseas-16654", "job_codes": ["sync-overseas-16654"], "processed_recent": 0, "completed_recent": 0, "failed_recent": 0, "blacklisted_recent": 0, }, runtime_snapshot={ "_snapshot_source": "runtime_ingest", "_created_at": stale_snapshot_created_at, "job": { "job_id": 384, "job_code": "sync-overseas-16654", "status": "running", "display_items_running": 2100, "display_active_threads": 2100, "node_stats": [ { "node_code": "mainland-controller-01-ah", "items_running": 62, "display_running": 62, }, { "node_code": "mainland-controller-01-bn", "items_running": 61, "display_running": 61, }, { "node_code": "unassigned", "items_running": 1863, "display_running": 1863, }, ], }, "queue_health": { "queue": { "items_total": 4000, "pending": 0, "claimed": 23, "running": 2100, "completed": 1117, "blacklisted": 0, "failed": 760, "display_running": 2100, }, "nodes": [ { "node_code": "mainland-controller-01-ah", "items_running": 62, "display_running": 62, }, { "node_code": "mainland-controller-01-bn", "items_running": 61, "display_running": 61, }, { "node_code": "unassigned", "items_running": 1863, "display_running": 1863, }, ], }, }, ) self.assertEqual(0, summary["display_items_running"]) self.assertEqual(0, summary["display_active_threads"]) self.assertEqual([], summary["display_active_node_codes"]) def test_get_detect_queue_health_aligns_runtime_job_identity_with_active_job(self) -> None: lease_row = (None, None, 0, 0) throughput_rows = [] step_throughput_rows = [] runtime_display_rows = [] conn = MagicMock() cursor_cm = MagicMock() cursor = MagicMock() conn.cursor.return_value = cursor_cm cursor_cm.__enter__.return_value = cursor db_cm = MagicMock() db_cm.__enter__.return_value = conn cursor.fetchone.return_value = lease_row cursor.fetchall.side_effect = [throughput_rows, step_throughput_rows, runtime_display_rows] active_job = { "job_id": 112, "job_code": "sync-overseas-599", "runtime_job_code": "sync-overseas-599", "runtime_snapshot_job_code": "sync-overseas-599", "status": "running", "items_total": 5274, "items_pending": 4959, "items_claimed": 0, "items_running": 840, "items_completed": 274, "items_blacklisted": 0, "items_failed": 21, "progress_percent": 5.59, "display_items_claimed": 17, "display_items_running": 1377, "display_active_threads": 1377, "display_max_threads": 81200, "node_stats": [], "distributed_node_stats": [], "step_stats": [], } runtime_activity = { "processed_recent": 93, "completed_recent": 24, "failed_recent": 69, "blacklisted_recent": 0, "focus_job_code": "sync-overseas-198", "job_codes": ["sync-overseas-198", "sync-overseas-97"], "step_code": "detect_360_site", "step_stats": {}, "nodes": {}, } runtime_snapshot = { "job": { "job_id": 112, "job_code": "sync-overseas-599", "status": "running", "progress_percent": 5.59, }, "queue_health": { "queue": { "items_total": 5274, "pending": 4959, "claimed": 0, "running": 1377, "display_claimed": 17, "display_running": 1377, "completed": 274, "blacklisted": 0, "failed": 21, }, "nodes": [], "steps": [], }, } with patch("app.services.detect_job_service.get_active_detect_job_summary", return_value=active_job): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value=runtime_activity): with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value=runtime_snapshot): with patch("app.services.detect_job_service.get_db", return_value=db_cm): with patch("app.services.detect_job_service._load_disabled_managed_node_codes", return_value=set()): health = get_detect_queue_health(window_minutes=15) self.assertEqual("sync-overseas-599", health["job"]["job_code"]) self.assertEqual("sync-overseas-599", health["job"]["runtime_job_code"]) self.assertEqual("sync-overseas-599", health["job"]["runtime_job_codes"][0]) self.assertEqual(1377, health["queue"]["running"]) self.assertEqual(1377, health["queue"]["display_running"]) def test_get_detect_queue_health_ignores_mismatched_runtime_overlay_for_unrelated_job(self) -> None: lease_row = (None, None, 0, 0) throughput_rows = [] step_throughput_rows = [] runtime_display_rows = [] conn = MagicMock() cursor_cm = MagicMock() cursor = MagicMock() conn.cursor.return_value = cursor_cm cursor_cm.__enter__.return_value = cursor db_cm = MagicMock() db_cm.__enter__.return_value = conn cursor.fetchone.return_value = lease_row cursor.fetchall.side_effect = [throughput_rows, step_throughput_rows, runtime_display_rows] active_job = { "job_id": 1, "job_code": "detect-20260421232649-acfa3d", "runtime_job_code": "sync-overseas-40", "status": "running", "items_total": 14961, "items_pending": 9478, "items_claimed": 0, "items_running": 1, "items_completed": 4996, "items_blacklisted": 0, "items_failed": 486, "progress_percent": 36.64, "display_items_claimed": 0, "display_items_running": 0, "display_active_threads": 0, "display_max_threads": 0, "node_stats": [], "distributed_node_stats": [ { "node_code": "mainland-worker-01", "items_total": 0, "items_pending": 0, "items_claimed": 0, "items_running": 0, "display_running": 1, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "metrics_source": "agent-heartbeat", "region": "mainland", "role": "worker", "status": "busy", "current_load": 1, "active_threads": 1, "max_threads": 1, "last_heartbeat_at": datetime.now().isoformat(sep=" ", timespec="seconds"), }, { "node_code": "mainland-controller-01", "items_total": 1, "items_pending": 0, "items_claimed": 0, "items_running": 1, "display_running": 0, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "metrics_source": "", "region": "", "role": "", "status": "", "current_load": 1, "active_threads": 0, "max_threads": 0, "last_heartbeat_at": "", }, { "node_code": "unassigned", "items_total": 14960, "items_pending": 14960, "items_claimed": 0, "items_running": 0, "display_running": 0, "items_completed": 0, "items_blacklisted": 0, "items_failed": 0, "metrics_source": "central_queue", }, ], "step_stats": [], } runtime_activity = { "processed_recent": 52, "completed_recent": 0, "failed_recent": 0, "blacklisted_recent": 0, "focus_job_code": "sync-overseas-40", "job_codes": ["sync-overseas-40", "sync-overseas-14"], "step_code": "detect_register", "nodes": { "mainland-controller-01-k": { "node_code": "mainland-controller-01-k", "processed_recent": 20, "completed_recent": 20, "failed_recent": 0, "blacklisted_recent": 0, }, "mainland-controller-01-a": { "node_code": "mainland-controller-01-a", "processed_recent": 16, "completed_recent": 16, "failed_recent": 0, "blacklisted_recent": 0, }, "mainland-controller-01-s": { "node_code": "mainland-controller-01-s", "processed_recent": 16, "completed_recent": 16, "failed_recent": 0, "blacklisted_recent": 0, }, }, } runtime_snapshot = { "_created_at": datetime.now().isoformat(sep=" ", timespec="seconds"), "job": { "job_id": 551, "job_code": "sync-overseas-17149", "status": "running", "progress_percent": 41.2, }, "queue_health": { "queue": { "items_total": 4000, "pending": 0, "claimed": 2352, "running": 51, "display_claimed": 2352, "display_running": 51, "completed": 0, "blacklisted": 0, "failed": 0, }, "nodes": [ { "node_code": "mainland-controller-01-k", "items_total": 1200, "items_pending": 0, "items_claimed": 1179, "items_running": 20, "display_running": 20, "current_load": 20, "active_threads": 20, "max_threads": 900, "status": "busy", "last_heartbeat_at": datetime.now().isoformat(sep=" ", timespec="seconds"), }, { "node_code": "mainland-controller-01-a", "items_total": 1400, "items_pending": 0, "items_claimed": 1383, "items_running": 17, "display_running": 17, "current_load": 17, "active_threads": 17, "max_threads": 900, "status": "busy", "last_heartbeat_at": datetime.now().isoformat(sep=" ", timespec="seconds"), }, { "node_code": "mainland-controller-01-s", "items_total": 1400, "items_pending": 0, "items_claimed": 1395, "items_running": 14, "display_running": 14, "current_load": 14, "active_threads": 14, "max_threads": 900, "status": "busy", "last_heartbeat_at": datetime.now().isoformat(sep=" ", timespec="seconds"), }, ], "steps": [], }, } with patch("app.services.detect_job_service.settings.node_region", "overseas"): with patch("app.services.detect_job_service.settings.node_role", "control"): with patch("app.services.detect_job_service.settings.node_code", "overseas-control-01"): with patch("app.services.detect_job_service.get_active_detect_job_summary", return_value=active_job): with patch("app.services.detect_job_service._load_runtime_activity_snapshot", return_value=runtime_activity): with patch("app.services.detect_job_service._load_latest_runtime_active_job_snapshot", return_value=runtime_snapshot): with patch("app.services.detect_job_service._load_runtime_display_rows", return_value=runtime_display_rows): with patch("app.services.detect_job_service.get_db", return_value=db_cm): health = get_detect_queue_health(window_minutes=15) node_codes = [item["node_code"] for item in health["nodes"]] self.assertEqual("detect-20260421232649-acfa3d", health["job"]["job_code"]) self.assertEqual(2, health["queue"]["running"]) self.assertEqual(2, health["queue"]["display_running"]) self.assertCountEqual(["mainland-worker-01", "mainland-controller-01", "unassigned"], node_codes) self.assertNotIn("mainland-controller-01-k", node_codes) def test_load_runtime_activity_snapshot_prefers_recent_runtime_snapshot_job_code(self) -> None: event_rows = [ ( "mainland-worker-01-a", "domain_completed", "", {"job_code": "sync-overseas-198", "detect_key": "detect_register"}, datetime(2026, 4, 23, 22, 8, 0), ), ( "mainland-controller-01-a", "domain_started", "", {"job_code": "sync-overseas-599", "detect_key": "detect_wayback"}, datetime(2026, 4, 23, 22, 12, 0), ), ( "mainland-controller-01-a", "domain_failed", "", {"job_code": "sync-overseas-599", "detect_key": "detect_wayback"}, datetime(2026, 4, 23, 22, 12, 5), ), ] fake_conn = _SequenceConn(fetchall_results=[event_rows]) with patch.object(detect_job_service_module.settings, "node_region", "overseas"): with patch.object(detect_job_service_module.settings, "node_role", "control"): with patch("app.services.detect_job_service.get_db", return_value=fake_conn): with patch( "app.services.detect_job_service._load_recent_runtime_snapshot_job_codes", return_value=["sync-overseas-599"], ): snapshot = detect_job_service_module._load_runtime_activity_snapshot(15) self.assertEqual("sync-overseas-599", snapshot["focus_job_code"]) self.assertEqual(["sync-overseas-599", "sync-overseas-198"], snapshot["job_codes"]) self.assertEqual(1, snapshot["processed_recent"]) self.assertEqual(1, snapshot["failed_recent"]) self.assertEqual(0, snapshot["completed_recent"]) self.assertEqual("detect_wayback", snapshot["step_code"]) self.assertIn("mainland-controller-01-a", snapshot["nodes"]) self.assertNotIn("mainland-worker-01-a", snapshot["nodes"]) def test_normalize_detect_step_code_accepts_supported_single_step(self) -> None: self.assertEqual("detect_baidu_site", normalize_detect_step_code("detect_baidu_site")) self.assertEqual("detect_wayback", normalize_detect_step_code("detect_wayback")) def test_normalize_detect_step_code_rejects_unknown_step(self) -> None: self.assertEqual("", normalize_detect_step_code("detect_unknown")) def test_resolve_detect_job_definition_builds_single_step_job(self) -> None: definition = resolve_detect_job_definition("detect_baidu_site") self.assertTrue(definition["is_single_step"]) self.assertEqual("single_step", definition["task_mode"]) self.assertEqual("detect_baidu_site", definition["step_code"]) def test_resolve_detect_job_definition_builds_wayback_single_step_job(self) -> None: definition = resolve_detect_job_definition("detect_wayback") self.assertTrue(definition["is_single_step"]) self.assertEqual("single_step", definition["task_mode"]) self.assertEqual("detect_wayback", definition["step_code"]) def test_build_step_payload_adds_wayback_recent_years_strategy(self) -> None: payload = _build_step_payload( step_code="detect_wayback", domain_snapshot={"domain": "example.com", "source_type": 2}, settings_payload={"detect_options": {"detect_wayback": True}}, ) self.assertEqual("detect_wayback", payload["step_code"]) self.assertEqual("recent_years", payload["wayback_strategy"]) self.assertEqual(5, payload["wayback_recent_years"]) self.assertTrue(payload["wayback_stop_on_first_hit"]) def test_resolve_detect_job_definition_defaults_to_domain_pipeline(self) -> None: definition = resolve_detect_job_definition(None) self.assertFalse(definition["is_single_step"]) self.assertEqual("domain_pipeline", definition["task_mode"]) self.assertEqual("", definition["step_code"]) def test_resolve_domain_pipeline_step_skips_yikoujia_register(self) -> None: step_code = resolve_domain_pipeline_step( { "source_type": 1, "register_status": 0, "baidu_site": {}, "qihu360_site": {}, "chinaz_info": {}, "aizhan_info": {}, "wayback_info": {}, "jucha_info": {}, "juziseo_info": {}, }, settings_payload={ "detect_options": { "detect_register": True, "detect_baidu_site": True, "detect_360_site": False, "detect_chinaz": False, "detect_aizhan": False, "detect_wayback": False, "detect_jucha": False, "detect_juziseo": False, "detect_order": ["detect_register", "detect_baidu_site"], } }, ) self.assertEqual("detect_baidu_site", step_code) def test_resolve_domain_pipeline_step_moves_to_next_incomplete_step(self) -> None: step_code = resolve_domain_pipeline_step( { "source_type": 2, "register_status": 3, "baidu_site": {"state": "passed"}, "qihu360_site": {}, "chinaz_info": {}, "aizhan_info": {}, "wayback_info": {}, "jucha_info": {}, "juziseo_info": {}, }, settings_payload={ "detect_options": { "detect_register": True, "detect_baidu_site": True, "detect_360_site": True, "detect_chinaz": False, "detect_aizhan": False, "detect_wayback": False, "detect_jucha": False, "detect_juziseo": False, "detect_order": ["detect_register", "detect_baidu_site", "detect_360_site"], } }, after_step_code="detect_baidu_site", ) self.assertEqual("detect_360_site", step_code) def test_resolve_initial_domain_pipeline_item_builds_first_step_payload(self) -> None: step_code, payload = resolve_initial_domain_pipeline_item( { "id": 10, "domain": "example.com", "source_type": 2, "register_status": 0, "baidu_site": {}, "qihu360_site": {}, "chinaz_info": {}, "aizhan_info": {}, "wayback_info": {}, "jucha_info": {}, "juziseo_info": {}, }, settings_payload={ "detect_options": { "detect_register": True, "detect_baidu_site": True, "detect_order": ["detect_register", "detect_baidu_site"], } }, ) self.assertEqual("detect_register", step_code) self.assertIsNotNone(payload) self.assertEqual("detect_register", payload["step_code"]) self.assertEqual("example.com", payload["domain"]) def test_resolve_initial_domain_pipeline_item_returns_empty_when_pipeline_already_done(self) -> None: step_code, payload = resolve_initial_domain_pipeline_item( { "id": 11, "domain": "done.com", "source_type": 2, "register_status": 3, "baidu_site": {"state": "passed"}, "qihu360_site": {}, "chinaz_info": {}, "aizhan_info": {}, "wayback_info": {}, "jucha_info": {}, "juziseo_info": {}, }, settings_payload={ "detect_options": { "detect_register": True, "detect_baidu_site": True, "detect_order": ["detect_register", "detect_baidu_site"], } }, ) self.assertEqual("", step_code) self.assertIsNone(payload) def test_classify_pipeline_item_outcome_retries_external_failure(self) -> None: outcome = _classify_pipeline_item_outcome( item_status="failed", result_payload={"state": "degraded", "message": "timeout", "retry_recommended": True}, step_code="detect_baidu_site", attempt_count=0, ) self.assertEqual("retry", outcome["action"]) self.assertTrue(outcome["should_retry"]) self.assertEqual("external_retry", outcome["reason_code"]) def test_classify_pipeline_item_outcome_rejects_business_failure(self) -> None: outcome = _classify_pipeline_item_outcome( item_status="failed", result_payload={"state": "rejected", "message": "title contains forbidden keyword"}, step_code="detect_chinaz", attempt_count=0, ) self.assertEqual("reject", outcome["action"]) self.assertFalse(outcome["should_retry"]) self.assertEqual("business_reject", outcome["reason_code"]) def test_classify_pipeline_item_outcome_marks_blacklisted_terminal(self) -> None: outcome = _classify_pipeline_item_outcome( item_status="blacklisted", result_payload={"state": "blacklisted", "message": "risk hit"}, step_code="detect_baidu_site", attempt_count=0, ) self.assertEqual("black_hit", outcome["action"]) self.assertFalse(outcome["should_retry"]) self.assertEqual("blacklisted", outcome["reason_code"]) if __name__ == "__main__": unittest.main()