Zordon/tests/test_preview_manager.py

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)