Flag for error on ignored versioned migration

schemachange/config/DeployConfig.py

@@ -27,6 +27,8 @@ class DeployConfig(BaseConfig):
     dry_run: bool = False
     query_tag: str | None = None
     oauth_config: dict | None = None
+    version_number_validation_regex: str | None = None
+    raise_exception_on_ignored_versioned_script: bool = False
 
     @classmethod
     def factory(

schemachange/config/parse_cli_args.py

@@ -180,6 +180,20 @@ def parse_cli_args(args) -> dict:
         '"https//...", "token-request-payload": {"client_id": "GUID_xyz",...},... })',
         required=False,
     )
+    parser_deploy.add_argument(
+        "--version-number-validation-regex",
+        type=str,
+        help="If supplied, version numbers will be validated with this regular expression.",
+        required=False,
+    )
+    parser_deploy.add_argument(
+        "--raise-exception-on-ignored-versioned-script",
+        action="store_const",
+        const=True,
+        default=None,
+        help="Raise an exception if an un-applied versioned script is ignored (the default is False)",
+        required=False,
+    )
 
     parser_render = subcommands.add_parser(
         "render",

schemachange/deploy.py

@@ -60,6 +60,7 @@ def deploy(config: DeployConfig, session: SnowflakeSession):
     # Find all scripts in the root folder (recursively) and sort them correctly
     all_scripts = get_all_scripts_recursively(
         root_directory=config.root_folder,
+        version_number_regex=config.version_number_validation_regex,
     )
     all_script_names = list(all_scripts.keys())
     # Sort scripts such that versioned scripts get applied first and then the repeatable ones.
@@ -106,12 +107,18 @@ def deploy(config: DeployConfig, session: SnowflakeSession):
                 and get_alphanum_key(script.version) <= max_published_version
             ):
                 if script_metadata is None:
-                    script_log.debug(
-                        "Skipping versioned script because it's older than the most recently applied change",
-                        max_published_version=max_published_version,
-                    )
-                    scripts_skipped += 1
-                    continue
+                    if config.raise_exception_on_ignored_versioned_script:
+                        raise ValueError(
+                            f"Versioned script will never be applied: {script.name}\n"
+                            f"Version number is less than the max version number: {max_published_version}"
+                        )
+                    else:
+                        script_log.debug(
+                            "Skipping versioned script because it's older than the most recently applied change",
+                            max_published_version=max_published_version,
+                        )
+                        scripts_skipped += 1
+                        continue
                 else:
                     script_log.debug(
                         "Script has already been applied",

schemachange/session/Script.py

@@ -41,6 +41,13 @@ def from_path(cls, file_path: Path, **kwargs) -> T:
         script_name = cls.get_script_name(file_path=file_path)
         name_parts = cls.pattern.search(file_path.name.strip())
         description = name_parts.group("description").replace("_", " ").capitalize()
+        if len(name_parts.group("separator")) != 2:
+            prefix = f"V{name_parts.group('version')}" if cls.type == "V" else cls.type
+
+            raise ValueError(
+                f'two underscores are required between "{ prefix }" and the description: '
+                f"{file_path}\n{str(file_path)}"
+            )
         # noinspection PyArgumentList
         return cls(
             name=script_name, file_path=file_path, description=description, **kwargs
@@ -50,15 +57,30 @@ def from_path(cls, file_path: Path, **kwargs) -> T:
 @dataclasses.dataclass(frozen=True)
 class VersionedScript(Script):
     pattern: ClassVar[re.Pattern[str]] = re.compile(
-        r"^(V)(?P<version>.+?)?__(?P<description>.+?)\.", re.IGNORECASE
+        r"^(V)(?P<version>([^_]|_(?!_))+)?(?P<separator>_{1,2})(?P<description>.+?)\.",
+        re.IGNORECASE,
     )
     type: ClassVar[Literal["V"]] = "V"
+    version_number_regex: ClassVar[str | None] = None
     version: str
 
     @classmethod
     def from_path(cls: T, file_path: Path, **kwargs) -> T:
         name_parts = cls.pattern.search(file_path.name.strip())
 
+        version = name_parts.group("version")
+        if version is None:
+            raise ValueError(
+                f"Versioned migrations must be prefixed with a version: {str(file_path)}"
+            )
+
+        if cls.version_number_regex:
+            if re.search(cls.version_number_regex, version, re.IGNORECASE) is None:
+                raise ValueError(
+                    f"change script version doesn't match the supplied regular expression: "
+                    f"{cls.version_number_regex}\n{str(file_path)}"
+                )
+
         return super().from_path(
             file_path=file_path, version=name_parts.group("version")
         )
@@ -67,15 +89,15 @@ def from_path(cls: T, file_path: Path, **kwargs) -> T:
 @dataclasses.dataclass(frozen=True)
 class RepeatableScript(Script):
     pattern: ClassVar[re.Pattern[str]] = re.compile(
-        r"^(R)__(?P<description>.+?)\.", re.IGNORECASE
+        r"^(R)(?P<separator>_{1,2})(?P<description>.+?)\.", re.IGNORECASE
     )
     type: ClassVar[Literal["R"]] = "R"
 
 
 @dataclasses.dataclass(frozen=True)
 class AlwaysScript(Script):
     pattern: ClassVar[re.Pattern[str]] = re.compile(
-        r"^(A)__(?P<description>.+?)\.", re.IGNORECASE
+        r"^(A)(?P<separator>_{1,2})(?P<description>.+?)\.", re.IGNORECASE
     )
     type: ClassVar[Literal["A"]] = "A"
 
@@ -95,7 +117,11 @@ def script_factory(
     logger.debug("ignoring non-change file", file_path=str(file_path))
 
 
-def get_all_scripts_recursively(root_directory: Path):
+def get_all_scripts_recursively(
+    root_directory: Path, version_number_regex: str | None = None
+):
+    VersionedScript.version_number_regex = version_number_regex
+
     all_files: dict[str, T] = dict()
     all_versions = list()
     # Walk the entire directory structure recursively

tests/config/test_Config.py

@@ -35,6 +35,8 @@ def yaml_config(_) -> DeployConfig:
         dry_run=True,
         query_tag="yaml_query_tag",
         oauth_config={"oauth": "yaml_oauth"},
+        version_number_validation_regex="yaml_version_number_validation_regex",
+        raise_exception_on_ignored_versioned_script=True,
     )
 
 
@@ -204,6 +206,7 @@ def test_invalid_root_folder(self, _):
                 change_history_table="some_history_table",
                 query_tag="some_query_tag",
                 oauth_config={"some": "values"},
+                version_number_validation_regex="some_regex",
             )
         e_info_value = str(e_info.value)
         assert "Path is not valid directory: some_root_folder_name" in e_info_value
@@ -225,6 +228,7 @@ def test_invalid_modules_folder(self, _):
                 change_history_table="some_history_table",
                 query_tag="some_query_tag",
                 oauth_config={"some": "values"},
+                version_number_validation_regex="some_regex",
             )
         e_info_value = str(e_info.value)
         assert "Path is not valid directory: some_modules_folder_name" in e_info_value

tests/config/test_get_merged_config.py

@@ -128,6 +128,9 @@ def test_all_cli_args(self, _):
                 "query-tag-from-cli",
                 "--oauth-config",
                 '{"token-provider-url": "https//...", "token-request-payload": {"client_id": "GUID_xyz"} }',
+                "--version-number-validation-regex",
+                "version_number_validation_regex-from-cli",
+                "--raise-exception-on-ignored-versioned-script",
             ],
         ):
             config = get_merged_config()

tests/config/test_parse_cli_args.py

@@ -21,6 +21,7 @@ def test_parse_args_defaults():
     assert parsed_args["create_change_history_table"] is None
     assert parsed_args["autocommit"] is None
     assert parsed_args["dry_run"] is None
+    assert parsed_args["raise_exception_on_ignored_versioned_script"] is None
     assert parsed_args["subcommand"] == "deploy"
 
 
@@ -46,6 +47,7 @@ def test_parse_args_deploy_names():
         ("--change-history-table", "some_history_table", "some_history_table"),
         ("--query-tag", "some_query_tag", "some_query_tag"),
         ("--oauth-config", json.dumps({"some": "values"}), {"some": "values"}),
+        ("--version-number-validation-regex", "some_regex", "some_regex"),
     ]
 
     for arg, value, expected_value in valued_test_args:
@@ -58,6 +60,7 @@ def test_parse_args_deploy_names():
         ("--create-change-history-table", True),
         ("--autocommit", True),
         ("--dry-run", True),
+        ("--raise-exception-on-ignored-versioned-script", True),
     ]
 
     for arg, expected_value in valueless_test_args:

tests/session/test_Script.py

@@ -43,6 +43,24 @@ def test_get_script_name(self, file_path: Path, expected: str):
                     version="123",
                 ),
             ),
+            (
+                Path("nested/file/V1.2.3__something.sql.jinja"),
+                VersionedScript(
+                    name="V1.2.3__something.sql",
+                    file_path=Path("nested/file/V1.2.3__something.sql.jinja"),
+                    description="Something",
+                    version="1.2.3",
+                ),
+            ),
+            (
+                Path("nested/file/V1_2_3__something.sql.jinja"),
+                VersionedScript(
+                    name="V1_2_3__something.sql",
+                    file_path=Path("nested/file/V1_2_3__something.sql.jinja"),
+                    description="Something",
+                    version="1_2_3",
+                ),
+            ),
             (
                 Path("nested/file/R__something.sql.jinja"),
                 RepeatableScript(
@@ -68,6 +86,24 @@ def test_get_script_name(self, file_path: Path, expected: str):
                     version="123",
                 ),
             ),
+            (
+                Path("nested/file/V1_2_3__something.sql"),
+                VersionedScript(
+                    name="V1_2_3__something.sql",
+                    file_path=Path("nested/file/V1_2_3__something.sql"),
+                    description="Something",
+                    version="1_2_3",
+                ),
+            ),
+            (
+                Path("nested/file/V1.2.3__something.sql"),
+                VersionedScript(
+                    name="V1.2.3__something.sql",
+                    file_path=Path("nested/file/V1.2.3__something.sql"),
+                    description="Something",
+                    version="1.2.3",
+                ),
+            ),
             (
                 Path("nested/file/R__something.sql"),
                 RepeatableScript(
@@ -100,6 +136,29 @@ def test_script_factory(self, file_path: Path, expected: Script):
         result = script_factory(file_path)
         assert result == expected
 
+    @pytest.mark.parametrize(
+        "file_path",
+        [
+            (Path("nested/file/V123_something.sql.jinja")),
+            (Path("nested/file/V1_2_3_something.sql.jinja")),
+            (Path("nested/file/V1.2.3_something.sql.jinja")),
+            (Path("nested/file/R_something.sql.jinja")),
+            (Path("nested/file/A_something.sql.jinja")),
+        ],
+    )
+    def test_single_underscore_should_raise_exception(self, file_path: Path):
+        with pytest.raises(ValueError) as e:
+            script_factory(file_path)
+        assert str(file_path) in str(e.value) and "two underscores" in str(e.value)
+
+    def test_missing_version_should_raise_exception(self):
+        file_path = Path("nested/file/V__something.sql.jinja")
+        with pytest.raises(ValueError) as e:
+            script_factory(file_path)
+        assert str(file_path) in str(
+            e.value
+        ) and "Versioned migrations must be prefixed with a version" in str(e.value)
+
 
 class TestGetAllScriptsRecursively:
     def test_given_empty_folder_should_return_empty(self):
@@ -141,13 +200,33 @@ def test_version_number_regex_numeric_happy_path(self):
                 [],
             ]
 
-            result = get_all_scripts_recursively(Path("scripts"))
+            result = get_all_scripts_recursively(
+                Path("scripts"),
+                version_number_regex=r"\d\.\d\.\d",  # noqa: W605
+            )
 
         assert len(result) == 3
         assert "v1.1.1__initial.sql" in result
         assert "v1.1.2__update.sql" in result
         assert "v1.1.3__update.sql" in result
 
+    def test_version_number_regex_numeric_exception(self):
+        with mock.patch("pathlib.Path.rglob") as mock_rglob:
+            mock_rglob.side_effect = [
+                [
+                    Path("V1.10.1__initial.sql"),
+                ],
+                [],
+            ]
+            with pytest.raises(ValueError) as e:
+                get_all_scripts_recursively(
+                    Path("scripts"),
+                    version_number_regex=r"\d\.\d\.\d",  # noqa: W605
+                )
+            assert str(e.value).startswith(
+                "change script version doesn't match the supplied regular expression"
+            )
+
     def test_version_number_regex_text_happy_path(self):
         with mock.patch("pathlib.Path.rglob") as mock_rglob:
             mock_rglob.side_effect = [
@@ -156,10 +235,30 @@ def test_version_number_regex_text_happy_path(self):
                 ],
                 [],
             ]
-            result = get_all_scripts_recursively(Path("scripts"))
+            result = get_all_scripts_recursively(
+                Path("scripts"),
+                version_number_regex=r"[a-z]\.[a-z]\.[a-z]",  # noqa: W605
+            )
         assert len(result) == 1
         assert "va.b.c__initial.sql" in result
 
+    def test_version_number_regex_text_exception(self):
+        with mock.patch("pathlib.Path.rglob") as mock_rglob:
+            mock_rglob.side_effect = [
+                [
+                    Path("V1.10.1__initial.sql"),
+                ],
+                [],
+            ]
+            with pytest.raises(ValueError) as e:
+                get_all_scripts_recursively(
+                    Path("scripts"),
+                    version_number_regex=r"[a-z]\.[a-z]\.[a-z]",  # noqa: W605
+                )
+            assert str(e.value).startswith(
+                "change script version doesn't match the supplied regular expression"
+            )
+
     def test_given_version_files_should_return_version_files(self):
         with mock.patch("pathlib.Path.rglob") as mock_rglob:
             mock_rglob.side_effect = [

tests/test_main.py

@@ -40,6 +40,8 @@
     "dry_run": False,
     "query_tag": None,
     "oauth_config": None,
+    "version_number_validation_regex": None,
+    "raise_exception_on_ignored_versioned_script": False,
 }
 
 required_args = [