66 lines
2.7 KiB
Python
66 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import app.services.detect_run_service as detect_run_service
|
|
|
|
|
|
class _TraceLock:
|
|
def __init__(self, order: list[str]) -> None:
|
|
self.order = order
|
|
|
|
def __enter__(self):
|
|
self.order.append("enter")
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
self.order.append("exit")
|
|
return False
|
|
|
|
|
|
class DetectRunServiceTests(unittest.TestCase):
|
|
def test_create_detect_run_snapshot_acquires_lock_before_loading_records(self) -> None:
|
|
order: list[str] = []
|
|
|
|
with patch.object(detect_run_service, "_DETECT_RUNS_LOCK", _TraceLock(order)):
|
|
with patch.object(detect_run_service, "_load", side_effect=lambda: order.append("load") or []):
|
|
with patch.object(detect_run_service, "_capture_worker_logs", return_value=[]):
|
|
with patch.object(detect_run_service, "_save", side_effect=lambda records: order.append("save")):
|
|
record = detect_run_service.create_detect_run_snapshot(
|
|
"start",
|
|
{"latest_start_time": "2026-04-23 15:00:00", "running": True},
|
|
{"running": 1},
|
|
{"thread_count": 1000},
|
|
)
|
|
|
|
self.assertEqual("enter", order[0])
|
|
self.assertIn("load", order)
|
|
self.assertIn("save", order)
|
|
self.assertEqual("exit", order[-1])
|
|
self.assertEqual("starting", record["status"])
|
|
|
|
def test_sync_detect_runs_acquires_lock_before_mutating_records(self) -> None:
|
|
order: list[str] = []
|
|
|
|
with patch.object(detect_run_service, "_DETECT_RUNS_LOCK", _TraceLock(order)):
|
|
with patch.object(detect_run_service, "_load", side_effect=lambda: order.append("load") or []):
|
|
with patch.object(detect_run_service, "_capture_worker_logs", return_value=[]):
|
|
with patch.object(detect_run_service, "_save", side_effect=lambda records: order.append("save")):
|
|
records = detect_run_service.sync_detect_runs(
|
|
{"running": True, "detecting": True, "latest_start_time": "2026-04-23 15:00:00"},
|
|
{"running": 1, "pending": 0},
|
|
{"thread_count": 1000},
|
|
active_job={"status": "running", "items_running": 1},
|
|
)
|
|
|
|
self.assertEqual("enter", order[0])
|
|
self.assertIn("load", order)
|
|
self.assertIn("save", order)
|
|
self.assertEqual("exit", order[-1])
|
|
self.assertEqual("running", records[0]["status"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|