diff --git a/src/snowflake/cli/_plugins/dcm/manager.py b/src/snowflake/cli/_plugins/dcm/manager.py index e77cdb5202..8b777b3028 100644 --- a/src/snowflake/cli/_plugins/dcm/manager.py +++ b/src/snowflake/cli/_plugins/dcm/manager.py @@ -14,11 +14,10 @@ import logging from dataclasses import dataclass, field from pathlib import Path -from typing import List +from typing import TYPE_CHECKING, List, Optional from snowflake.cli._plugins.dcm.models import MANIFEST_FILE_NAME, SOURCES_FOLDER from snowflake.cli._plugins.dcm.utils import collect_output -from snowflake.cli._plugins.stage.diff import _to_diff_line from snowflake.cli._plugins.stage.manager import StageManager from snowflake.cli.api.artifacts.utils import bundle_artifacts from snowflake.cli.api.commands.utils import parse_key_value_variables @@ -35,6 +34,9 @@ from snowflake.cli.api.stage_path import StagePath from snowflake.connector.cursor import SnowflakeCursor +if TYPE_CHECKING: + from snowflake.cli._plugins.dcm.progress import DeployProgressTracker + log = logging.getLogger(__name__) @@ -52,6 +54,11 @@ class UploadPlan: class DCMProjectManager(SqlExecutionMixin): + @property + def connection(self): + """Exposes the underlying Snowflake connection.""" + return self._conn + def deploy( self, project_identifier: FQN, @@ -253,7 +260,9 @@ def _get_configuration_and_variables_query( @staticmethod def sync_local_files( - project_identifier: FQN, source_directory: str | None = None + project_identifier: FQN, + source_directory: str | None = None, + progress: Optional["DeployProgressTracker"] = None, ) -> str: source_path = ( SecurePath(source_directory).resolve() @@ -265,57 +274,86 @@ def sync_local_files( project_identifier, source_path, ) + return DCMProjectManager._sync_local_files_impl( + project_identifier=project_identifier, + source_path=source_path, + progress=progress, + ) + + @staticmethod + def _sync_local_files_impl( + project_identifier: FQN, + source_path: SecurePath, + progress: Optional["DeployProgressTracker"], + ) -> str: + stage_fqn = FQN.from_resource( + ObjectType.DCM_PROJECT, project_identifier, "TMP_STAGE" + ) + plan = DCMProjectManager._build_upload_plan( + source_path.path, stage_fqn.identifier + ) - with cli_console.phase("Uploading definition files"): - stage_fqn = FQN.from_resource( - ObjectType.DCM_PROJECT, project_identifier, "TMP_STAGE" + project_paths = ProjectPaths(project_root=source_path.path) + project_paths.remove_up_bundle_root() + SecurePath(project_paths.bundle_root).mkdir(parents=True, exist_ok=True) + + def _set_upload_details() -> None: + parent_scope = project_identifier.prefix or str(project_identifier) + stage_message = f"Creating temporary stage inside {parent_scope}." + file_summaries = DCMProjectManager._summarize_upload_paths( + plan.relative_paths_to_upload ) - plan = DCMProjectManager._build_upload_plan( - source_path.path, stage_fqn.identifier + if progress: + progress.set_upload_context( + stage_message=stage_message, + file_summaries=file_summaries, + ) + else: + cli_console.step(stage_message) + for summary in file_summaries: + cli_console.step(summary) + + try: + bundle_artifacts( + project_paths, + plan.artifacts, + pattern_type=PatternMatchingType.GLOB, ) - project_paths = ProjectPaths(project_root=source_path.path) - project_paths.remove_up_bundle_root() - SecurePath(project_paths.bundle_root).mkdir(parents=True, exist_ok=True) + _set_upload_details() - try: - bundle_artifacts( - project_paths, - plan.artifacts, - pattern_type=PatternMatchingType.GLOB, - ) + if progress: + progress.set_upload_file_total(len(plan.relative_paths_to_upload)) - stage_manager = StageManager() - cli_console.step(f"Creating temporary stage {stage_fqn.identifier}.") - stage_manager.create( - fqn=FQN.from_stage(stage_fqn.identifier), temporary=True - ) + stage_manager = StageManager() + stage_manager.create( + fqn=FQN.from_stage(stage_fqn.identifier), temporary=True + ) - DCMProjectManager._report_files_to_be_deployed(plan) + for result in stage_manager.put_recursive( + local_path=project_paths.bundle_root, + stage_path=stage_fqn.identifier, + temp_directory=project_paths.bundle_root, + ): + if progress: + progress.advance_upload() + log.info( + "Uploaded %s to %s", + result["source"], + result["target"], + ) - cli_console.step( - f"Uploading files from local {project_paths.bundle_root} directory to temporary stage." + for entry in plan.individual_files: + stage_manager.put(local_path=entry.file, stage_path=entry.dest) + if progress: + progress.advance_upload() + log.info( + "Uploaded %s to %s", + entry.file.relative_to(source_path.path), + entry.dest, ) - for result in stage_manager.put_recursive( - local_path=project_paths.bundle_root, - stage_path=stage_fqn.identifier, - temp_directory=project_paths.bundle_root, - ): - log.info( - "Uploaded %s to %s", - result["source"], - result["target"], - ) - - for entry in plan.individual_files: - stage_manager.put(local_path=entry.file, stage_path=entry.dest) - log.info( - "Uploaded %s to %s", - entry.file.relative_to(source_path.path), - entry.dest, - ) - finally: - project_paths.clean_up_output() + finally: + project_paths.clean_up_output() log.info( "Finished syncing DCM files (project_identifier=%s, stage=%s).", @@ -365,11 +403,40 @@ def _sources_stage_destination(relative: Path, stage_root: str) -> str: return dest_dir @staticmethod - def _report_files_to_be_deployed(plan: UploadPlan) -> None: - if not plan.relative_paths_to_upload: - return - - cli_console.message("Local changes to be deployed:") - with cli_console.indented(): - for rel in plan.relative_paths_to_upload: - cli_console.message(_to_diff_line("added", rel, rel)) + def _summarize_upload_paths(relative_paths: List[str]) -> List[str]: + """Summarize files to upload, grouped by sources/ subfolder.""" + if not relative_paths: + return [] + + manifest_count = 0 + folder_counts: dict[str, int] = {} + + for rel in relative_paths: + if rel == MANIFEST_FILE_NAME: + manifest_count += 1 + elif rel.startswith(f"{SOURCES_FOLDER}/"): + parts = Path(rel).parts + folder = ( + "/".join(parts[:2]) + "/" + if len(parts) >= 3 + else f"{SOURCES_FOLDER}/" + ) + folder_counts[folder] = folder_counts.get(folder, 0) + 1 + else: + folder_counts[rel] = folder_counts.get(rel, 0) + 1 + + def _folder_label(folder: str) -> str: + if folder == f"{SOURCES_FOLDER}/": + return folder + return folder.rstrip("/") + + lines: List[str] = [] + if manifest_count: + lines.append(f"Upload {MANIFEST_FILE_NAME}") + for folder in sorted(folder_counts): + count = folder_counts[folder] + # Pad the singular "file" with a trailing space so it aligns with + # the plural "files" on adjacent rows in the upload details block. + file_word = "file " if count == 1 else "files" + lines.append(f"Upload {count} {file_word} from {_folder_label(folder)}") + return lines diff --git a/src/snowflake/cli/_plugins/dcm/progress.py b/src/snowflake/cli/_plugins/dcm/progress.py new file mode 100644 index 0000000000..3dfeaaffcd --- /dev/null +++ b/src/snowflake/cli/_plugins/dcm/progress.py @@ -0,0 +1,632 @@ +# 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. +"""Live phase-checklist progress display for DCM deploy operations.""" + +from __future__ import annotations + +import json +import logging +import re +import time +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime +from enum import Enum +from typing import Callable, Iterator, List, Literal, Optional, Sequence + +from rich import get_console +from rich.live import Live +from rich.text import Text +from snowflake.cli._plugins.dcm import styles +from snowflake.cli.api.cli_global_context import get_cli_context +from snowflake.connector import SnowflakeConnection +from snowflake.connector.constants import QueryStatus +from snowflake.connector.cursor import SnowflakeCursor + +log = logging.getLogger(__name__) + +UPLOAD_PHASE = "UPLOAD" +BACKEND_PHASES = ["RENDER", "COMPILE", "PLAN", "DEPLOY"] +PLAN_OPERATION_PHASES = ["RENDER", "COMPILE", "PLAN"] +# The "compile" operation runs the server-side ANALYZE statement, but ANALYZE +# is an implementation detail and is never shown as a phase. The user sees the +# same RENDER → COMPILE vocabulary as the other operations; it simply stops +# 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"] + +# 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. +_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], +} + +# Only PLAN and DEPLOY emit a meaningful 0-100 progress value. +PHASES_WITH_PROGRESS_BAR = {"PLAN", "DEPLOY"} + +# Per-operation display aliases for phase names. The internal phase name must +# stay aligned with what ``SYSTEM$GET_DCM_PROJECT_PROGRESS`` reports (purge +# runs the standard DEPLOY phase server-side), but the user-facing label can +# differ — purge shows ``PURGE`` instead of ``DEPLOY``. +_DISPLAY_NAME_OVERRIDES: dict[OperationMode, dict[str, str]] = { + "purge": {"DEPLOY": "PURGE"}, +} + +_FAST_POLL_INTERVAL = 1.0 # seconds — used for the first 100 s +_SLOW_POLL_INTERVAL = 10.0 # seconds — used after 100 s (~85 % fewer calls) +_FAST_POLL_THRESHOLD = 100.0 # seconds + +_PHASE_COL_WIDTH = 10 +_BAR_WIDTH = 40 +_LIVE_REFRESH_PER_SECOND = 10 # ~100 ms per repaint; smooth spinner cadence. + +# Heavy-horizontal box-drawing chars for a minimal pip/rich-style progress +# bar: ``━`` cells for both the filled and empty portions (colored vs dim +# to distinguish them) and a ``╺`` "leading edge" cell at the boundary +# while in progress. +_BAR_CELL = "━" +_BAR_LEADING_EDGE = "╺" + +# Braille-dot spinner frames; one frame per 100 ms gives a full cycle each +# second when paired with ``_LIVE_REFRESH_PER_SECOND``. +_SPINNER_FRAMES = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏" + + +def _spinner_glyph() -> str: + """Pick the current braille spinner glyph from wall-clock time. + + Rich's :class:`Live` re-renders the tracker on every refresh tick; + deriving the frame from ``time.monotonic()`` is what turns that into a + smooth animation without us needing to bookkeep a frame counter. + """ + return _SPINNER_FRAMES[ + int(time.monotonic() * _LIVE_REFRESH_PER_SECOND) % len(_SPINNER_FRAMES) + ] + + +class PhaseStatus(str, Enum): + """Lifecycle state of a single phase in the live checklist.""" + + PENDING = "pending" + RUNNING = "running" + DONE = "done" + FAILED = "failed" + + +_TERMINAL_STATUSES = { + QueryStatus.FAILED_WITH_ERROR, + QueryStatus.FAILED_WITH_INCIDENT, + QueryStatus.ABORTED, + QueryStatus.ABORTING, +} + +# Snowflake query IDs are UUIDs (``xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx``). +# ``_poll_progress`` interpolates the sfqid into a SQL literal; this regex +# is the gate that keeps anything that doesn't look like a query id out of +# that SQL string. Defense-in-depth: the sfqid is server-generated, but we +# never want to embed an unvalidated string into SQL. +_SFQID_RE = re.compile( + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-" r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$" +) + + +def _format_duration(seconds: float) -> str: + if seconds < 60: + return f"{seconds:.1f}s" + minutes = int(seconds // 60) + secs = seconds % 60 + if secs < 0.05: + return f"{minutes}m" + return f"{minutes}m {secs:.0f}s" + + +@dataclass +class _Phase: + name: str + status: PhaseStatus = PhaseStatus.PENDING + progress: int = 0 # 0-100; only meaningful for PHASES_WITH_PROGRESS_BAR + started_at: Optional[datetime] = None + completed_at: Optional[datetime] = None + hide_timing: bool = False # Suppress duration display (used for simulated phases) + + def observe_running(self, progress: int, ts: datetime) -> None: + if self.status == PhaseStatus.PENDING and not self.hide_timing: + self.started_at = ts + self.status = PhaseStatus.RUNNING + self.progress = progress + + def observe_done(self, ts: datetime) -> None: + if not self.hide_timing: + if not self.started_at: + self.started_at = ts + self.completed_at = ts + self.status = PhaseStatus.DONE + + def observe_failed(self, ts: Optional[datetime] = None) -> None: + end = ts or datetime.now() + if not self.started_at: + self.started_at = end + self.completed_at = end + self.status = PhaseStatus.FAILED + + def duration_seconds(self, *, now: Optional[datetime] = None) -> Optional[float]: + if not self.started_at: + return None + end = self.completed_at or now + if end is None: + return None + return (end - self.started_at).total_seconds() + + +class DeployProgressTracker: + """ + Renders a live phase checklist for DCM upload and deploy operations. + + Phases: UPLOAD (client-side) → RENDER → COMPILE → PLAN → DEPLOY (server-side). + + Usage:: + + tracker = DeployProgressTracker(conn=manager.connection) + with tracker.session(): + stage = manager.sync_local_files(..., progress=tracker) + sfqid = manager.deploy_async(...) + result = tracker.run_deploy_poll(sfqid) + """ + + def __init__( + self, + conn: SnowflakeConnection, + *, + operation: OperationMode = "deploy", + ) -> None: + self._conn = conn + self._sfqid: Optional[str] = None + self._operation = operation + phase_names = _PHASES_BY_OPERATION[operation] + self._phases = [_Phase(name=p) for p in phase_names] + self._upload_stage_message: str = "" + self._upload_file_summaries: List[str] = [] + self._upload_file_total = 0 + self._upload_files_done = 0 + self._live: Optional[Live] = None + + def __rich__(self) -> Text: + """Render the current tracker frame. + + Implementing ``__rich__`` lets us pass ``self`` to :class:`rich.live.Live` + instead of a static snapshot. Rich's refresh loop then calls this + method on every tick, so spinner animation comes "for free" — no + manual repaint is needed for in-place phase updates. + """ + return self._render() + + def _get_phase(self, name: str) -> _Phase: + for phase in self._phases: + if phase.name == name: + return phase + raise KeyError(name) + + def _has_upload_phase(self) -> bool: + return any(p.name == UPLOAD_PHASE for p in self._phases) + + def _refresh_display(self) -> None: + """Force an immediate Live repaint (state changed; don't wait for tick).""" + if self._live is not None: + self._live.refresh() + + # ------------------------------------------------------------------ # + # Upload phase (client-side) # + # ------------------------------------------------------------------ # + + def start_upload(self) -> None: + self._get_phase(UPLOAD_PHASE).observe_running(0, datetime.now()) + self._refresh_display() + + def set_upload_context( + self, *, stage_message: str, file_summaries: list[str] + ) -> None: + self._upload_stage_message = stage_message + self._upload_file_summaries = file_summaries + self._refresh_display() + + def set_upload_file_total(self, total: int) -> None: + """Set expected file count (from the upload plan, no extra I/O).""" + self._upload_file_total = max(0, total) + self._upload_files_done = 0 + if self._upload_file_total > 0: + self._get_phase(UPLOAD_PHASE).observe_running(0, datetime.now()) + self._refresh_display() + + def _upload_percent(self) -> int: + if self._upload_file_total <= 0: + return 0 + return min(100, int(100 * self._upload_files_done / self._upload_file_total)) + + def advance_upload(self, count: int = 1) -> None: + """Increment upload progress; refreshes only when the percent changes.""" + if self._upload_file_total <= 0: + return + prev_percent = self._upload_percent() + self._upload_files_done = min( + self._upload_files_done + count, self._upload_file_total + ) + new_percent = self._upload_percent() + if new_percent != prev_percent: + self._get_phase(UPLOAD_PHASE).observe_running(new_percent, datetime.now()) + self._refresh_display() + + def complete_upload(self) -> None: + 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()) + self._refresh_display() + + def fail_upload(self) -> None: + """Mark UPLOAD as failed only if it has not already terminated. + + A failure raised by a downstream phase (RENDER/COMPILE/PLAN/DEPLOY) + bubbles up through :meth:`session`'s exception handler; UPLOAD itself + was already ``done`` at that point and must not be reset to ``failed``. + """ + upload = self._get_phase(UPLOAD_PHASE) + if upload.status in (PhaseStatus.DONE, PhaseStatus.FAILED): + return + upload.observe_failed() + self._refresh_display() + + @contextmanager + def session(self) -> Iterator["DeployProgressTracker"]: + """Keeps a single Live display open for upload and deploy polling.""" + if get_cli_context().silent: + yield self + return + + console = get_console() + if self._has_upload_phase(): + self.start_upload() + with Live( + self, console=console, refresh_per_second=_LIVE_REFRESH_PER_SECOND + ) as live: + self._live = live + try: + yield self + except Exception: + if self._has_upload_phase(): + self.fail_upload() + live.refresh() + raise + finally: + self._live = None + + # ------------------------------------------------------------------ # + # Query-status helpers # + # ------------------------------------------------------------------ # + + def _is_still_running(self) -> bool: + assert self._sfqid is not None + status = self._conn.get_query_status(self._sfqid) + return self._conn.is_still_running(status) + + def _query_failed(self) -> bool: + assert self._sfqid is not None + return self._conn.get_query_status(self._sfqid) in _TERMINAL_STATUSES + + # ------------------------------------------------------------------ # + # Progress polling # + # ------------------------------------------------------------------ # + + def _poll_progress(self) -> Optional[dict]: + assert self._sfqid is not None + # ``sfqid`` is server-generated, but we refuse to interpolate + # anything that doesn't match the UUID shape into SQL. A failed + # match disables polling for this run instead of risking SQL + # injection from a malformed identifier. + if not _SFQID_RE.match(self._sfqid): + log.debug( + "Refusing to poll progress: sfqid=%r does not match expected shape.", + self._sfqid, + ) + return None + try: + with self._conn.cursor() as cursor: + cursor.execute( + f"SELECT SYSTEM$GET_DCM_PROJECT_PROGRESS('{self._sfqid}')" + ) + row = cursor.fetchone() + if row and row[0]: + return json.loads(row[0]) + except Exception: # noqa: BLE001 + log.debug( + "Progress poll failed for sfqid=%s; will retry next cycle.", + self._sfqid, + exc_info=True, + ) + return None + + def _update_from_poll(self, data: Optional[dict]) -> None: + if not data: + return + raw_phase = data.get("phase") + current_phase_name = raw_phase.upper() if isinstance(raw_phase, str) else "" + # Defensive: refuse to mutate phase state from a payload that doesn't + # name a known phase. Without this guard an empty / missing / unknown + # ``phase`` field flips every server-side phase to ``done`` on the + # first poll while the query is still running (silent UI regression). + if current_phase_name not in {p.name for p in self._phases}: + log.debug( + "Ignoring progress poll with unknown phase %r (sfqid=%s).", + raw_phase, + self._sfqid, + ) + return + + progress = int(data.get("progress", 0)) + ts = datetime.now() + + found_current = False + for phase in self._phases: + if phase.name == UPLOAD_PHASE: + continue + if phase.name == current_phase_name: + found_current = True + bar_progress = progress if phase.name in PHASES_WITH_PROGRESS_BAR else 0 + phase.observe_running(bar_progress, ts) + elif not found_current: + if phase.status not in (PhaseStatus.DONE, PhaseStatus.FAILED): + phase.observe_done(ts) + + # ------------------------------------------------------------------ # + # Finalisation # + # ------------------------------------------------------------------ # + + def _finalize_success(self) -> None: + ts = datetime.now() + for phase in self._phases: + if phase.name == UPLOAD_PHASE: + continue + if phase.status in (PhaseStatus.PENDING, PhaseStatus.RUNNING): + phase.observe_done(ts) + + def _finalize_failure(self) -> None: + ts = datetime.now() + for phase in reversed(self._phases): + if phase.name == UPLOAD_PHASE: + continue + if phase.status == PhaseStatus.RUNNING: + phase.observe_failed(ts) + return + for phase in self._phases: + if phase.name == UPLOAD_PHASE: + continue + if phase.status == PhaseStatus.PENDING: + phase.observe_failed(ts) + return + + # ------------------------------------------------------------------ # + # Rendering # + # ------------------------------------------------------------------ # + + def _duration_suffix(self, phase: _Phase) -> str: + if phase.hide_timing: + return "" + now = datetime.now() if phase.status == PhaseStatus.RUNNING else None + seconds = phase.duration_seconds(now=now) + if seconds is None: + return "" + return f" ({_format_duration(seconds)})" + + def _simulate_instant_phase(self, name: str) -> None: + phase = self._get_phase(name) + phase.hide_timing = True + phase.status = PhaseStatus.DONE + + def _phase_shows_progress_bar(self, phase: _Phase) -> bool: + if phase.name == UPLOAD_PHASE: + return self._upload_file_total > 0 + if phase.name in PHASES_WITH_PROGRESS_BAR: + return self._operation in ("deploy", "purge") + return False + + def _append_progress_bar(self, out: Text, progress: int) -> None: + """Render a pip-style minimal progress bar. + + Heavy-horizontal cells (``━``) in blue for the completed portion, a + ``╺`` leading-edge cell at the boundary while in progress, and dim + ``━`` cells for the remaining portion — followed by " NNN%" in + blue. No surrounding brackets. Matches the look of pip/rich's + ``BarColumn`` so the UI feels familiar. + """ + filled = int(_BAR_WIDTH * progress / 100) + in_progress = 0 < filled < _BAR_WIDTH + + if in_progress: + out.append(_BAR_CELL * filled + _BAR_LEADING_EDGE, style=styles.BLUE) + out.append(_BAR_CELL * (_BAR_WIDTH - filled - 1), style="dim") + elif filled == 0: + out.append(_BAR_CELL * _BAR_WIDTH, style="dim") + else: + out.append(_BAR_CELL * _BAR_WIDTH, style=styles.BLUE) + + out.append(f" {progress:>3}%", style=styles.BLUE) + + def _append_upload_details(self, out: Text) -> None: + """Render the stage-creation message and folder counters indented + beneath the UPLOAD phase line (dim, two-space indent).""" + if self._upload_stage_message: + out.append(f" {self._upload_stage_message}\n", style="dim") + for summary in self._upload_file_summaries: + out.append(f" {summary}\n", style="dim") + + def _display_name(self, phase: _Phase) -> str: + """User-facing label for a phase. + + Defaults to the internal phase name but can be overridden per + operation (see :data:`_DISPLAY_NAME_OVERRIDES`) so that, e.g., purge + renders ``PURGE`` while still tracking the server's ``DEPLOY`` phase. + """ + overrides = _DISPLAY_NAME_OVERRIDES.get(self._operation, {}) + return overrides.get(phase.name, phase.name) + + def _render_phase_line(self, out: Text, phase: _Phase) -> None: + # Wall-clock start time is intentionally omitted — the parenthesised + # elapsed duration that follows the status indicator is the single + # source of timing information. + duration_str = self._duration_suffix(phase) + name_col = f"{self._display_name(phase):<{_PHASE_COL_WIDTH}}" + + if phase.status == PhaseStatus.DONE: + out.append(name_col, style=styles.PHASE_DONE_STYLE) + out.append("✓", style="bold green") + out.append(duration_str + "\n", style="dim") + elif phase.status == PhaseStatus.FAILED: + out.append(name_col, style=styles.PHASE_FAILED_STYLE) + out.append("✗", style="bold red") + out.append(duration_str + "\n", style="dim") + elif phase.status == PhaseStatus.RUNNING: + out.append(name_col, style=styles.PHASE_RUNNING_STYLE) + if self._phase_shows_progress_bar(phase): + self._append_progress_bar(out, phase.progress) + else: + # Animated braille spinner; Rich's Live re-renders us each + # refresh tick (see :meth:`__rich__`), so the glyph cycles. + # Blue matches the active-indicator color used by the + # pip-style progress bar so all "this phase is in flight" + # signals share the same hue. + out.append(_spinner_glyph(), style=styles.BLUE) + out.append(duration_str + "\n", style="dim") + else: # PENDING + out.append(name_col + "·\n", style="dim") + + def _render(self) -> Text: + out = Text("\n") + for phase in self._phases: + self._render_phase_line(out, phase) + if phase.name == UPLOAD_PHASE and ( + self._upload_stage_message or self._upload_file_summaries + ): + self._append_upload_details(out) + return out + + # ------------------------------------------------------------------ # + # Main entry points # + # ------------------------------------------------------------------ # + + @contextmanager + def _ensure_live(self) -> Iterator[Optional[Live]]: + """Yield a Live instance to drive repaints, or ``None`` when silent. + + Three cases collapse into one: + + * silent → yield ``None``; callers skip repaints entirely. + * a session is already active → reuse ``self._live``. + * standalone call → open a fresh Live around this block. + """ + if get_cli_context().silent: + yield None + return + if self._live is not None: + yield self._live + return + with Live( + self, + console=get_console(), + refresh_per_second=_LIVE_REFRESH_PER_SECOND, + ) as live: + previous = self._live + self._live = live + try: + yield live + finally: + self._live = previous + + def run_deploy_poll(self, sfqid: str) -> SnowflakeCursor: + """Poll server-side deploy progress until the query finishes. + + Call inside :meth:`session` after upload completes (if any). + """ + self._sfqid = sfqid + if self._has_upload_phase(): + self.complete_upload() + + start = time.monotonic() + with self._ensure_live() as live: + while self._is_still_running(): + self._update_from_poll(self._poll_progress()) + elapsed = time.monotonic() - start + interval = ( + _FAST_POLL_INTERVAL + if elapsed < _FAST_POLL_THRESHOLD + else _SLOW_POLL_INTERVAL + ) + time.sleep(interval) + + if self._query_failed(): + self._finalize_failure() + else: + self._finalize_success() + if live is not None: + live.refresh() + + result_cursor = self._conn.cursor() + result_cursor.get_results_from_sfqid(self._sfqid) + return result_cursor + + def run_loader_phase( + self, + execute_fn: Callable[[], SnowflakeCursor], + *, + phase_name: str, + simulated_phases: Sequence[str] = (), + ) -> SnowflakeCursor: + """Run a blocking SQL operation while the spinner animates in place. + + The phase named ``phase_name`` is marked running before ``execute_fn`` + is invoked; the braille spinner glyph cycles next to it via Rich's + Live auto-refresh (see :meth:`__rich__`). On success the phase + transitions to done with a real duration; on exception it transitions + to failed. + + ``simulated_phases`` are marked instantly done first — used by ``plan`` + to fast-forward ``RENDER`` / ``COMPILE`` before settling on ``PLAN``. + """ + # Silent mode (JSON/CSV output, or ``--silent``) has no UI to drive, + # so we skip all phase-state bookkeeping entirely. This keeps the + # tracker free of stale RUNNING entries that could mislead later + # introspection — and avoids paying for ``datetime.now()`` calls + # nobody will see. + if get_cli_context().silent: + return execute_fn() + + self.complete_upload() + for name in simulated_phases: + self._simulate_instant_phase(name) + + phase = self._get_phase(phase_name) + phase.observe_running(0, datetime.now()) + self._refresh_display() + + try: + result = execute_fn() + except Exception: + phase.observe_failed(datetime.now()) + raise + else: + phase.observe_done(datetime.now()) + return result diff --git a/src/snowflake/cli/_plugins/dcm/styles.py b/src/snowflake/cli/_plugins/dcm/styles.py index bb68e13bcc..506d5dd99e 100644 --- a/src/snowflake/cli/_plugins/dcm/styles.py +++ b/src/snowflake/cli/_plugins/dcm/styles.py @@ -13,13 +13,17 @@ # limitations under the License. from rich.style import Style -_COLOR_BLUE = "#a0a8fe" +# Single source of truth for the DCM "blue" hue. Use the terminal-default +# named color (not a fixed hex) so it respects the user's theme and stays +# identical everywhere it's used: progress spinner/bar, running phase, +# refresh status, and unknown rows. +BLUE = "blue" DOMAIN_STYLE = Style(color="cyan") BOLD_STYLE = Style(bold=True) # Refresh -STATUS_STYLE = Style(color=_COLOR_BLUE) +STATUS_STYLE = Style(color=BLUE) REMOVED_STYLE = Style(color="red", italic=True) INSERTED_STYLE = Style(color="green", italic=True) @@ -31,4 +35,9 @@ CREATE_STYLE = Style(color="green") ALTER_STYLE = Style(color="yellow") DROP_STYLE = Style(color="red") -UNKNOWN_STYLE = Style(color=_COLOR_BLUE) +UNKNOWN_STYLE = Style(color=BLUE) + +# Deploy progress phases +PHASE_DONE_STYLE = Style(color="green", bold=True) +PHASE_RUNNING_STYLE = Style(color=BLUE, bold=True) +PHASE_FAILED_STYLE = Style(color="red", bold=True) diff --git a/tests/dcm/test_manager.py b/tests/dcm/test_manager.py index 5336be9a22..0e3cb559d2 100644 --- a/tests/dcm/test_manager.py +++ b/tests/dcm/test_manager.py @@ -758,3 +758,25 @@ def test_sync_local_files_uploads_hidden_files( assert any( filename in q and stage_dest in q for q in put_queries ), f"expected a PUT for {filename} to {stage_dest}; got: {put_queries}" + + +class TestSummarizeUploadPaths: + def test_groups_by_sources_subfolder(self): + paths = [ + MANIFEST_FILE_NAME, + f"{SOURCES_FOLDER}/definitions/a.sql", + f"{SOURCES_FOLDER}/definitions/b.sql", + f"{SOURCES_FOLDER}/macros/m.sql", + f"{SOURCES_FOLDER}/.DS_Store", + ] + lines = DCMProjectManager._summarize_upload_paths(paths) # noqa: SLF001 + # Singular "file " is padded with a trailing space so adjacent rows + # line up vertically with the plural "files" rows in the upload + # details block. The literal expected strings below intentionally + # contain that double-space; do not "fix" it. + assert lines == [ + f"Upload {MANIFEST_FILE_NAME}", + f"Upload 1 file from {SOURCES_FOLDER}/", + f"Upload 2 files from {SOURCES_FOLDER}/definitions", + f"Upload 1 file from {SOURCES_FOLDER}/macros", + ] diff --git a/tests/dcm/test_progress.py b/tests/dcm/test_progress.py new file mode 100644 index 0000000000..7dfb98d588 --- /dev/null +++ b/tests/dcm/test_progress.py @@ -0,0 +1,327 @@ +# 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. +"""Unit tests for the DCM deploy/plan live progress tracker rendering.""" + +from datetime import datetime +from unittest import mock +from unittest.mock import MagicMock + +from snowflake.cli._plugins.dcm.progress import ( + _SPINNER_FRAMES, + UPLOAD_PHASE, + DeployProgressTracker, + PhaseStatus, +) + + +def _stripped_lines(text): + return [line for line in text.split("\n") if line.strip()] + + +class TestUploadDetailsLayout: + """The UPLOAD phase line comes first, followed by dim indented details.""" + + def _tracker(self, *, operation="deploy", with_context=True, file_total=3): + tracker = DeployProgressTracker(conn=MagicMock(), operation=operation) + if with_context: + tracker.set_upload_context( + stage_message="Creating temporary stage inside DCM_DEMO.PROJECTS.", + file_summaries=[ + "Upload manifest.yml", + "Upload 2 files from sources/definitions", + ], + ) + tracker.set_upload_file_total(file_total) + return tracker + + def test_upload_line_renders_before_details(self): + tracker = self._tracker() + + lines = _stripped_lines(tracker._render().plain) # noqa: SLF001 + + assert lines[0].startswith(UPLOAD_PHASE) + assert lines[1].lstrip() == "Creating temporary stage inside DCM_DEMO.PROJECTS." + assert lines[2].lstrip() == "Upload manifest.yml" + assert lines[3].lstrip() == "Upload 2 files from sources/definitions" + + def test_details_are_indented_two_spaces(self): + tracker = self._tracker() + + lines = _stripped_lines(tracker._render().plain) # noqa: SLF001 + + assert lines[0] == lines[0].lstrip() + for detail_line in lines[1:4]: + assert detail_line.startswith(" ") + assert not detail_line.startswith(" ") + + def test_subsequent_phases_render_below_upload_details(self): + tracker = self._tracker(operation="deploy") + + lines = _stripped_lines(tracker._render().plain) # noqa: SLF001 + + assert lines[0].startswith("UPLOAD") + assert lines[4].startswith("RENDER") + assert lines[5].startswith("COMPILE") + assert lines[6].startswith("PLAN") + assert lines[7].startswith("DEPLOY") + + def test_plan_mode_uses_plan_phase_list(self): + tracker = self._tracker(operation="plan") + + lines = _stripped_lines(tracker._render().plain) # noqa: SLF001 + + assert lines[0].startswith("UPLOAD") + assert lines[4].startswith("RENDER") + assert lines[5].startswith("COMPILE") + assert lines[6].startswith("PLAN") + assert all(not line.startswith("DEPLOY") for line in lines) + + def test_purge_mode_has_no_upload_and_renders_purge_label(self): + # purge runs entirely server-side: no UPLOAD phase, and the final + # DEPLOY phase is displayed as PURGE. + tracker = DeployProgressTracker(conn=MagicMock(), operation="purge") + + lines = _stripped_lines(tracker._render().plain) # noqa: SLF001 + + assert all(not line.startswith("UPLOAD") for line in lines) + assert lines[0].startswith("RENDER") + assert lines[1].startswith("COMPILE") + assert lines[2].startswith("PLAN") + assert lines[3].startswith("PURGE") + assert all(not line.startswith("DEPLOY") for line in lines) + + def test_running_progress_phase_shows_pip_style_bar(self): + """PLAN/DEPLOY phases that report 0–100 progress render a + heavy-horizontal pip-style bar with a ``╺`` leading edge while in + progress (no surrounding brackets, no block characters).""" + tracker = DeployProgressTracker(conn=MagicMock(), operation="deploy") + tracker.complete_upload() + # Set DEPLOY mid-progress so we get both filled and empty halves. + tracker._get_phase("DEPLOY").observe_running(50, datetime.now()) # noqa: SLF001 + + rendered = tracker._render().plain # noqa: SLF001 + deploy_line = next(line for line in rendered.split("\n") if "DEPLOY" in line) + + assert "━" in deploy_line + assert "╺" in deploy_line # the leading-edge transition cell + assert " 50%" in deploy_line + # Old block-style chars are gone. + assert "█" not in deploy_line + assert "░" not in deploy_line + + def test_running_no_progress_phase_shows_spinner_glyph(self): + """COMPILE (and PLAN in plan mode) have no progress bar — they show + an animated braille spinner where the running indicator goes.""" + tracker = DeployProgressTracker(conn=MagicMock(), operation="compile") + tracker.complete_upload() + # Mark COMPILE running, then verify a spinner glyph appears on its line. + tracker._get_phase("COMPILE").observe_running(0, datetime.now()) # noqa: SLF001 + + rendered = tracker._render().plain # noqa: SLF001 + compile_line = next(line for line in rendered.split("\n") if "COMPILE" in line) + + # The static "…" placeholder is gone; one of the braille spinner + # frames is on the line instead. + assert "…" not in compile_line + assert any(frame in compile_line for frame in _SPINNER_FRAMES) + + def test_compile_mode_renders_upload_render_compile(self): + tracker = self._tracker(operation="compile") + + lines = _stripped_lines(tracker._render().plain) # noqa: SLF001 + + assert lines[0].startswith("UPLOAD") + assert lines[4].startswith("RENDER") + assert lines[5].startswith("COMPILE") + # ANALYZE is an implementation detail and is never shown as a phase; + # neither are the later PLAN/DEPLOY phases. + assert all(not line.startswith(("PLAN", "DEPLOY", "ANALYZE")) for line in lines) + + def test_no_details_block_when_context_is_unset(self): + tracker = self._tracker(with_context=False) + + lines = _stripped_lines(tracker._render().plain) # noqa: SLF001 + + assert all(not line.startswith(" ") for line in lines) + + +class TestFailUpload: + """``fail_upload`` must only mark UPLOAD failed when UPLOAD itself failed. + + Downstream phase failures (PLAN, DEPLOY, …) bubble up through + ``session()``'s exception handler, which always calls ``fail_upload``. + UPLOAD has already been marked ``done`` by that point and must not be + flipped back to ``failed``. + """ + + def _tracker(self): + tracker = DeployProgressTracker(conn=MagicMock(), operation="deploy") + tracker.set_upload_file_total(2) + return tracker + + def _upload_phase(self, tracker): + return tracker._get_phase(UPLOAD_PHASE) # noqa: SLF001 + + def test_fail_upload_marks_running_upload_failed(self): + """If UPLOAD itself is still running, ``fail_upload`` flips it failed.""" + tracker = self._tracker() + tracker.start_upload() + assert self._upload_phase(tracker).status == PhaseStatus.RUNNING + + tracker.fail_upload() + + assert self._upload_phase(tracker).status == PhaseStatus.FAILED + + def test_fail_upload_is_noop_when_upload_already_done(self): + """A downstream PLAN/DEPLOY failure must not flip a finished UPLOAD.""" + tracker = self._tracker() + tracker.start_upload() + tracker.complete_upload() + assert self._upload_phase(tracker).status == PhaseStatus.DONE + + tracker.fail_upload() + + assert self._upload_phase(tracker).status == PhaseStatus.DONE + + def test_fail_upload_is_idempotent_when_already_failed(self): + tracker = self._tracker() + tracker.start_upload() + tracker.fail_upload() + first_completed_at = self._upload_phase(tracker).completed_at + assert first_completed_at is not None + + tracker.fail_upload() + + assert self._upload_phase(tracker).status == PhaseStatus.FAILED + # The original failure timestamp is preserved on repeated calls. + assert self._upload_phase(tracker).completed_at == first_completed_at + + +class TestUpdateFromPoll: + """``_update_from_poll`` must not mutate phase state from junk payloads. + + Regression coverage for an earlier bug where an empty / missing / + unknown ``phase`` field on the very first poll flipped every server-side + phase to ``done`` while the query was still running — the UI showed a + fully-green checklist immediately. + """ + + def _tracker(self): + t = DeployProgressTracker(conn=MagicMock(), operation="deploy") + t.complete_upload() + return t + + def _server_phases(self, tracker): + return [p for p in tracker._phases if p.name != UPLOAD_PHASE] # noqa: SLF001 + + def test_empty_phase_string_is_ignored(self): + tracker = self._tracker() + + tracker._update_from_poll({"phase": "", "progress": 0}) # noqa: SLF001 + + assert all( + p.status == PhaseStatus.PENDING for p in self._server_phases(tracker) + ) + + def test_missing_phase_key_is_ignored(self): + tracker = self._tracker() + + tracker._update_from_poll({"progress": 0}) # noqa: SLF001 + + assert all( + p.status == PhaseStatus.PENDING for p in self._server_phases(tracker) + ) + + def test_unknown_phase_name_is_ignored(self): + tracker = self._tracker() + + tracker._update_from_poll({"phase": "BUILD", "progress": 50}) # noqa: SLF001 + + assert all( + p.status == PhaseStatus.PENDING for p in self._server_phases(tracker) + ) + + def test_non_string_phase_is_ignored(self): + tracker = self._tracker() + + tracker._update_from_poll({"phase": 42, "progress": 0}) # noqa: SLF001 + + assert all( + p.status == PhaseStatus.PENDING for p in self._server_phases(tracker) + ) + + def test_known_phase_advances_state(self): + """Sanity: a well-formed payload still drives the state machine.""" + tracker = self._tracker() + + tracker._update_from_poll({"phase": "PLAN", "progress": 25}) # noqa: SLF001 + + statuses = {p.name: p.status for p in self._server_phases(tracker)} + # Phases earlier than PLAN are auto-completed; PLAN itself runs; + # later phases remain PENDING. + assert statuses["RENDER"] == PhaseStatus.DONE + assert statuses["COMPILE"] == PhaseStatus.DONE + assert statuses["PLAN"] == PhaseStatus.RUNNING + assert statuses["DEPLOY"] == PhaseStatus.PENDING + + +class TestRunLoaderPhaseSilent: + """In silent mode (JSON/CSV/``--silent``), ``run_loader_phase`` must not + mutate phase state — there's no UI to drive and stale RUNNING entries + would mislead any code that introspects the tracker later.""" + + def _tracker(self, operation="plan"): + return DeployProgressTracker(conn=MagicMock(), operation=operation) + + def test_silent_mode_skips_phase_transitions(self): + tracker = self._tracker(operation="plan") + before = [(p.name, p.status) for p in tracker._phases] # noqa: SLF001 + + with mock.patch("snowflake.cli._plugins.dcm.progress.get_cli_context") as ctx: + ctx.return_value.silent = True + sentinel = object() + result = tracker.run_loader_phase( + lambda: sentinel, + phase_name="PLAN", + simulated_phases=("RENDER", "COMPILE"), + ) + + assert result is sentinel + after = [(p.name, p.status) for p in tracker._phases] # noqa: SLF001 + assert before == after # no phases touched + + def test_silent_mode_still_propagates_exceptions(self): + tracker = self._tracker(operation="plan") + + class _BoomError(RuntimeError): + pass + + def _raise(): + raise _BoomError("server-side failure") + + with mock.patch("snowflake.cli._plugins.dcm.progress.get_cli_context") as ctx: + ctx.return_value.silent = True + try: + tracker.run_loader_phase(_raise, phase_name="PLAN") + except _BoomError: + pass + else: + raise AssertionError("Expected _BoomError") + + # Even though the call raised, no phase state was mutated. + assert all( + p.status == PhaseStatus.PENDING + for p in tracker._phases # noqa: SLF001 + if p.name != UPLOAD_PHASE + )