feat: flag for error on ignored versioned migration and version numbe…

…r regex

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

@@ -58,6 +58,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.
@@ -104,12 +105,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

@@ -53,12 +53,21 @@ class VersionedScript(Script):
         r"^(V)(?P<version>.+?)?__(?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())
 
+        if cls.version_number_regex:
+            version = name_parts.group("version")
+            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")
         )
@@ -95,7 +104,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

@@ -141,13 +141,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 +176,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 = [