95 lines
3.9 KiB
Python
95 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from encar_scraper.worker import tasks
|
|
|
|
|
|
class TestWorkerTaskLockHelpers(unittest.TestCase):
|
|
def test_acquire_lock_returns_true_on_success(self) -> None:
|
|
redis_client = MagicMock()
|
|
redis_client.set.return_value = True
|
|
|
|
acquired = tasks._acquire_lock(redis_client, "lock:key", "owner-token", 120)
|
|
|
|
self.assertTrue(acquired)
|
|
redis_client.set.assert_called_once_with("lock:key", "owner-token", nx=True, ex=120)
|
|
|
|
def test_refresh_lock_if_owner_extends_ttl(self) -> None:
|
|
redis_client = MagicMock()
|
|
mock_script = MagicMock(return_value=1)
|
|
redis_client.register_script.return_value = mock_script
|
|
|
|
refreshed = tasks._refresh_lock_if_owner(redis_client, "lock:key", "owner-token", 120)
|
|
|
|
self.assertTrue(refreshed)
|
|
redis_client.register_script.assert_called_once()
|
|
mock_script.assert_called_once_with(keys=["lock:key"], args=["owner-token", 120])
|
|
|
|
def test_release_lock_if_owner_uses_owner_token(self) -> None:
|
|
redis_client = MagicMock()
|
|
mock_script = MagicMock(return_value=1)
|
|
redis_client.register_script.return_value = mock_script
|
|
|
|
tasks._release_lock_if_owner(redis_client, "lock:key", "owner-token")
|
|
|
|
redis_client.register_script.assert_called_once()
|
|
mock_script.assert_called_once_with(keys=["lock:key"], args=["owner-token"])
|
|
|
|
def test_encar_sync_listing_task_skips_when_lock_not_acquired(self) -> None:
|
|
with patch.object(tasks, "_get_persistence") as get_persistence, \
|
|
patch.object(tasks, "_get_redis") as get_redis, \
|
|
patch.object(tasks, "_acquire_lock", return_value=False):
|
|
persistence = MagicMock()
|
|
get_persistence.return_value = persistence
|
|
get_redis.return_value = MagicMock()
|
|
|
|
tasks.encar_sync_listing_task.push_request(id="task-123")
|
|
try:
|
|
result = tasks.encar_sync_listing_task.run()
|
|
finally:
|
|
tasks.encar_sync_listing_task.pop_request()
|
|
|
|
persistence.create_tables.assert_called_once()
|
|
self.assertEqual(result["status"], "skipped")
|
|
self.assertEqual(result["reason"], "sync_already_running")
|
|
|
|
def test_encar_sync_listing_task_releases_owned_lock(self) -> None:
|
|
mock_scraper_instance = MagicMock()
|
|
mock_scraper_instance.sync_listing.return_value = {
|
|
"total_available": 100,
|
|
"items_collected": 10,
|
|
"cars_synced": 8,
|
|
"cars_failed": 2,
|
|
}
|
|
|
|
with patch.object(tasks, "_get_persistence") as get_persistence, \
|
|
patch.object(tasks, "_get_redis") as get_redis, \
|
|
patch.object(tasks, "_acquire_lock", return_value=True), \
|
|
patch.object(tasks, "_start_lock_heartbeat") as start_heartbeat, \
|
|
patch.object(tasks, "_release_lock_if_owner") as release_lock, \
|
|
patch("encar_scraper.worker.tasks.EncarScraper", return_value=mock_scraper_instance):
|
|
persistence = MagicMock()
|
|
get_persistence.return_value = persistence
|
|
redis_client = MagicMock()
|
|
get_redis.return_value = redis_client
|
|
stop_event = MagicMock()
|
|
heartbeat_thread = MagicMock()
|
|
start_heartbeat.return_value = (stop_event, heartbeat_thread)
|
|
|
|
tasks.encar_sync_listing_task.push_request(id="task-123")
|
|
try:
|
|
with patch.object(tasks.encar_sync_listing_task, "update_state"):
|
|
result = tasks.encar_sync_listing_task.run(car_type="all")
|
|
finally:
|
|
tasks.encar_sync_listing_task.pop_request()
|
|
|
|
self.assertEqual(result["status"], "success")
|
|
stop_event.set.assert_called_once()
|
|
heartbeat_thread.join.assert_called_once()
|
|
release_lock.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |