qananasikq/encar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add tests

Browse Source
This commit is contained in:
qananasikq
2026-04-16 18:01:37 +03:00
parent 8140fb6f41
commit 80f7a1353d
5 changed files with 520 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
tests/test_worker_tasks.py
View File

@@ -1,9 +1,9 @@
from __future__ import annotations
from __future__ import annotations
import unittest
from unittest.mock import MagicMock, patch
from iaai_scraper.worker import tasks
from encar_scraper.worker import tasks
class TestWorkerTaskLockHelpers(unittest.TestCase):
@@ -18,25 +18,26 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
def test_refresh_lock_if_owner_extends_ttl(self) -> None:
redis_client = MagicMock()
redis_client.eval.return_value = 1
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.eval.assert_called_once()
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.eval.assert_called_once()
args = redis_client.eval.call_args[0]
self.assertEqual(args[1], 1)
self.assertEqual(args[2], "lock:key")
self.assertEqual(args[3], "owner-token")
redis_client.register_script.assert_called_once()
mock_script.assert_called_once_with(keys=["lock:key"], args=["owner-token"])
def test_sync_listing_task_skips_when_lock_not_acquired(self) -> None:
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):
@@ -44,30 +45,31 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
get_persistence.return_value = persistence
get_redis.return_value = MagicMock()
tasks.sync_listing_task.push_request(id="task-123")
tasks.encar_sync_listing_task.push_request(id="task-123")
try:
result = tasks.sync_listing_task.run()
result = tasks.encar_sync_listing_task.run()
finally:
tasks.sync_listing_task.pop_request()
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_sync_listing_task_releases_owned_lock(self) -> None:
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.object(tasks, "_run_browser_job", return_value={
"run_id": 7,
"cars_upserted": 2,
"cars_failed": 0,
"images_upserted": 4,
"skipped_existing": 1,
"elapsed_seconds": 1.25,
}):
patch("encar_scraper.worker.tasks.EncarScraper", return_value=mock_scraper_instance):
persistence = MagicMock()
get_persistence.return_value = persistence
redis_client = MagicMock()
@@ -76,11 +78,12 @@ class TestWorkerTaskLockHelpers(unittest.TestCase):
heartbeat_thread = MagicMock()
start_heartbeat.return_value = (stop_event, heartbeat_thread)
tasks.sync_listing_task.push_request(id="task-123")
tasks.encar_sync_listing_task.push_request(id="task-123")
try:
result = tasks.sync_listing_task.run(make="Toyota")
with patch.object(tasks.encar_sync_listing_task, "update_state"):
result = tasks.encar_sync_listing_task.run(car_type="all")
finally:
tasks.sync_listing_task.pop_request()
tasks.encar_sync_listing_task.pop_request()
self.assertEqual(result["status"], "success")
stop_event.set.assert_called_once()