video-processor/tests/test_procrastinate_migration.py

"""Tests for Procrastinate migration utilities."""

import pytest

from video_processor.tasks.migration import ProcrastinateMigrationHelper, create_migration_script
from video_processor.tasks.compat import IS_PROCRASTINATE_3_PLUS


class TestProcrastinateMigrationHelper:
    """Test migration helper functionality."""

    def test_migration_helper_creation(self):
        """Test migration helper initialization."""
        database_url = "postgresql://test:test@localhost/test"
        helper = ProcrastinateMigrationHelper(database_url)
        
        assert helper.database_url == database_url
        assert helper.version_info is not None
        assert "procrastinate_version" in helper.version_info

    def test_migration_steps_generation(self):
        """Test migration steps generation."""
        helper = ProcrastinateMigrationHelper("postgresql://fake/db")
        steps = helper.get_migration_steps()
        
        assert isinstance(steps, list)
        assert len(steps) > 0
        
        if IS_PROCRASTINATE_3_PLUS:
            # Should have pre/post migration steps
            assert len(steps) >= 7  # Pre, deploy, post, verify
            assert any("pre-migration" in step.lower() for step in steps)
            assert any("post-migration" in step.lower() for step in steps)
        else:
            # Should have single migration step
            assert len(steps) >= 2  # Migrate, verify
            assert any("migration" in step.lower() for step in steps)

    def test_print_migration_plan(self, capsys):
        """Test migration plan printing."""
        helper = ProcrastinateMigrationHelper("postgresql://fake/db")
        helper.print_migration_plan()
        
        captured = capsys.readouterr()
        assert "Procrastinate Migration Plan" in captured.out
        assert "Version Info:" in captured.out
        assert "Current Version:" in captured.out

    def test_migration_command_structure(self):
        """Test that migration commands have correct structure."""
        helper = ProcrastinateMigrationHelper("postgresql://fake/db")
        
        # Test method availability
        assert hasattr(helper, 'apply_pre_migration')
        assert hasattr(helper, 'apply_post_migration')
        assert hasattr(helper, 'apply_legacy_migration')
        assert hasattr(helper, 'check_schema')
        assert hasattr(helper, 'run_migration_command')

    def test_migration_command_validation(self):
        """Test migration command validation without actually running."""
        helper = ProcrastinateMigrationHelper("postgresql://fake/db")
        
        # Test that methods return appropriate responses for invalid DB
        if IS_PROCRASTINATE_3_PLUS:
            # Pre-migration should be available
            assert hasattr(helper, 'apply_pre_migration')
            assert hasattr(helper, 'apply_post_migration')
        else:
            # Legacy migration should be available
            assert hasattr(helper, 'apply_legacy_migration')


class TestMigrationScriptGeneration:
    """Test migration script generation."""

    def test_script_generation(self):
        """Test that migration script is generated correctly."""
        script_content = create_migration_script()
        
        assert isinstance(script_content, str)
        assert len(script_content) > 0
        
        # Check for essential script components
        assert "#!/usr/bin/env python3" in script_content
        assert "Procrastinate migration script" in script_content
        assert "migrate_database" in script_content
        assert "asyncio" in script_content
        
        # Check for command line argument handling
        assert "--pre" in script_content or "--post" in script_content

    def test_script_has_proper_structure(self):
        """Test that generated script has proper Python structure."""
        script_content = create_migration_script()
        
        # Should have proper Python script structure
        lines = script_content.split('\n')
        
        # Check shebang
        assert lines[0] == "#!/usr/bin/env python3"
        
        # Check for main function
        assert 'def main():' in script_content
        
        # Check for asyncio usage
        assert 'asyncio.run(main())' in script_content


class TestMigrationWorkflow:
    """Test complete migration workflow scenarios."""

    def test_version_aware_migration_selection(self):
        """Test that correct migration path is selected based on version."""
        helper = ProcrastinateMigrationHelper("postgresql://fake/db")
        
        if IS_PROCRASTINATE_3_PLUS:
            # 3.x should use pre/post migrations
            steps = helper.get_migration_steps()
            step_text = ' '.join(steps).lower()
            assert 'pre-migration' in step_text
            assert 'post-migration' in step_text
        else:
            # 2.x should use legacy migration
            steps = helper.get_migration_steps()
            step_text = ' '.join(steps).lower()
            assert 'migration' in step_text
            assert 'pre-migration' not in step_text

    def test_migration_helper_consistency(self):
        """Test that migration helper provides consistent information."""
        helper = ProcrastinateMigrationHelper("postgresql://fake/db")
        
        # Version info should be consistent
        version_info = helper.version_info
        steps = helper.get_migration_steps()
        
        assert version_info["is_v3_plus"] == IS_PROCRASTINATE_3_PLUS
        
        # Steps should match version
        if version_info["is_v3_plus"]:
            assert len(steps) > 4  # Should have multiple steps for 3.x
        else:
            assert len(steps) >= 2  # Should have basic steps for 2.x


@pytest.mark.asyncio
class TestAsyncMigration:
    """Test async migration functionality."""

    async def test_migrate_database_function_exists(self):
        """Test that async migration function exists and is callable."""
        from video_processor.tasks.migration import migrate_database
        
        # Function should exist and be async
        assert callable(migrate_database)
        
        # Should handle invalid database gracefully (don't actually run)
        # Just test that it exists and has the right signature
        import inspect
        sig = inspect.signature(migrate_database)
        
        expected_params = ['database_url', 'pre_migration_only', 'post_migration_only']
        actual_params = list(sig.parameters.keys())
        
        for param in expected_params:
            assert param in actual_params


class TestRegressionPrevention:
    """Tests to prevent regressions in migration functionality."""

    def test_migration_helper_backwards_compatibility(self):
        """Ensure migration helper maintains backwards compatibility."""
        helper = ProcrastinateMigrationHelper("postgresql://fake/db")
        
        # Essential methods should always exist
        required_methods = [
            'get_migration_steps',
            'print_migration_plan',
            'run_migration_command',
            'check_schema',
        ]
        
        for method in required_methods:
            assert hasattr(helper, method)
            assert callable(getattr(helper, method))

    def test_version_detection_stability(self):
        """Test that version detection is stable and predictable."""
        from video_processor.tasks.compat import get_version_info, PROCRASTINATE_VERSION
        
        info1 = get_version_info()
        info2 = get_version_info()
        
        # Should return consistent results
        assert info1 == info2
        assert info1["version_tuple"] == PROCRASTINATE_VERSION

    def test_feature_flags_consistency(self):
        """Test that feature flags are consistent with version."""
        from video_processor.tasks.compat import FEATURES, IS_PROCRASTINATE_3_PLUS
        
        # 3.x features should only be available in 3.x
        v3_features = [
            "graceful_shutdown", 
            "job_cancellation", 
            "pre_post_migrations",
            "psycopg3_support"
        ]
        
        for feature in v3_features:
            if IS_PROCRASTINATE_3_PLUS:
                assert FEATURES[feature] is True, f"{feature} should be True in 3.x"
            else:
                assert FEATURES[feature] is False, f"{feature} should be False in 2.x"