Improve snow dcm test and refresh output

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

@@ -34,7 +34,6 @@
 )
 from snowflake.cli._plugins.dcm.utils import (
     RENDERED_DEFINITIONS_FOLDER,
-    announce_compile_separator,
     announce_rendered_definitions,
     clear_command_artifacts,
     mock_dcm_response,
@@ -577,10 +576,9 @@ def compile_project(
     reporter = AnalyzeErrorsReporter(save_output=save_output)
     if save_output:
         announce_rendered_definitions()
-    try:
-        return reporter.process(result)
-    finally:
-        announce_compile_separator()
+    # The reporter prints the trailing "=" divider itself (Reporter.print_separator),
+    # on both the success and error paths, so no separate separator is needed here.
+    return reporter.process(result)
 
 
 @app.command(
@@ -849,9 +847,13 @@ def refresh(
     context = _resolve_context_with_optional_manifest(from_location, identifier, target)
     project_id = context.project_identifier
 
-    with cli_console.spinner() as spinner:
-        spinner.add_task(description=f"Refreshing dcm project {project_id}", total=None)
-        result = DCMProjectManager().refresh(project_identifier=project_id)
+    manager = DCMProjectManager()
+    tracker = DeployProgressTracker(conn=manager.connection, operation="refresh")
+    with tracker.session():
+        result = tracker.run_loader_phase(
+            lambda: manager.refresh(project_identifier=project_id),
+            phase_name="REFRESH",
+        )
 
     reporter = RefreshReporter(save_output=save_output)
     return reporter.process(result)
@@ -874,9 +876,13 @@ def test(
     context = _resolve_context_with_optional_manifest(from_location, identifier, target)
     project_id = context.project_identifier
 
-    with cli_console.spinner() as spinner:
-        spinner.add_task(description=f"Testing dcm project {project_id}", total=None)
-        result = DCMProjectManager().test(project_identifier=project_id)
+    manager = DCMProjectManager()
+    tracker = DeployProgressTracker(conn=manager.connection, operation="test")
+    with tracker.session():
+        result = tracker.run_loader_phase(
+            lambda: manager.test(project_identifier=project_id),
+            phase_name="TEST",
+        )
 
     reporter = TestReporter(save_output=save_output)
     return reporter.process(result)

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

@@ -45,17 +45,20 @@
 # after COMPILE (there are no server-side progress phases to poll, so RENDER is
 # fast-forwarded and COMPILE shows the live spinner while the server works).
 COMPILE_OPERATION_PHASES = ["RENDER", "COMPILE"]
-OperationMode = Literal["deploy", "plan", "compile", "purge"]
+OperationMode = Literal["deploy", "plan", "compile", "purge", "test", "refresh"]
 
 # Server-side phase list per operation mode. UPLOAD is first for operations
-# that sync local files; purge runs entirely on the server (no local upload),
-# so it has no UPLOAD phase. Every operation uses the same
-# RENDER → COMPILE → PLAN → DEPLOY/PURGE vocabulary.
+# that sync local files; purge/test/refresh run entirely on the server against
+# already-deployed objects (no local upload), so they have no UPLOAD phase.
+# Deploy-family operations share the RENDER → COMPILE → PLAN → DEPLOY/PURGE
+# vocabulary; test/refresh are single-phase since they neither render nor plan.
 _PHASES_BY_OPERATION: dict[OperationMode, list[str]] = {
     "deploy": [UPLOAD_PHASE, *BACKEND_PHASES],
     "plan": [UPLOAD_PHASE, *PLAN_OPERATION_PHASES],
     "compile": [UPLOAD_PHASE, *COMPILE_OPERATION_PHASES],
     "purge": [*BACKEND_PHASES],
+    "test": ["TEST"],
+    "refresh": ["REFRESH"],
 }
 
 # Only PLAN and DEPLOY emit a meaningful 0-100 progress value.
@@ -273,6 +276,11 @@ def advance_upload(self, count: int = 1) -> None:
             self._refresh_display()
 
     def complete_upload(self) -> None:
+        # Operations that don't sync local files (purge/test/refresh) have no
+        # UPLOAD phase; nothing to complete. ``run_loader_phase`` calls this
+        # unconditionally, so the guard keeps single-phase modes working.
+        if not self._has_upload_phase():
+            return
         if self._upload_file_total > 0:
             self._get_phase(UPLOAD_PHASE).observe_running(100, datetime.now())
         self._get_phase(UPLOAD_PHASE).observe_done(datetime.now())

src/snowflake/cli/_plugins/dcm/reporters/base.py

@@ -34,6 +34,11 @@
 
 T = TypeVar("T")
 
+# Visual divider printed after every DCM command's reporter output so the
+# command's final line is clearly separated from the next shell prompt
+# (or from a CliError box on failure).
+_REPORT_SEPARATOR = "=" * 80
+
 
 class Reporter(ABC, Generic[T]):
     def __init__(self, save_output: bool = False) -> None:
@@ -78,6 +83,13 @@ def print_summary(self) -> None:
             cli_console.styled_message(renderable.plain, style=renderable.style)
         cli_console.styled_message("\n")
 
+    def print_separator(self) -> None:
+        """Print a divider line that marks the end of the command's output.
+
+        Automatically muted in JSON/CSV output formats (cf. `cli_console`)."""
+        cli_console.styled_message(_REPORT_SEPARATOR, style="dim")
+        cli_console.styled_message("\n")
+
     def _try_save_response(self, result_json: Dict[str, Any]) -> None:
         """Save raw JSON response if save_output is enabled and raw data is available."""
         if self.save_output:
@@ -96,10 +108,15 @@ def process_payload(self, result_json: Dict[str, Any]) -> None:
         self.print_renderables(parsed_data)
         if self._is_success():
             self.print_summary()
+            self.print_separator()
         else:
             message = "".join(
                 renderable.plain for renderable in self._generate_summary_renderables()
             )
+            # Print the separator before raising so the user still sees a
+            # clean divider between the partial body output and the error
+            # box that the CLI framework renders for the CliError.
+            self.print_separator()
             raise CliError(message)
 
     @staticmethod

src/snowflake/cli/_plugins/dcm/reporters/refresh.py

@@ -16,6 +16,7 @@
 from enum import Enum
 from typing import Any, Dict, Iterator, List, Optional
 
+import typer
 from pydantic import BaseModel, Field, ValidationError
 from rich.text import Text
 from snowflake.cli._plugins.dcm import styles
@@ -152,11 +153,16 @@ class RefreshReporter(Reporter[RefreshRow]):
     class Summary:
         up_to_date: int = 0
         refreshed: int = 0
-        unknown: int = 0
+        failed: int = 0
+
+        @property
+        def successful(self) -> int:
+            """Tables that completed without error (refreshed or already up-to-date)."""
+            return self.up_to_date + self.refreshed
 
         @property
         def total(self):
-            return self.up_to_date + self.refreshed + self.unknown
+            return self.successful + self.failed
 
     def __init__(self, save_output: bool = False):
         super().__init__(save_output=save_output)
@@ -187,7 +193,7 @@ def parse_data(self, data: List[RefreshTableResult]) -> Iterator[RefreshRow]:
             elif parsed.status == RefreshStatus.REFRESHED:
                 self._summary.refreshed += 1
             else:
-                self._summary.unknown += 1
+                self._summary.failed += 1
             yield parsed
 
     def print_renderables(self, data: Iterator[RefreshRow]) -> None:
@@ -208,25 +214,44 @@ def print_renderables(self, data: Iterator[RefreshRow]) -> None:
             cli_console.styled_message("\n")
 
     def _generate_summary_renderables(self) -> List[Text]:
-        total = self._summary.total
-        if total == 0:
+        if self._summary.total == 0:
             return [Text("No dynamic tables found in the project.")]
 
-        parts = []
-        if (refreshed := self._summary.refreshed) > 0:
-            parts.append(f"{refreshed} refreshed")
-        if (up_to_date := self._summary.up_to_date) > 0:
-            parts.append(f"{up_to_date} up-to-date")
-        if (unknown := self._summary.unknown) > 0:
-            parts.append(f"{unknown} unknown")
-
-        summary = ""
-        for i, part in enumerate(parts):
-            if i > 0:
-                summary += ", "
-            summary += part
-        summary += "."
-        return [Text(summary)]
+        renderables: List[Text] = []
+        successful = self._summary.successful
+        failed = self._summary.failed
+
+        if successful > 0:
+            tables_word = "Dynamic Table" if successful == 1 else "Dynamic Tables"
+            renderables.append(
+                Text(
+                    f"{successful} {tables_word} refreshed successfully.",
+                    styles.PASS_STYLE,
+                )
+            )
+
+        if failed > 0:
+            refreshes_word = "Refresh" if failed == 1 else "Refreshes"
+            if renderables:
+                renderables.append(Text("\n"))
+            renderables.append(
+                Text(f"{failed} {refreshes_word} failed.", styles.FAIL_STYLE)
+            )
+
+        return renderables
 
     def _is_success(self) -> bool:
-        return self._summary.unknown == 0
+        # Always treat the run as "success" from the base reporter's perspective
+        # so that `print_summary()` is always called — we want the styled
+        # green/red summary rendered above the divider on every run (cf.
+        # AnalyzeErrorsReporter). Exit code is set separately in
+        # ``process_payload``.
+        return True
+
+    def process_payload(self, result_json: Dict[str, Any]) -> None:
+        super().process_payload(result_json)
+        if self._summary.failed > 0:
+            # Failures fail the command, but the styled summary is already on
+            # screen — exit silently so we don't double-render the message
+            # inside an "Error" box.
+            raise typer.Exit(code=1)

src/snowflake/cli/_plugins/dcm/reporters/test.py

@@ -16,6 +16,7 @@
 from enum import Enum
 from typing import Any, Dict, Iterator, List, Optional
 
+import typer
 from rich.text import Text
 from snowflake.cli._plugins.dcm import styles
 from snowflake.cli._plugins.dcm.reporters.base import Reporter, cli_console
@@ -151,18 +152,47 @@ def _generate_summary_renderables(self) -> List[Text]:
         if total == 0:
             return [Text("No expectations found in the project.")]
 
-        result = [
-            (Text(f"{self._summary.passed} passed", styles.PASS_STYLE)),
-            (Text(", ")),
-            (Text(f"{self._summary.failed} failed", styles.FAIL_STYLE)),
+        test_word = "test" if total == 1 else "tests"
+        # The trailing noun (``expectation`` / ``expectations``) attaches to
+        # the last visible bucket — ``unknown`` if present, otherwise
+        # ``failed`` — and pluralizes with that bucket's count.
+        trailing_count = (
+            self._summary.unknown if self._summary.unknown > 0 else self._summary.failed
+        )
+        noun = "expectation" if trailing_count == 1 else "expectations"
+
+        result: List[Text] = [
+            Text("Ran "),
+            Text(f"{total}", styles.BOLD_STYLE),
+            Text(f" data quality {test_word}: "),
+            Text(f"{self._summary.passed} passed", styles.PASS_STYLE),
+            Text(", "),
         ]
         if self._summary.unknown > 0:
+            result.append(Text(f"{self._summary.failed} failed", styles.FAIL_STYLE))
             result.append(Text(", "))
-            result.append(Text(f"{self._summary.unknown} unknown", styles.FAIL_STYLE))
-        result.append(Text(" out of "))
-        result.append(Text(f"{total}", styles.BOLD_STYLE))
-        result.append(Text(" total."))
+            result.append(
+                Text(f"{self._summary.unknown} unknown {noun}", styles.FAIL_STYLE)
+            )
+        else:
+            result.append(
+                Text(f"{self._summary.failed} failed {noun}", styles.FAIL_STYLE)
+            )
+        result.append(Text("."))
         return result
 
     def _is_success(self) -> bool:
-        return self._summary.failed + self._summary.unknown == 0
+        # Always treat the run as "success" from the base reporter's
+        # perspective so that ``print_summary()`` is always called — the
+        # styled "✓ N passed, ✗ M failed" summary should render on every run
+        # (cf. ``AnalyzeErrorsReporter`` / ``RefreshReporter``). Exit code is
+        # set separately by ``process_payload`` below.
+        return True
+
+    def process_payload(self, result_json: Dict[str, Any]) -> None:
+        super().process_payload(result_json)
+        if self._summary.failed + self._summary.unknown > 0:
+            # Failures fail the command, but the styled summary is already
+            # on screen — exit silently so we don't double-render the
+            # message inside an "Error" box.
+            raise typer.Exit(code=1)

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

@@ -35,8 +35,6 @@
 # downloaded. Shared by the ``compile`` and ``dependencies`` commands so the
 # rendered output always lands in the same, descriptively-named place.
 RENDERED_DEFINITIONS_FOLDER = "rendered_definitions"
-# Width of the trailing separator line printed at the end of a ``compile`` run.
-COMPILE_SEPARATOR_WIDTH = 61
 
 
 def clear_command_artifacts(
@@ -91,12 +89,6 @@ def announce_rendered_definitions() -> None:
     cli_console.styled_message("\n")
 
 
-def announce_compile_separator() -> None:
-    """Print a separator line marking the end of a ``compile`` run."""
-    cli_console.styled_message("=" * COMPILE_SEPARATOR_WIDTH)
-    cli_console.styled_message("\n")
-
-
 def save_command_response(
     command_name: str,
     raw_data: Dict[str, Any] | str,