迁移目录

docker/crawler/tests/test_runtime_guard.py

@@ -0,0 +1,46 @@
+import importlib.util
+import pathlib
+import sys
+import tempfile
+import types
+import unittest
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+sys.modules.setdefault("src.core.alert_manager", types.ModuleType("src.core.alert_manager"))
+sys.modules["src.core.alert_manager"].CrontabAlertManager = object
+sys.modules.setdefault("src.core.config", types.ModuleType("src.core.config"))
+sys.modules["src.core.config"].load_config = lambda: None
+sys.modules.setdefault("src.core.error_collector", types.ModuleType("src.core.error_collector"))
+sys.modules["src.core.error_collector"].ErrorCollector = object
+sys.modules.setdefault("src.core.logger", types.ModuleType("src.core.logger"))
+sys.modules["src.core.logger"].setup_logger = lambda: None
+SPEC = importlib.util.spec_from_file_location("crawler_main", ROOT / "main.py")
+crawler_main = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(crawler_main)
+
+
+class RuntimeGuardTest(unittest.TestCase):
+    def test_action_lock_blocks_duplicate_running_action(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            first = crawler_main.CrawlerActionLock("nba_match", pathlib.Path(tmp))
+            second = crawler_main.CrawlerActionLock("nba_match", pathlib.Path(tmp))
+
+            self.assertTrue(first.acquire())
+            try:
+                self.assertFalse(second.acquire())
+            finally:
+                first.release()
+
+            self.assertTrue(second.acquire())
+            second.release()
+
+    def test_action_timeout_uses_specific_value_or_default(self):
+        self.assertEqual(crawler_main.get_action_timeout("nba_match"), 180)
+        self.assertEqual(crawler_main.get_action_timeout("live_detail"), 180)
+        self.assertEqual(crawler_main.get_action_timeout("article_content_fetch"), 900)
+        self.assertEqual(crawler_main.get_action_timeout("unknown_task"), 600)
+
+
+if __name__ == "__main__":
+    unittest.main()