diff --git a/src/sqlbuild/cli/commands/_helpers/entry/observability.py b/src/sqlbuild/cli/commands/_helpers/entry/observability.py index 8e6c05f867..3da7bf8c49 100644 --- a/src/sqlbuild/cli/commands/_helpers/entry/observability.py +++ b/src/sqlbuild/cli/commands/_helpers/entry/observability.py @@ -90,6 +90,7 @@ def cli_observability_scope(*, args: CliNamespace, project_dir: Path) -> Iterato if machine_output or args.debug or args.command == DEBUG_COMMAND else sys.stdout, use_color=not machine_output and not args.no_color and supports_color(), + debug=args.debug, ) unsubscribe_progress: Unsubscribe = dispatcher.subscribe_lifecycle( subscriber=projector.consume, diff --git a/src/sqlbuild/cli/commands/_helpers/seed/execution.py b/src/sqlbuild/cli/commands/_helpers/seed/execution.py index de021fb4bd..3806f4d4d2 100644 --- a/src/sqlbuild/cli/commands/_helpers/seed/execution.py +++ b/src/sqlbuild/cli/commands/_helpers/seed/execution.py @@ -4,6 +4,7 @@ import time from collections.abc import Callable +from dataclasses import replace from typing import TextIO from sqlbuild.cli.commands.models import ( @@ -13,6 +14,10 @@ ) from sqlbuild.cli.output.classes.execution_event_writer import ExecutionEventWriter from sqlbuild.cli.progress.classes.connection_progress_reporter import ConnectionProgressReporter +from sqlbuild.cli.progress.classes.native_progress_projector import ( + NativeProgressProjector, + current_native_progress_projector, +) from sqlbuild.compiler.planner.models import PlanOutput from sqlbuild.executor.build.models import SeedExecutionResult from sqlbuild.executor.build.types import ExecutionStatus @@ -82,7 +87,17 @@ def _build_on_complete( event_writer: ExecutionEventWriter, plan: PlanOutput, ) -> Callable[[SeedExecutionResult], None]: + projector: NativeProgressProjector | None = current_native_progress_projector() + if projector is not None: + projector.configure_resources(ordinals=seed_order, total=total_count) + def _on_complete(result: SeedExecutionResult) -> None: + if projector is not None: + canonical_duration_ms: float | None = projector.consume_resource_terminal( + resource_name=result.seed_name, resource_id=f"seed:{result.seed_name}" + ) + if canonical_duration_ms is not None: + result = replace(result, duration_ms=int(canonical_duration_ms)) status_text: str = "OK" if result.status == ExecutionStatus.SUCCESS else "FAIL" style: CliStyle = CliStyle(use_color=use_color) status: str = style.status(status=status_text) diff --git a/src/sqlbuild/cli/progress/classes/native_progress_projector.py b/src/sqlbuild/cli/progress/classes/native_progress_projector.py index d9306be657..90469d686b 100644 --- a/src/sqlbuild/cli/progress/classes/native_progress_projector.py +++ b/src/sqlbuild/cli/progress/classes/native_progress_projector.py @@ -90,8 +90,9 @@ class NativeProgressProjector: """Project canonical starts and correlate enriched terminal presentation.""" - def __init__(self, *, stream: TextIO, use_color: bool) -> None: + def __init__(self, *, stream: TextIO, use_color: bool, debug: bool = False) -> None: self._stream: TextIO = stream + self._debug: bool = debug self._lines: TransientLineCoordinator = shared_transient_line_coordinator() self._style: CliStyle = CliStyle(use_color=use_color) self._is_tty: bool = hasattr(stream, "isatty") and stream.isatty() @@ -104,6 +105,7 @@ def __init__(self, *, stream: TextIO, use_color: bool) -> None: self._resource_ids_by_name: defaultdict[str, set[str]] = defaultdict(set) self._rendered_terminal_ids: set[str] = set() self._operation_starts: dict[str, LifecycleEvent] = {} + self._quiet_operation_ids: set[str] = set() self._enriched_resource_names: set[str] = set() self._resource_names: dict[str, str] = {} self._statement_starts: dict[str, LifecycleEvent] = {} @@ -249,7 +251,7 @@ def _consume_resource_start(self, event: LifecycleEvent) -> None: self._resource_names[resource_id] = resource_name self._render_prior_attempts(resource_id=resource_id, resource_name=resource_name) self._attempts_by_id[attempt_id] = event - if self._is_tty and resource_name in self._enriched_resource_names: + if self._is_tty and (resource_name in self._enriched_resource_names or not self._debug): return ordinal: int | None = self._resource_ordinals.get(resource_name) counter: str = "" @@ -367,6 +369,9 @@ def _consume_operation_start(self, event: LifecycleEvent) -> None: ): return self._operation_starts[operation_id] = event + if event.resource_attempt_id is not None and not self._debug: + self._quiet_operation_ids.add(operation_id) + return status: str = self._style.status(status="START") prefix: str = " " if event.resource_attempt_id is not None else "" attempt: str = _operation_attempt_label(event) @@ -378,9 +383,14 @@ def _consume_operation_terminal(self, event: LifecycleEvent) -> None: return start: LifecycleEvent = self._operation_starts.pop(operation_id) operation_name: object = start.payload.get("operation_name") + failed: bool = event.event_type.endswith("failed") + if operation_id in self._quiet_operation_ids: + self._quiet_operation_ids.discard(operation_id) + if not failed: + return if not isinstance(operation_name, str): return - status_text: str = "FAIL" if event.event_type.endswith("failed") else "OK" + status_text: str = "FAIL" if failed else "OK" status: str = self._style.status(status=status_text) duration: object = event.payload.get("duration_ms") elapsed: str = ( @@ -413,7 +423,8 @@ def _consume_statement_start(self, event: LifecycleEvent) -> None: if statement_id is None or event.resource_id not in self._resource_names: return self._statement_starts[statement_id] = event - self._write(f" statement {self._statement_context(event)} START") + if self._debug: + self._write(f" statement {self._statement_context(event)} START") def _consume_statement_submitted(self, event: LifecycleEvent) -> None: statement_id: str | None = event.statement_id @@ -427,6 +438,8 @@ def _consume_statement_submitted(self, event: LifecycleEvent) -> None: if self._statement_query_ids.get(statement_id) == query_id: return self._statement_query_ids[statement_id] = query_id + if not self._debug: + return start: LifecycleEvent = self._statement_starts.get(statement_id, event) self._write(f" statement {self._statement_context(start)} query_id={query_id}") diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py index b7a821740c..a8bb92c0fb 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py @@ -199,6 +199,14 @@ class TerminalTestProgressE2ETestCase: expected_summary: str +@dataclass(frozen=True) +class TerminalStatementRowsE2ETestCase: + description: str + columns: int + command: tuple[str, ...] + expected_statement_rows_visible: bool + + @dataclass(frozen=True) class ChainCteScopeTestCase: description: str @@ -222,3 +230,13 @@ class CommentedFixtureE2ETestCase: test_files: dict[str, str] expected_exit_code: int expected_output_fragments: tuple[str, ...] + + +@dataclass(frozen=True) +class TerminalStepRowsE2ETestCase: + description: str + columns: int + command: tuple[str, ...] + expected_absent_fragments: tuple[str, ...] + expected_absent_line_prefixes: tuple[str, ...] + expected_screen_fragments: tuple[str, ...] diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/test/test_terminal_progress.py b/tests/e2e/src/sqlbuild/cli/commands/main/test/test_terminal_progress.py index fe2eff34c0..8b91b4e77e 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/test/test_terminal_progress.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/test/test_terminal_progress.py @@ -8,6 +8,8 @@ import pytest from tests.e2e.src.sqlbuild.cli.commands.main.test._test_types import ( + TerminalStatementRowsE2ETestCase, + TerminalStepRowsE2ETestCase, TerminalTestProgressE2ETestCase, ) from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import ( @@ -19,6 +21,15 @@ run_sqb_with_pty, ) +_STEP_ROW_FRAGMENTS: tuple[str, ...] = ( + "START", + "Create staging relation", + "Promote relation", + "loader callable", + "Audit evaluation", + "SQL test assertion", +) + @pytest.mark.parametrize( "test_case", @@ -83,5 +94,129 @@ def test_given_built_project_when_testing_in_terminal_then_no_spinner_row_surviv assert test_case.expected_summary in rendered +@pytest.mark.parametrize( + "test_case", + [ + TerminalStatementRowsE2ETestCase( + description="test without debug shows no statement rows", + columns=120, + command=("test",), + expected_statement_rows_visible=False, + ), + TerminalStatementRowsE2ETestCase( + description="build without debug shows no statement rows", + columns=120, + command=("build",), + expected_statement_rows_visible=False, + ), + TerminalStatementRowsE2ETestCase( + description="test with debug shows statement start rows", + columns=120, + command=("--debug", "test"), + expected_statement_rows_visible=True, + ), + ], + ids=lambda case: case.description, +) +def test_given_built_project_when_running_in_terminal_then_statement_rows_follow_debug_flag( + tmp_path: Path, test_case: TerminalStatementRowsE2ETestCase +) -> None: + project_dir: Path = prepare_waffle_shop(tmp_path) + built: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "build"), project_dir=project_dir + ) + + result: subprocess.CompletedProcess[str] = run_sqb_with_pty( + command=test_case.command, + project_dir=project_dir, + columns=test_case.columns, + timeout_seconds=180.0, + ) + + assert built.returncode == 0, built.stdout + built.stderr + screen: tuple[str, ...] = render_terminal_screen( + output=result.stdout, columns=test_case.columns + ) + rendered: str = "\n".join(screen) + assert result.returncode == 0, rendered + statement_row_count: int = sum(line.lstrip().startswith("statement model=") for line in screen) + assert (statement_row_count > 0) is test_case.expected_statement_rows_visible, rendered + + +@pytest.mark.parametrize( + "test_case", + [ + TerminalStepRowsE2ETestCase( + description="passing build shows result rows without start or step rows", + columns=120, + command=("build",), + expected_absent_fragments=_STEP_ROW_FRAGMENTS, + expected_absent_line_prefixes=(), + expected_screen_fragments=( + "seed waffle_types", + "✓ Completed successfully", + ), + ), + TerminalStepRowsE2ETestCase( + description="passing seed shows one result row per seed", + columns=120, + command=("seed",), + expected_absent_fragments=_STEP_ROW_FRAGMENTS, + expected_absent_line_prefixes=(" seed waffle_types",), + expected_screen_fragments=("1/1 seed waffle_types", "✓ Completed successfully"), + ), + TerminalStepRowsE2ETestCase( + description="passing test shows no start or step rows", + columns=120, + command=("test",), + expected_absent_fragments=_STEP_ROW_FRAGMENTS, + expected_absent_line_prefixes=(), + expected_screen_fragments=("PASS=",), + ), + TerminalStepRowsE2ETestCase( + description="debug build shows start and step rows", + columns=120, + command=("--debug", "build"), + expected_absent_fragments=(), + expected_absent_line_prefixes=(), + expected_screen_fragments=( + "Create staging relation START", + "Create staging relation OK", + "loader callable START", + ), + ), + ], + ids=lambda case: case.description, +) +def test_given_built_project_when_running_in_terminal_then_step_rows_follow_debug_flag( + tmp_path: Path, test_case: TerminalStepRowsE2ETestCase +) -> None: + project_dir: Path = prepare_waffle_shop(tmp_path) + built: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "build"), project_dir=project_dir + ) + + result: subprocess.CompletedProcess[str] = run_sqb_with_pty( + command=test_case.command, + project_dir=project_dir, + columns=test_case.columns, + timeout_seconds=180.0, + ) + + assert built.returncode == 0, built.stdout + built.stderr + screen: tuple[str, ...] = render_terminal_screen( + output=result.stdout, columns=test_case.columns + ) + rendered: str = "\n".join(screen) + assert result.returncode == 0, rendered + assert not any(fragment in rendered for fragment in test_case.expected_absent_fragments), ( + rendered + ) + assert not any(line.startswith(test_case.expected_absent_line_prefixes) for line in screen), ( + rendered + ) + assert all(fragment in rendered for fragment in test_case.expected_screen_fragments), rendered + + if __name__ == "__main__": pytest.main([__file__, "-vv"]) diff --git a/tests/unit/src/sqlbuild/cli/progress/classes/_test_types.py b/tests/unit/src/sqlbuild/cli/progress/classes/_test_types.py index 04d62b57e6..bcb5f8ed7a 100644 --- a/tests/unit/src/sqlbuild/cli/progress/classes/_test_types.py +++ b/tests/unit/src/sqlbuild/cli/progress/classes/_test_types.py @@ -47,6 +47,7 @@ class StatementProgressCase: adapter: str query_id: str | None expected_context: str + debug: bool @dataclass(frozen=True) @@ -71,3 +72,11 @@ class FreshnessMetadataProjectionCase: terminal_event_type: str terminal_payload: dict[str, object] expected_terminal_line: str + + +@dataclass(frozen=True) +class StepProgressCase: + description: str + tty: bool + debug: bool + expected_line_prefixes: tuple[str, ...] diff --git a/tests/unit/src/sqlbuild/cli/progress/classes/helpers.py b/tests/unit/src/sqlbuild/cli/progress/classes/helpers.py index 144bfa2d26..a25ec85cf2 100644 --- a/tests/unit/src/sqlbuild/cli/progress/classes/helpers.py +++ b/tests/unit/src/sqlbuild/cli/progress/classes/helpers.py @@ -19,7 +19,10 @@ from sqlbuild.runtime.observability.classes.statement_lifecycle import StatementLifecycle from sqlbuild.runtime.observability.classes.statement_monitor import StatementMonitor from sqlbuild.runtime.observability.models import LifecycleEvent -from tests.unit.src.sqlbuild.cli.progress.classes._test_types import StatementProgressCase +from tests.unit.src.sqlbuild.cli.progress.classes._test_types import ( + StatementProgressCase, + StepProgressCase, +) class FakeSlowAdapter: @@ -113,7 +116,9 @@ def execute_statement_progress_case( """Execute one fake resource-scoped statement and capture projected output.""" stream: StringIO = StringIO() - projector: NativeProgressProjector = NativeProgressProjector(stream=stream, use_color=False) + projector: NativeProgressProjector = NativeProgressProjector( + stream=stream, use_color=False, debug=test_case.debug + ) adapter: FakeSlowAdapter = adapter_type(adapter=test_case.adapter, query_id=test_case.query_id) dispatcher: EventDispatcher = EventDispatcher() dispatcher.subscribe_lifecycle(subscriber=projector.consume, accepts_opaque=False) @@ -165,3 +170,56 @@ def run_monitor_capture_race(*, query_id: str) -> tuple[tuple[str, ...], int, bo provider.release.set() stopper.join(timeout=1.0) return tuple(submissions), provider.call_count, stopper.is_alive() + + +class TerminalStream(StringIO): + """String stream reporting a configurable terminal state.""" + + def __init__(self, *, tty: bool) -> None: + super().__init__() + self._tty: bool = tty + + def isatty(self) -> bool: + return self._tty + + +def complete_step_normally(step: OperationLifecycle) -> None: + """Let the step operation exit without an explicit terminal call.""" + + del step + + +def fail_step(step: OperationLifecycle) -> None: + """Mark the step operation failed.""" + + step.failed(error=RuntimeError("staging failed")) + + +def project_resource_step( + *, test_case: StepProgressCase, finish_step: Callable[[OperationLifecycle], None] +) -> tuple[str, ...]: + """Project one resource attempt containing one visible step operation.""" + + stream: TerminalStream = TerminalStream(tty=test_case.tty) + projector: NativeProgressProjector = NativeProgressProjector( + stream=stream, use_color=False, debug=test_case.debug + ) + dispatcher: EventDispatcher = EventDispatcher() + dispatcher.subscribe_lifecycle(subscriber=projector.consume, accepts_opaque=False) + with ( + invocation_scope("inv-step-progress"), + run_scope("run-step-progress"), + dispatcher_scope(dispatcher), + ResourceAttemptLifecycle( + resource_id="model:orders", + resource_kind="table", + resource_name="orders", + ), + OperationLifecycle( + operation_kind="warehouse", + operation_name="staging_creation", + ) as step, + ): + finish_step(step) + projector.close() + return tuple(stream.getvalue().splitlines()) diff --git a/tests/unit/src/sqlbuild/cli/progress/classes/test_native_progress_projector.py b/tests/unit/src/sqlbuild/cli/progress/classes/test_native_progress_projector.py index 56188ecd99..19187794ed 100644 --- a/tests/unit/src/sqlbuild/cli/progress/classes/test_native_progress_projector.py +++ b/tests/unit/src/sqlbuild/cli/progress/classes/test_native_progress_projector.py @@ -109,7 +109,9 @@ def test_given_retrying_child_operation_when_projected_then_each_line_flushes_be test_case: RetryProjectionCase, ) -> None: stream: _FlushRecordingStream = _FlushRecordingStream() - projector: NativeProgressProjector = NativeProgressProjector(stream=stream, use_color=False) + projector: NativeProgressProjector = NativeProgressProjector( + stream=stream, use_color=False, debug=True + ) dispatcher: EventDispatcher = EventDispatcher() dispatcher.subscribe_lifecycle(subscriber=projector.consume, accepts_opaque=False) @@ -492,7 +494,9 @@ def test_given_missing_terminal_on_tty_when_closed_then_only_cursor_is_restored( test_case: CursorCleanupCase, ) -> None: stream: _FlushRecordingStream = _FlushRecordingStream(tty=True) - projector: NativeProgressProjector = NativeProgressProjector(stream=stream, use_color=False) + projector: NativeProgressProjector = NativeProgressProjector( + stream=stream, use_color=False, debug=True + ) projector.consume( lifecycle_event( "resource_attempt_started", diff --git a/tests/unit/src/sqlbuild/cli/progress/classes/test_statement_progress.py b/tests/unit/src/sqlbuild/cli/progress/classes/test_statement_progress.py index 5d6d482754..9efb354f94 100644 --- a/tests/unit/src/sqlbuild/cli/progress/classes/test_statement_progress.py +++ b/tests/unit/src/sqlbuild/cli/progress/classes/test_statement_progress.py @@ -23,6 +23,7 @@ adapter="snowflake", query_id="01-query-orders", expected_context="model=orders phase=create kind=CREATE", + debug=True, ), ), ids=lambda case: case.description, @@ -73,6 +74,7 @@ def test_given_query_id_capture_race_when_monitor_stops_then_submission_is_publi adapter="snowflake", query_id="01-query-failed", expected_context="model=orders phase=create kind=CREATE", + debug=True, ), ), ids=lambda case: case.description, @@ -98,6 +100,7 @@ def test_given_slow_statement_when_it_fails_then_failure_has_query_id_and_monito adapter="duckdb", query_id=None, expected_context="model=orders phase=create kind=CREATE", + debug=True, ), ), ids=lambda case: case.description, @@ -114,3 +117,63 @@ def test_given_adapter_without_query_ids_when_statement_runs_then_id_noise_is_om assert "query_id=" not in output assert monitor_thread.is_alive() is False assert error is None + + +@pytest.mark.parametrize( + "test_case", + ( + StatementProgressCase( + description="snowflake completion without debug keeps only the slow-statement heartbeat", + adapter="snowflake", + query_id="01-query-quiet", + expected_context="model=orders phase=create kind=CREATE", + debug=False, + ), + StatementProgressCase( + description="duckdb completion without debug keeps only the slow-statement heartbeat", + adapter="duckdb", + query_id=None, + expected_context="model=orders phase=create kind=CREATE", + debug=False, + ), + ), + ids=lambda case: case.description, +) +def test_given_no_debug_when_slow_statement_completes_then_start_and_submission_rows_are_hidden( + test_case: StatementProgressCase, +) -> None: + output, monitor_thread, error = execute_statement_progress_case( + test_case=test_case, adapter_type=FakeSlowAdapter + ) + + assert f"statement {test_case.expected_context} START" not in output + assert f"statement {test_case.expected_context} query_id=" not in output + assert f"statement {test_case.expected_context} RUNNING" in output + assert monitor_thread.is_alive() is False + assert error is None + + +@pytest.mark.parametrize( + "test_case", + ( + StatementProgressCase( + description="snowflake failure without debug still reports the failed statement", + adapter="snowflake", + query_id="01-query-quiet-failed", + expected_context="model=orders phase=create kind=CREATE", + debug=False, + ), + ), + ids=lambda case: case.description, +) +def test_given_no_debug_when_slow_statement_fails_then_failure_row_keeps_query_id( + test_case: StatementProgressCase, +) -> None: + output, monitor_thread, error = execute_statement_progress_case( + test_case=test_case, adapter_type=FakeFailingSlowAdapter + ) + + assert f"statement {test_case.expected_context} START" not in output + assert f"statement {test_case.expected_context} FAIL query_id={test_case.query_id}" in output + assert monitor_thread.is_alive() is False + assert str(error) == "warehouse statement failed" diff --git a/tests/unit/src/sqlbuild/cli/progress/classes/test_step_progress.py b/tests/unit/src/sqlbuild/cli/progress/classes/test_step_progress.py new file mode 100644 index 0000000000..89ee0e86e7 --- /dev/null +++ b/tests/unit/src/sqlbuild/cli/progress/classes/test_step_progress.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import pytest + +from tests.unit.src.sqlbuild.cli.progress.classes._test_types import StepProgressCase +from tests.unit.src.sqlbuild.cli.progress.classes.helpers import ( + complete_step_normally, + fail_step, + project_resource_step, +) + + +@pytest.mark.parametrize( + "test_case", + ( + StepProgressCase( + description="redirected output hides passing step rows by default", + tty=False, + debug=False, + expected_line_prefixes=(" table orders START", " table orders OK "), + ), + StepProgressCase( + description="terminal output keeps only the resource result row by default", + tty=True, + debug=False, + expected_line_prefixes=(" table orders OK ",), + ), + StepProgressCase( + description="redirected output shows passing step rows with debug", + tty=False, + debug=True, + expected_line_prefixes=( + " table orders START", + " Create staging relation START", + " Create staging relation OK (", + " table orders OK ", + ), + ), + StepProgressCase( + description="terminal output shows start and step rows with debug", + tty=True, + debug=True, + expected_line_prefixes=( + " table orders START", + " Create staging relation START", + " Create staging relation OK (", + " table orders OK ", + ), + ), + ), + ids=lambda case: case.description, +) +def test_given_passing_resource_step_when_projected_then_step_rows_follow_debug_flag( + test_case: StepProgressCase, +) -> None: + lines: tuple[str, ...] = project_resource_step( + test_case=test_case, finish_step=complete_step_normally + ) + + assert len(lines) == len(test_case.expected_line_prefixes), lines + assert all( + line.startswith(prefix) + for line, prefix in zip(lines, test_case.expected_line_prefixes, strict=True) + ), lines + + +@pytest.mark.parametrize( + "test_case", + ( + StepProgressCase( + description="redirected output still reports a failed step by default", + tty=False, + debug=False, + expected_line_prefixes=( + " table orders START", + " Create staging relation FAIL (", + " table orders OK ", + ), + ), + StepProgressCase( + description="terminal output still reports a failed step by default", + tty=True, + debug=False, + expected_line_prefixes=( + " Create staging relation FAIL (", + " table orders OK ", + ), + ), + ), + ids=lambda case: case.description, +) +def test_given_failed_resource_step_when_projected_without_debug_then_failure_row_is_shown( + test_case: StepProgressCase, +) -> None: + lines: tuple[str, ...] = project_resource_step(test_case=test_case, finish_step=fail_step) + + assert len(lines) == len(test_case.expected_line_prefixes), lines + assert all( + line.startswith(prefix) + for line, prefix in zip(lines, test_case.expected_line_prefixes, strict=True) + ), lines