feat: [SNOW-2306301] extend dcm plan/deploy --from flag to support local paths (#2596)

Author: sfc-gh-jwilkowski

src/snowflake/cli/_plugins/dcm/commands.py

@@ -39,6 +39,7 @@
     QueryJsonValueResult,
     QueryResult,
 )
+from snowflake.cli.api.utils.path_utils import is_stage_path
 
 app = SnowTyperFactory(
     name="dcm",
@@ -56,9 +57,10 @@
     help="Configuration of the DCM Project to use. If not specified default configuration is used.",
     show_default=False,
 )
-from_option = OverrideableOption(
+from_option = typer.Option(
     None,
     "--from",
+    help="Source location: stage path (starting with '@') or local directory path. Omit to use current directory.",
     show_default=False,
 )
 
@@ -106,9 +108,7 @@
 @app.command(requires_connection=True)
 def deploy(
     identifier: FQN = dcm_identifier,
-    from_stage: Optional[str] = from_option(
-        help="Deploy DCM Project deployment from a given stage."
-    ),
+    from_location: Optional[str] = from_option,
     variables: Optional[List[str]] = variables_flag,
     configuration: Optional[str] = configuration_flag,
     alias: Optional[str] = alias_option,
@@ -118,14 +118,14 @@ def deploy(
     Applies changes defined in DCM Project to Snowflake.
     """
     manager = DCMProjectManager()
-    if not from_stage:
-        from_stage = manager.sync_local_files(project_identifier=identifier)
+    effective_stage = _get_effective_stage(identifier, from_location)
+
     with cli_console.spinner() as spinner:
         spinner.add_task(description=f"Deploying dcm project {identifier}", total=None)
         result = manager.execute(
             project_identifier=identifier,
             configuration=configuration,
-            from_stage=from_stage,
+            from_stage=effective_stage,
             variables=variables,
             alias=alias,
             output_path=None,
@@ -136,9 +136,7 @@ def deploy(
 @app.command(requires_connection=True)
 def plan(
     identifier: FQN = dcm_identifier,
-    from_stage: Optional[str] = from_option(
-        help="Plan DCM Project deployment from a given stage."
-    ),
+    from_location: Optional[str] = from_option,
     variables: Optional[List[str]] = variables_flag,
     configuration: Optional[str] = configuration_flag,
     output_path: Optional[str] = output_path_option(
@@ -150,15 +148,14 @@ def plan(
     Plans a DCM Project deployment (validates without executing).
     """
     manager = DCMProjectManager()
-    if not from_stage:
-        from_stage = manager.sync_local_files(project_identifier=identifier)
+    effective_stage = _get_effective_stage(identifier, from_location)
 
     with cli_console.spinner() as spinner:
         spinner.add_task(description=f"Planning dcm project {identifier}", total=None)
         result = manager.execute(
             project_identifier=identifier,
             configuration=configuration,
-            from_stage=from_stage,
+            from_stage=effective_stage,
             dry_run=True,
             variables=variables,
             output_path=output_path,
@@ -236,3 +233,16 @@ def drop_deployment(
     return MessageResult(
         f"Deployment '{deployment_name}' dropped from DCM Project '{identifier}'."
     )
+
+
+def _get_effective_stage(identifier: FQN, from_location: Optional[str]):
+    manager = DCMProjectManager()
+    if not from_location:
+        from_stage = manager.sync_local_files(project_identifier=identifier)
+    elif is_stage_path(from_location):
+        from_stage = from_location
+    else:
+        from_stage = manager.sync_local_files(
+            project_identifier=identifier, source_directory=from_location
+        )
+    return from_stage

src/snowflake/cli/_plugins/dcm/manager.py

@@ -137,10 +137,20 @@ def drop_deployment(
         return self.execute_query(query=query)
 
     @staticmethod
-    def sync_local_files(project_identifier: FQN) -> str:
-        dcm_manifest_file = SecurePath.cwd() / MANIFEST_FILE_NAME
+    def sync_local_files(
+        project_identifier: FQN, source_directory: str | None = None
+    ) -> str:
+        source_path = (
+            SecurePath(source_directory).resolve()
+            if source_directory
+            else SecurePath.cwd()
+        )
+
+        dcm_manifest_file = source_path / MANIFEST_FILE_NAME
         if not dcm_manifest_file.exists():
-            raise CliError(f"{MANIFEST_FILE_NAME} was not found in project directory")
+            raise CliError(
+                f"{MANIFEST_FILE_NAME} was not found in directory {source_path.path}"
+            )
 
         with dcm_manifest_file.open(read_file_limit_mb=DEFAULT_SIZE_LIMIT_MB) as fd:
             dcm_manifest = yaml.safe_load(fd)
@@ -160,10 +170,10 @@ def sync_local_files(project_identifier: FQN) -> str:
 
         with cli_console.phase(f"Uploading definition files"):
             stage_fqn = FQN.from_resource(
-                ObjectType.DCM_PROJECT, project_identifier, "_TMP_STAGE"
+                ObjectType.DCM_PROJECT, project_identifier, "TMP_STAGE"
             )
             sync_artifacts_with_stage(
-                project_paths=ProjectPaths(project_root=Path.cwd()),
+                project_paths=ProjectPaths(project_root=source_path.path),
                 stage_root=stage_fqn.identifier,
                 use_temporary_stage=True,
                 artifacts=[PathMapping(src=definition) for definition in definitions],

tests/__snapshots__/test_help_messages.ambr

@@ -7388,8 +7388,9 @@
   |                            [required]                                        |
   +------------------------------------------------------------------------------+
   +- Options --------------------------------------------------------------------+
-  | --from                   TEXT  Deploy DCM Project deployment from a given    |
-  |                                stage.                                        |
+  | --from                   TEXT  Source location: stage path (starting with    |
+  |                                '@') or local directory path. Omit to use     |
+  |                                current directory.                            |
   | --variable       -D      TEXT  Variables for the execution context; for      |
   |                                example: -D "<key>=<value>".                  |
   | --configuration          TEXT  Configuration of the DCM Project to use. If   |
@@ -8204,8 +8205,9 @@
   |                            [required]                                        |
   +------------------------------------------------------------------------------+
   +- Options --------------------------------------------------------------------+
-  | --from                   TEXT  Plan DCM Project deployment from a given      |
-  |                                stage.                                        |
+  | --from                   TEXT  Source location: stage path (starting with    |
+  |                                '@') or local directory path. Omit to use     |
+  |                                current directory.                            |
   | --variable       -D      TEXT  Variables for the execution context; for      |
   |                                example: -D "<key>=<value>".                  |
   | --configuration          TEXT  Configuration of the DCM Project to use. If   |

tests/api/test_fqn.py

@@ -200,7 +200,7 @@ def test_git_fqn():
     assert fqn.name == "git_repo"
 
 
-class TestRelatedToResource:
+class TestFromResource:
     @pytest.fixture
     def mock_time(self):
         with mock.patch(

tests/dcm/test_commands.py

@@ -215,6 +215,43 @@ def test_deploy_project_with_sync(
         assert "DCM_FOOBAR" in call_args.kwargs["from_stage"]
         assert call_args.kwargs["from_stage"].endswith("_TMP_STAGE")
 
+    @mock.patch(DCMProjectManager)
+    def test_deploy_project_with_from_local_directory(
+        self,
+        mock_pm,
+        runner,
+        project_directory,
+        mock_cursor,
+        mock_connect,
+        tmp_path,
+    ):
+        mock_pm().execute.return_value = mock_cursor(
+            rows=[("[]",)], columns=("operations")
+        )
+        mock_pm().sync_local_files.return_value = (
+            "MockDatabase.MockSchema.DCM_FOOBAR_1234567890_TMP_STAGE"
+        )
+
+        source_dir = tmp_path / "source_project"
+        source_dir.mkdir()
+
+        manifest_file = source_dir / "manifest.yml"
+        manifest_file.write_text("type: dcm_project\n")
+
+        with project_directory("dcm_project"):
+            result = runner.invoke(
+                ["dcm", "deploy", "my_project", "--from", str(source_dir)]
+            )
+            assert result.exit_code == 0, result.output
+
+        mock_pm().sync_local_files.assert_called_once_with(
+            project_identifier=FQN.from_string("my_project"),
+            source_directory=str(source_dir),
+        )
+
+        call_args = mock_pm().execute.call_args
+        assert call_args.kwargs["from_stage"].endswith("_TMP_STAGE")
+
 
 class TestDCMPlan:
     @mock.patch(DCMProjectManager)
@@ -376,6 +413,42 @@ def test_plan_project_with_sync(
             assert "DCM_FOOBAR_" in call_args.kwargs["from_stage"]
             assert call_args.kwargs["from_stage"].endswith("_TMP_STAGE")
 
+    @mock.patch(DCMProjectManager)
+    def test_plan_project_with_from_local_directory(
+        self,
+        mock_pm,
+        runner,
+        project_directory,
+        mock_cursor,
+        mock_connect,
+        tmp_path,
+    ):
+        mock_pm().execute.return_value = mock_cursor(
+            rows=[("[]",)], columns=("operations")
+        )
+        mock_pm().sync_local_files.return_value = (
+            "MockDatabase.MockSchema.DCM_FOOBAR_1234567890_TMP_STAGE"
+        )
+
+        source_dir = tmp_path / "source_project"
+        source_dir.mkdir()
+        manifest_file = source_dir / "manifest.yml"
+        manifest_file.write_text("type: dcm_project\n")
+
+        with project_directory("dcm_project"):
+            result = runner.invoke(
+                ["dcm", "plan", "my_project", "--from", str(source_dir)]
+            )
+            assert result.exit_code == 0, result.output
+
+        mock_pm().sync_local_files.assert_called_once_with(
+            project_identifier=FQN.from_string("my_project"),
+            source_directory=str(source_dir),
+        )
+
+        call_args = mock_pm().execute.call_args
+        assert call_args.kwargs["from_stage"].endswith("_TMP_STAGE")
+
 
 class TestDCMList:
     def test_list_command_alias(self, mock_connect, runner):

tests/dcm/test_manager.py

@@ -1,3 +1,4 @@
+import os
 from pathlib import Path
 from unittest import mock
 
@@ -251,7 +252,7 @@ def test_raises_when_manifest_file_is_missing(self, project_directory):
             (project_dir / MANIFEST_FILE_NAME).unlink()
             with pytest.raises(
                 CliError,
-                match=f"{MANIFEST_FILE_NAME} was not found in project directory",
+                match=f"{MANIFEST_FILE_NAME} was not found in directory",
             ):
                 DCMProjectManager.sync_local_files(project_identifier=TEST_PROJECT)
 
@@ -314,3 +315,75 @@ def test_calls_sync_artifacts_with_stage(
             actual_project_root = call_args.kwargs["project_paths"].project_root
             expected_project_root = project_dir.resolve()
             assert actual_project_root.resolve() == expected_project_root.resolve()
+
+    @mock.patch("snowflake.cli._plugins.dcm.manager.sync_artifacts_with_stage")
+    @mock.patch("snowflake.cli._plugins.dcm.manager.StageManager.create")
+    def test_sync_local_files_with_source_directory(
+        self,
+        _mock_create_stage,
+        mock_sync_artifacts_with_stage,
+        tmp_path,
+        mock_connect,
+        mock_cursor,
+        mock_from_resource,
+    ):
+        source_dir = tmp_path / "custom_source"
+        source_dir.mkdir()
+
+        manifest_content = {
+            "type": "dcm_project",
+            "include_definitions": ["definitions/custom_query.sql"],
+        }
+        manifest_file = source_dir / MANIFEST_FILE_NAME
+        with open(manifest_file, "w") as f:
+            yaml.dump(manifest_content, f)
+
+        # Create the definition file
+        definitions_dir = source_dir / "definitions"
+        definitions_dir.mkdir()
+        (definitions_dir / "custom_query.sql").write_text("SELECT 1;")
+
+        DCMProjectManager.sync_local_files(
+            project_identifier=TEST_PROJECT, source_directory=str(source_dir)
+        )
+
+        mock_sync_artifacts_with_stage.assert_called_once()
+        call_args = mock_sync_artifacts_with_stage.call_args
+        actual_project_root = call_args.kwargs["project_paths"].project_root
+        assert actual_project_root.resolve() == source_dir.resolve()
+
+    @mock.patch("snowflake.cli._plugins.dcm.manager.sync_artifacts_with_stage")
+    @mock.patch("snowflake.cli._plugins.dcm.manager.StageManager.create")
+    def test_sync_local_files_with_relative_source_directory(
+        self,
+        _mock_create_stage,
+        mock_sync_artifacts_with_stage,
+        tmp_path,
+        mock_connect,
+        mock_cursor,
+        mock_from_resource,
+    ):
+        source_dir = tmp_path / "relative_source"
+        source_dir.mkdir()
+
+        manifest_file = source_dir / MANIFEST_FILE_NAME
+        with open(manifest_file, "w") as f:
+            yaml.dump({"type": "dcm_project"}, f)
+
+        original_cwd = os.getcwd()
+        try:
+            os.chdir(tmp_path)
+
+            DCMProjectManager.sync_local_files(
+                project_identifier=TEST_PROJECT,
+                source_directory="relative_source",  # relative path
+            )
+
+            mock_sync_artifacts_with_stage.assert_called_once()
+            call_args = mock_sync_artifacts_with_stage.call_args
+
+            actual_project_root = call_args.kwargs["project_paths"].project_root
+            assert actual_project_root.is_absolute()
+            assert actual_project_root.resolve() == source_dir.resolve()
+        finally:
+            os.chdir(original_cwd)