47 lines
1.9 KiB
Python
47 lines
1.9 KiB
Python
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()
|