feat: [SNOW-2326970] implement dcm test command

Author: sfc-gh-jwilkowski

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

@@ -11,10 +11,12 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import json
 from typing import List, Optional
 
 import typer
 from snowflake.cli._plugins.dcm.manager import DCMProjectManager
+from snowflake.cli._plugins.dcm.utils import format_test_failures
 from snowflake.cli._plugins.object.command_aliases import add_object_command_aliases
 from snowflake.cli._plugins.object.commands import scope_option
 from snowflake.cli._plugins.object.manager import ObjectManager
@@ -235,6 +237,47 @@ def drop_deployment(
     )
 
 
+@app.command(requires_connection=True)
+def test(
+    identifier: FQN = dcm_identifier,
+    **options,
+):
+    """
+    Test all expectations set for tables, views and dynamic tables defined
+    in DCM project.
+    """
+    with cli_console.spinner() as spinner:
+        spinner.add_task(description=f"Testing dcm project {identifier}", total=None)
+        result = DCMProjectManager().test(project_identifier=identifier)
+
+    row = result.fetchone()
+    if not row:
+        return MessageResult("No data.")
+
+    result_data = row[0]
+    result_json = (
+        json.loads(result_data) if isinstance(result_data, str) else result_data
+    )
+
+    expectations = result_json.get("expectations", [])
+
+    if not expectations:
+        return MessageResult("No expectations defined in the project.")
+
+    if result_json.get("status") == "EXPECTATION_VIOLATED":
+        failed_expectations = [
+            exp for exp in expectations if exp.get("expectation_violated", False)
+        ]
+        total_tests = len(expectations)
+        failed_count = len(failed_expectations)
+        error_message = format_test_failures(
+            failed_expectations, total_tests, failed_count
+        )
+        raise CliError(error_message)
+
+    return MessageResult(f"All {len(expectations)} expectation(s) passed successfully.")
+
+
 def _get_effective_stage(identifier: FQN, from_location: Optional[str]):
     manager = DCMProjectManager()
     if not from_location:

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

@@ -136,6 +136,10 @@ def drop_deployment(
         query += f' "{deployment_name}"'
         return self.execute_query(query=query)
 
+    def test(self, project_identifier: FQN):
+        query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} TEST ALL"
+        return self.execute_query(query=query)
+
     @staticmethod
     def sync_local_files(
         project_identifier: FQN, source_directory: str | None = None

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

@@ -0,0 +1,43 @@
+# Copyright (c) 2024 Snowflake Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+def format_test_failures(
+    failed_expectations: list, total_tests: int, failed_count: int
+) -> str:
+    """Format test failures into a nice error message."""
+    lines = [
+        "Failed expectations:",
+    ]
+
+    for failed in failed_expectations:
+        table_name = failed.get("table_name", "Unknown")
+        expectation_name = failed.get("expectation_name", "Unknown")
+        metric_name = failed.get("metric_name", "Unknown")
+        expectation_expr = failed.get("expectation_expression", "N/A")
+        value = failed.get("value", "N/A")
+
+        lines.append(f"  Table: {table_name}")
+        lines.append(f"  Expectation: {expectation_name}")
+        lines.append(f"  Metric: {metric_name}")
+        lines.append(f"  Expression: {expectation_expr}")
+        lines.append(f"  Actual value: {value}")
+        lines.append("")
+
+    passed_tests = total_tests - failed_count
+    lines.append(
+        f"Tests completed: {passed_tests} passed, {failed_count} failed out of {total_tests} total."
+    )
+
+    return "\n".join(lines)