import threading
|
from unittest.mock import MagicMock, patch
|
|
from label_studio.core.redis import is_job_on_worker
|
|
|
def test_is_job_on_worker_does_not_call_get_job_ids():
|
"""Ensure membership check avoids StartedJobRegistry.get_job_ids, preventing signal usage in threads."""
|
fake_connection = MagicMock()
|
fake_connection.zscore.return_value = None
|
|
with patch('label_studio.core.redis.StartedJobRegistry') as registry_cls:
|
registry = registry_cls.return_value
|
registry.connection = fake_connection
|
registry.key = 'rq:started:low'
|
registry.get_job_ids.side_effect = ValueError('should not be called')
|
|
assert is_job_on_worker('job123', 'low') is False
|
|
fake_connection.zscore.assert_called_once_with('rq:started:low', b'job123')
|
|
# Ensure None job IDs short-circuit without touching Redis.
|
fake_connection.zscore.reset_mock()
|
assert is_job_on_worker(None, 'low') is False
|
fake_connection.zscore.assert_not_called()
|
|
|
def test_is_job_on_worker_safe_from_non_main_thread(monkeypatch):
|
"""Simulate the original failure: registry.get_job_ids would raise when used from non-main threads."""
|
import signal
|
|
original_signal = signal.signal
|
|
def fake_signal(sig, handler):
|
if threading.current_thread() is not threading.main_thread():
|
raise ValueError('signal only works in main thread of the main interpreter')
|
return original_signal(sig, handler)
|
|
monkeypatch.setattr(signal, 'signal', fake_signal)
|
|
fake_connection = MagicMock()
|
fake_connection.zscore.return_value = None
|
|
class DummyRegistry:
|
def __init__(self, queue_name, connection):
|
self.connection = fake_connection
|
self.key = f'rq:started:{queue_name}'
|
|
def get_job_ids(self):
|
# The old implementation would call this, which uses signal and fails in threads
|
signal.signal(signal.SIGALRM, lambda *args: None)
|
return []
|
|
with patch('label_studio.core.redis.StartedJobRegistry', DummyRegistry):
|
result: dict[str, object] = {}
|
|
def runner():
|
try:
|
result['value'] = is_job_on_worker('job123', 'low')
|
except Exception as exc: # pragma: no cover - used for regression verification
|
result['error'] = exc
|
|
t = threading.Thread(target=runner)
|
t.start()
|
t.join()
|
|
if 'error' in result:
|
raise result['error'] # type: ignore[misc]
|
|
assert result['value'] is False
|
