mirror of
https://github.com/blw1138/Zordon.git
synced 2026-06-09 13:39:24 -05:00
Add Unit Tests (#132)
* Add tests and new github workflow * Add new unit tests * Add Github CI workflow * Workflow fix * Add pytest install to workflow file * More CI / test updates * More test cleanup * Whitespace cleanup and link complexity override * More whitespace cleanup * Make lint less strict * More lint tweaks
This commit is contained in:
@@ -0,0 +1,169 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from src.api.preview_manager import PreviewManager
|
||||
|
||||
|
||||
def make_mock_job(job_id='test-job-1', input_path='/tmp/test.blend',
|
||||
file_list=None, **kwargs):
|
||||
job = MagicMock()
|
||||
job.id = job_id
|
||||
job.input_path = input_path
|
||||
job.file_list.return_value = file_list or []
|
||||
for k, v in kwargs.items():
|
||||
setattr(job, k, v)
|
||||
return job
|
||||
|
||||
|
||||
class TestPreviewManagerDefaults:
|
||||
"""Default state."""
|
||||
|
||||
def test_storage_path_none_initially(self, preview_manager_instance):
|
||||
assert preview_manager_instance.storage_path is not None
|
||||
|
||||
def test_running_jobs_empty_initially(self, preview_manager_instance):
|
||||
assert preview_manager_instance._running_jobs == {}
|
||||
|
||||
|
||||
class TestGeneratePreviewWorker:
|
||||
"""Core preview generation logic."""
|
||||
|
||||
def test_skips_when_no_supported_files(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job(file_list=[str(tmp_path / 'output.txt')])
|
||||
|
||||
with patch.object(preview_manager_instance, '_generate_job_preview_worker') as mock_gen:
|
||||
preview_manager_instance._update_previews_for_job(job)
|
||||
mock_gen.assert_called_once_with(job, False)
|
||||
|
||||
def test_uses_input_path_when_no_file_list(self, preview_manager_instance, tmp_path):
|
||||
"""When file_list is empty, falls back to input_path."""
|
||||
job = make_mock_job(input_path=str(tmp_path / 'output.mp4'))
|
||||
|
||||
with patch.object(preview_manager_instance, '_generate_job_preview_worker') as mock_gen:
|
||||
preview_manager_instance._update_previews_for_job(job)
|
||||
mock_gen.assert_called_once()
|
||||
|
||||
def test_generate_preview_checks_existing_files(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job(input_path=str(tmp_path / 'test.jpg'))
|
||||
|
||||
with patch('src.api.preview_manager.save_first_frame') as mock_save:
|
||||
with patch('src.api.preview_manager.os.path.exists', return_value=False):
|
||||
preview_manager_instance._generate_job_preview_worker(job)
|
||||
|
||||
# No file_list → falls back to input_path → label is "input"
|
||||
expected_img = str(tmp_path / f'{job.id}-input-480.jpg')
|
||||
mock_save.assert_called_once_with(
|
||||
source_path=job.input_path,
|
||||
dest_path=expected_img,
|
||||
max_width=480,
|
||||
)
|
||||
|
||||
|
||||
class TestUpdatePreviewsForJob:
|
||||
"""Dispatch of preview generation."""
|
||||
|
||||
def test_starts_new_thread(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job(file_list=[str(tmp_path / 'test.mp4')])
|
||||
|
||||
with patch('src.api.preview_manager.threading.Thread') as mock_thread:
|
||||
thread_instance = MagicMock()
|
||||
mock_thread.return_value = thread_instance
|
||||
|
||||
preview_manager_instance._update_previews_for_job(job)
|
||||
|
||||
mock_thread.assert_called_once()
|
||||
thread_instance.start.assert_called_once()
|
||||
assert preview_manager_instance._running_jobs[job.id] == thread_instance
|
||||
|
||||
def test_reuses_existing_thread(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job()
|
||||
|
||||
existing_thread = MagicMock()
|
||||
existing_thread.is_alive.return_value = True
|
||||
preview_manager_instance._running_jobs[job.id] = existing_thread
|
||||
|
||||
with patch('src.api.preview_manager.threading.Thread') as mock_thread:
|
||||
preview_manager_instance._update_previews_for_job(job)
|
||||
mock_thread.assert_not_called()
|
||||
|
||||
def test_join_when_wait_requested(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job()
|
||||
|
||||
with patch('src.api.preview_manager.threading.Thread') as mock_thread:
|
||||
thread_instance = MagicMock()
|
||||
mock_thread.return_value = thread_instance
|
||||
|
||||
preview_manager_instance._update_previews_for_job(
|
||||
job, wait_until_completion=True, timeout=30,
|
||||
)
|
||||
thread_instance.join.assert_called_once_with(timeout=30)
|
||||
|
||||
|
||||
class TestGetPreviewsForJob:
|
||||
"""Reading preview files."""
|
||||
|
||||
def test_returns_empty_when_no_previews(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job()
|
||||
|
||||
result = preview_manager_instance._get_previews_for_job(job)
|
||||
assert result == {}
|
||||
|
||||
def test_returns_preview_files(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job(job_id='abc')
|
||||
|
||||
# Create a fake preview file
|
||||
(tmp_path / 'abc-output-480.jpg').write_text('preview')
|
||||
|
||||
result = preview_manager_instance._get_previews_for_job(job)
|
||||
assert 'output' in result
|
||||
assert len(result['output']) == 1
|
||||
assert result['output'][0]['kind'] == 'image'
|
||||
|
||||
|
||||
class TestDeletePreviewsForJob:
|
||||
"""Cleaning up preview files."""
|
||||
|
||||
def test_deletes_existing_previews(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job(job_id='abc')
|
||||
|
||||
preview_file = tmp_path / 'abc-output-480.jpg'
|
||||
preview_file.write_text('preview')
|
||||
|
||||
preview_manager_instance._delete_previews_for_job(job)
|
||||
assert not preview_file.exists()
|
||||
|
||||
def test_no_error_when_no_previews(self, preview_manager_instance, tmp_path):
|
||||
preview_manager_instance.storage_path = str(tmp_path)
|
||||
job = make_mock_job()
|
||||
preview_manager_instance._delete_previews_for_job(job)
|
||||
|
||||
|
||||
class TestForwarders:
|
||||
"""Classmethod forwarders delegate to instance."""
|
||||
|
||||
def test_update_previews_for_job_forwarder(self, preview_manager_instance):
|
||||
job = make_mock_job()
|
||||
with patch.object(preview_manager_instance, '_update_previews_for_job') as mock_method:
|
||||
PreviewManager.update_previews_for_job(job)
|
||||
mock_method.assert_called_once_with(job, False, False, None)
|
||||
|
||||
def test_get_previews_for_job_forwarder(self, preview_manager_instance):
|
||||
job = make_mock_job()
|
||||
with patch.object(preview_manager_instance, '_get_previews_for_job',
|
||||
return_value={'input': []}) as mock_method:
|
||||
result = PreviewManager.get_previews_for_job(job)
|
||||
assert result == {'input': []}
|
||||
mock_method.assert_called_once_with(job)
|
||||
|
||||
def test_delete_previews_for_job_forwarder(self, preview_manager_instance):
|
||||
job = make_mock_job()
|
||||
with patch.object(preview_manager_instance, '_delete_previews_for_job') as mock_method:
|
||||
PreviewManager.delete_previews_for_job(job)
|
||||
mock_method.assert_called_once_with(job)
|
||||
Reference in New Issue
Block a user