MPT-17670: add --check option to validate duplicate migration_id in migrat…

README.md

@@ -12,7 +12,6 @@ to manage both schema and data migrations across multiple backends, ensuring con
 
 📚 **[Complete Usage Guide](docs/PROJECT_DESCRIPTION.md)**
 
-
 ## Getting started
 
 ### Prerequisites

docs/PROJECT_DESCRIPTION.md

@@ -135,6 +135,28 @@ class Migration(DataBaseMigration, MPTAPIClientMixin, AirtableAPIClientMixin):
         self.log.info(f"Processed {len(records)} records")
 ```
 
+### Checking Migrations
+Before running migrations, you can validate your migration folder for issues:
+
+```bash
+  mpt-tool migrate --check
+```
+
+This command:
+- Verifies the migration folder structure
+- Detects duplicate migration_id values (which could happen if migrations were created with the same name)
+- Exits with code 0 if all checks pass
+- Exits with code 1 and shows a detailed error message if duplicates are found
+
+**Example output when duplicates are found:**
+
+```bash
+Checking migrations...
+Error running check command: Duplicate migration_id found in migrations: 20260113180013_duplicate_name.py, 20260114190014_duplicate_name.py
+```
+
+**Best Practice:** Run `--check` as part of your CI/CD pipeline to catch migration issues before deployment.
+
 ### Running Migrations
 - **Run all pending data migrations:**
   ```bash
@@ -240,6 +262,11 @@ Run `mpt-tool --help` to see all available commands and params:
 
 ## Best Practices
 
+### Migration Validation
+- Run `mpt-tool migrate --check` before committing migration files
+- Include `--check` in your CI/CD pipeline to catch issues early
+- Verify there are no duplicate migration_id values before deployment
+
 ### Migration Naming
 - Use descriptive, snake_case names (e.g., `add_user_table`, `fix_null_emails`, `sync_agreements_from_api`)
 - Keep names concise but meaningful

mpt_tool/cli.py

@@ -18,6 +18,9 @@ def callback() -> None:
 @app.command("migrate")
 def migrate(  # noqa: WPS211
     ctx: typer.Context,
+    check: Annotated[  # noqa: FBT002
+        bool, typer.Option("--check", help="Check for duplicate migration_id in migrations.")
+    ] = False,
     data: Annotated[bool, typer.Option("--data", help="Run data migrations.")] = False,  # noqa: FBT002
     schema: Annotated[bool, typer.Option("--schema", help="Run schema migrations.")] = False,  # noqa: FBT002
     fake: Annotated[

mpt_tool/commands/check.py

@@ -0,0 +1,22 @@
+from typing import override
+
+from mpt_tool.commands.base import BaseCommand
+from mpt_tool.use_cases import CheckMigrationsUseCase
+
+
+class CheckCommand(BaseCommand):
+    """Checks migrations for duplicate migration_id."""
+
+    @override
+    @property
+    def start_message(self) -> str:
+        return "Checking migrations..."
+
+    @override
+    @property
+    def success_message(self) -> str:
+        return "Migrations check passed successfully."
+
+    @override
+    def run(self) -> None:
+        CheckMigrationsUseCase().execute()

mpt_tool/commands/factory.py

@@ -3,6 +3,7 @@
 from mpt_tool.commands.base import (
     BaseCommand,
 )
+from mpt_tool.commands.check import CheckCommand
 from mpt_tool.commands.data import DataCommand
 from mpt_tool.commands.errors import CommandNotFoundError
 from mpt_tool.commands.fake import FakeCommand
@@ -28,7 +29,9 @@ def get_instance(cls, param_data: dict[str, bool | str | None]) -> BaseCommand:
         Raises:
             CommandNotFoundError: If no command is found.
         """
-        match param_data:
+        match param_data:  # noqa: WPS242
+            case {"check": True}:
+                return CheckCommand()
             case {"data": True}:
                 return DataCommand()
             case {"schema": True}:

mpt_tool/use_cases/__init__.py

@@ -1,10 +1,12 @@
 from mpt_tool.use_cases.apply_migration import ApplyMigrationUseCase
+from mpt_tool.use_cases.check_migrations import CheckMigrationsUseCase
 from mpt_tool.use_cases.list_migrations import ListMigrationsUseCase
 from mpt_tool.use_cases.new_migration import NewMigrationUseCase
 from mpt_tool.use_cases.run_migrations import RunMigrationsUseCase
 
 __all__ = [
     "ApplyMigrationUseCase",
+    "CheckMigrationsUseCase",
     "ListMigrationsUseCase",
     "NewMigrationUseCase",
     "RunMigrationsUseCase",

mpt_tool/use_cases/check_migrations.py

@@ -0,0 +1,42 @@
+from collections import Counter
+
+from mpt_tool.managers import FileMigrationManager
+from mpt_tool.managers.errors import MigrationFolderError
+from mpt_tool.use_cases.errors import CheckMigrationError
+
+
+class CheckMigrationsUseCase:
+    """Use case for checking migrations for duplicate migration_id."""
+
+    def __init__(self, file_migration_manager: FileMigrationManager | None = None):
+        self.file_migration_manager = file_migration_manager or FileMigrationManager()
+
+    def execute(self) -> None:  # noqa: WPS210
+        """Check for duplicate migration_id in migration files.
+
+        Raises:
+            CheckMigrationError: If duplicate migration_id is found or a migration folder
+                error occurs.
+        """
+        try:
+            migration_files = self.file_migration_manager.retrieve_migration_files()
+        except MigrationFolderError as error:
+            raise CheckMigrationError(str(error)) from error
+
+        if not migration_files:
+            return
+
+        migration_id_counter = Counter(migration.migration_id for migration in migration_files)
+        duplicated_migration_ids = [
+            migration_id for migration_id, count in migration_id_counter.items() if count > 1
+        ]
+
+        if duplicated_migration_ids:
+            duplicate_migrations = [
+                migration.file_name
+                for migration in migration_files
+                if migration.migration_id in duplicated_migration_ids
+            ]
+            migrations_list = ", ".join(duplicate_migrations)
+            error_message = f"Duplicate migration_id found in migrations: {migrations_list}"
+            raise CheckMigrationError(error_message)

mpt_tool/use_cases/errors.py

@@ -5,13 +5,17 @@ class UseCaseError(BaseError):
     """Base error for use cases."""
 
 
+class ApplyMigrationError(UseCaseError):
+    """Error applying migration."""
+
+
+class CheckMigrationError(UseCaseError):
+    """Error checking migrations."""
+
+
 class NewMigrationError(UseCaseError):
     """Error creating new migration."""
 
 
 class RunMigrationError(UseCaseError):
     """Error running migration."""
-
-
-class ApplyMigrationError(UseCaseError):
-    """Error applying migration."""

tests/cli/test_cli.py

@@ -99,3 +99,35 @@ def test_migrate_list_no_migrations(runner):
 
     assert result.exit_code == 0, result.output
     assert "No migrations found." in result.output
+
+
+def test_migrate_check_no_duplicates(runner, migration_folder):
+    (migration_folder / "20260101010101_first.py").touch()
+    (migration_folder / "20260102020202_second.py").touch()
+
+    result = runner.invoke(app, ["migrate", "--check"])
+
+    assert result.exit_code == 0, result.output
+    assert "Checking migrations..." in result.output
+    assert "Migrations check passed successfully." in result.output
+
+
+def test_migrate_check_with_duplicate_id(runner, migration_folder):
+    (migration_folder / "20260101010101_duplicate_name.py").touch()
+    (migration_folder / "20260102020202_duplicate_name.py").touch()
+
+    result = runner.invoke(app, ["migrate", "--check"])
+
+    assert result.exit_code == 1, result.output
+    assert "Duplicate migration_id found in migrations" in result.output
+    assert "20260101010101_duplicate_name.py" in result.output
+    assert "20260102020202_duplicate_name.py" in result.output
+
+
+@pytest.mark.usefixtures("migration_folder")
+def test_migrate_check_empty_folder(runner):
+    result = runner.invoke(app, ["migrate", "--check"])
+
+    assert result.exit_code == 0, result.output
+    assert "Checking migrations..." in result.output
+    assert "Migrations check passed successfully." in result.output

tests/use_cases/test_check_migrations.py

@@ -0,0 +1,66 @@
+import pytest
+
+from mpt_tool.managers import FileMigrationManager
+from mpt_tool.models import MigrationFile
+from mpt_tool.use_cases.check_migrations import CheckMigrationsUseCase
+from mpt_tool.use_cases.errors import CheckMigrationError
+
+
+def test_check_migrations_no_duplicates(mocker):
+    migration_files = (
+        MigrationFile.build_from_path(mocker.Mock(stem="20260101010101_first")),
+        MigrationFile.build_from_path(mocker.Mock(stem="20260102020202_second")),
+    )
+    file_migration_manager = mocker.Mock(spec=FileMigrationManager)
+    file_migration_manager.retrieve_migration_files.return_value = migration_files
+    use_case = CheckMigrationsUseCase(file_migration_manager=file_migration_manager)
+
+    use_case.execute()  # act
+
+    file_migration_manager.retrieve_migration_files.assert_called_once()
+
+
+def test_check_migrations_empty_folder(mocker):
+    file_migration_manager = mocker.Mock(spec=FileMigrationManager)
+    file_migration_manager.retrieve_migration_files.return_value = ()
+    use_case = CheckMigrationsUseCase(file_migration_manager=file_migration_manager)
+
+    use_case.execute()  # act
+
+    file_migration_manager.retrieve_migration_files.assert_called_once()
+
+
+def test_check_migrations_with_duplicate_id(mocker):
+    migration_files = (
+        MigrationFile.build_from_path(mocker.Mock(stem="20260101010101_duplicate_name")),
+        MigrationFile.build_from_path(mocker.Mock(stem="20260102020202_duplicate_name")),
+    )
+    file_migration_manager = mocker.Mock(spec=FileMigrationManager)
+    file_migration_manager.retrieve_migration_files.return_value = migration_files
+    use_case = CheckMigrationsUseCase(file_migration_manager=file_migration_manager)
+
+    with pytest.raises(CheckMigrationError) as exc_info:
+        use_case.execute()
+
+    assert "Duplicate migration_id found in migrations" in str(exc_info.value)
+    assert "20260101010101_duplicate_name.py" in str(exc_info.value)
+    assert "20260102020202_duplicate_name.py" in str(exc_info.value)
+
+
+def test_check_migrations_multiple_duplicates(mocker):
+    migration_files = (
+        MigrationFile.build_from_path(mocker.Mock(stem="20260101010101_duplicate_one")),
+        MigrationFile.build_from_path(mocker.Mock(stem="20260102020202_duplicate_one")),
+        MigrationFile.build_from_path(mocker.Mock(stem="20260103030303_duplicate_two")),
+        MigrationFile.build_from_path(mocker.Mock(stem="20260104040404_duplicate_two")),
+    )
+    file_migration_manager = mocker.Mock(spec=FileMigrationManager)
+    file_migration_manager.retrieve_migration_files.return_value = migration_files
+    use_case = CheckMigrationsUseCase(file_migration_manager=file_migration_manager)
+
+    with pytest.raises(CheckMigrationError) as exc_info:
+        use_case.execute()
+
+    assert "Duplicate migration_id found in migrations" in str(exc_info.value)
+    assert "20260101010101_duplicate_one" in str(exc_info.value)
+    assert "20260104040404_duplicate_two" in str(exc_info.value)