From 6efbdb4e08552796a9d24f7fbdd44ac0cf98896a Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Tue, 6 Oct 2026 07:53:53 +0100 Subject: [PATCH 1/2] fix: exit failed builds without a capture traceback The compute log tees called their failure callback with the error positionally, but ScopedComputeLogCapture bound _report with a keyword-only error, so any capture write failure raised TypeError. After an early planning or compile failure the transient status line was left running; its atexit close then wrote through the stale stdout tee to closed storage and printed the TypeError as an ignored exception. Report write failures through a matching one-argument method, detach the tees when capture ends so late writers reach only the terminal, and clear any live transient line before the dispatch error is reported. --- Makefile | 1 + .../_helpers/entry/dispatch_errors.py | 14 ++- .../classes/transient_line_coordinator.py | 10 ++ .../compute_logs/classes/binary_tee.py | 8 +- .../classes/scoped_compute_log_capture.py | 9 +- .../runtime/compute_logs/classes/text_tee.py | 9 +- .../cli/commands/main/build/_test_types.py | 9 ++ .../main/build/test_terminal_failure.py | 74 ++++++++++++ .../test_transient_line_coordinator.py | 29 +++++ .../compute_logs/classes/_test_types.py | 33 ++++++ .../classes/test_compute_log_tees.py | 84 ++++++++++++++ .../test_scoped_compute_log_capture.py | 106 +++++++++++++++++- 12 files changed, 380 insertions(+), 6 deletions(-) create mode 100644 tests/e2e/src/sqlbuild/cli/commands/main/build/test_terminal_failure.py diff --git a/Makefile b/Makefile index 6673fb6922..0cc2ba70e3 100644 --- a/Makefile +++ b/Makefile @@ -92,6 +92,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_selector_surface.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_table_function_dependency.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_template_expressions.py \ + tests/e2e/src/sqlbuild/cli/commands/main/build/test_terminal_failure.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_validation_failures.py E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \ diff --git a/src/sqlbuild/cli/commands/_helpers/entry/dispatch_errors.py b/src/sqlbuild/cli/commands/_helpers/entry/dispatch_errors.py index 5a8cab06bf..6e402f2cf4 100644 --- a/src/sqlbuild/cli/commands/_helpers/entry/dispatch_errors.py +++ b/src/sqlbuild/cli/commands/_helpers/entry/dispatch_errors.py @@ -23,6 +23,7 @@ SkillMaintenanceResult, ) from sqlbuild.lint.exceptions import LintError +from sqlbuild.presentation.main.transient_line_coordinator import shared_transient_line_coordinator from sqlbuild.rule_engine.exceptions import RulesError @@ -37,7 +38,7 @@ def dispatch_and_handle_errors( _report_skill_freshness(invocation=invocation) try: - return dispatch_with_observability(args=args, handlers=handlers) + return _dispatch_clearing_transient_line(args=args, handlers=handlers) except SystemExit as error: return error.code if isinstance(error.code, int) else 1 except (CliUserError, RulesError) as error: @@ -80,6 +81,17 @@ def dispatch_and_handle_errors( return 1 +def _dispatch_clearing_transient_line( + *, args: CliNamespace, handlers: CliEntrypointHandlers +) -> int: + """Stop a status line a failed command left running before its error is reported.""" + + try: + return dispatch_with_observability(args=args, handlers=handlers) + finally: + shared_transient_line_coordinator().clear_active() + + def _expected_command_failure_types() -> tuple[type[Exception], ...]: """Load command-specific failures only after a command raises.""" diff --git a/src/sqlbuild/presentation/classes/transient_line_coordinator.py b/src/sqlbuild/presentation/classes/transient_line_coordinator.py index f3eddb2db6..9975f4cf9a 100644 --- a/src/sqlbuild/presentation/classes/transient_line_coordinator.py +++ b/src/sqlbuild/presentation/classes/transient_line_coordinator.py @@ -28,6 +28,16 @@ def release(self, *, owner: TransientLineOwner) -> None: if self._active is not None and self._active[1] is owner: self._active = None + def clear_active(self) -> None: + """Erase and forget any live transient line, so nothing redraws it afterwards.""" + + with self.lock: + if self._active is None: + return + owner: TransientLineOwner = self._active[1] + self._active = None + owner.clear_transient_line() + def write_persistent(self, *, stream: TextIO, text: str) -> None: """Write lasting text, clearing and then redrawing any live transient line.""" diff --git a/src/sqlbuild/runtime/compute_logs/classes/binary_tee.py b/src/sqlbuild/runtime/compute_logs/classes/binary_tee.py index bb6e779a3c..c36aa801c1 100644 --- a/src/sqlbuild/runtime/compute_logs/classes/binary_tee.py +++ b/src/sqlbuild/runtime/compute_logs/classes/binary_tee.py @@ -35,6 +35,7 @@ def __init__( self._before_write: Callable[[], None] | None = before_write self._failure_callback: Callable[[Exception], None] | None = failure_callback self._storage_failed: bool = False + self._detached: bool = False def write(self, data: bytes | bytearray) -> int: """Preserve sink behavior while isolating capture append failures.""" @@ -53,7 +54,7 @@ def write(self, data: bytes | bytearray) -> int: : accepted_write_count(result=result, offered_count=len(text)) ] accepted = accepted_text.encode(self._encoding, self._errors) - if accepted and not self._storage_failed: + if accepted and not self._storage_failed and not self._detached: try: self._storage.append( invocation_id=self._invocation_id, stream=self._stream, data=accepted @@ -64,6 +65,11 @@ def write(self, data: bytes | bytearray) -> int: self._failure_callback(error) return result if isinstance(result, int) else len(offered) + def detach(self) -> None: + """Stop retaining writes once capture ends; late writers reach only the sink.""" + + self._detached = True + def flush(self) -> None: """Flush the original sink without closing it.""" diff --git a/src/sqlbuild/runtime/compute_logs/classes/scoped_compute_log_capture.py b/src/sqlbuild/runtime/compute_logs/classes/scoped_compute_log_capture.py index 9d44a7a53b..5025e62098 100644 --- a/src/sqlbuild/runtime/compute_logs/classes/scoped_compute_log_capture.py +++ b/src/sqlbuild/runtime/compute_logs/classes/scoped_compute_log_capture.py @@ -5,7 +5,6 @@ import sys from collections.abc import Callable from datetime import UTC, datetime -from functools import partial from typing import Any from sqlbuild.diagnostics.classes.invocation_diagnostic_routing import InvocationDiagnosticRouting @@ -73,7 +72,7 @@ def run(self, *, operation: Callable[[], int]) -> int: return operation_result def _install(self) -> None: - write_failure: Callable[[Exception], None] = partial(self._report, channel="capture_write") + write_failure: Callable[[Exception], None] = self._report_write_failure self._stdout_tee = TextComputeLogTee( sink=self._original_stdout, storage=self._storage, @@ -110,6 +109,9 @@ def _cleanup(self) -> bool: self._report(error=error, channel="capture_flush") sys.stdout = self._original_stdout sys.stderr = self._original_stderr + for tee in (self._stdout_tee, self._stderr_tee): + if tee is not None: + tee.detach() if self._routing is not None: try: self._routing.__exit__(None, None, None) @@ -151,6 +153,9 @@ def _close_storage(self) -> None: except Exception as error: self._report(error=error, channel="capture_close") + def _report_write_failure(self, error: Exception) -> None: + self._report(error=error, channel="capture_write") + def _report(self, *, error: Exception, channel: str) -> None: try: self._failure_callback(error, channel) diff --git a/src/sqlbuild/runtime/compute_logs/classes/text_tee.py b/src/sqlbuild/runtime/compute_logs/classes/text_tee.py index 92f15f9824..84f76dc406 100644 --- a/src/sqlbuild/runtime/compute_logs/classes/text_tee.py +++ b/src/sqlbuild/runtime/compute_logs/classes/text_tee.py @@ -30,6 +30,7 @@ def __init__( self._errors: str = getattr(sink, "errors", None) or "strict" self._failure_callback: Callable[[Exception], None] | None = failure_callback self._storage_failed: bool = False + self._detached: bool = False binary_sink: Any = getattr(sink, "buffer", sink) binary_uses_text_sink: bool = binary_sink is sink self._buffer: BinaryComputeLogTee = BinaryComputeLogTee( @@ -49,7 +50,7 @@ def write(self, text: str) -> int: result: object = self._sink.write(text) accepted_count: int = accepted_write_count(result=result, offered_count=len(text)) - if accepted_count and not self._storage_failed: + if accepted_count and not self._storage_failed and not self._detached: try: emitted: bytes = text[:accepted_count].encode(self._encoding, self._errors) self._storage.append( @@ -61,6 +62,12 @@ def write(self, text: str) -> int: self._failure_callback(error) return result if isinstance(result, int) else len(text) + def detach(self) -> None: + """Stop retaining writes once capture ends; late writers reach only the sink.""" + + self._detached = True + self._buffer.detach() + def flush(self) -> None: """Flush the existing text sink without changing its ownership.""" diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py index f47dc5c487..40cb42e596 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py @@ -1206,3 +1206,12 @@ class QueryFingerprintEditE2ETestCase: format_flags: tuple[str, ...] expected_format_summary: str expected_reasons: dict[str, str] + + +@dataclass(frozen=True) +class TerminalBuildFailureE2ETestCase: + description: str + command: tuple[str, ...] + model_replacements: tuple[tuple[str, str, str], ...] + expected_screen_fragments: tuple[str, ...] + columns: int = 120 diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_terminal_failure.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_terminal_failure.py new file mode 100644 index 0000000000..759bbe8b4e --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_terminal_failure.py @@ -0,0 +1,74 @@ +"""E2E coverage for early build failures in a real terminal.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.build._test_types import ( + TerminalBuildFailureE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import ( + SPINNER_GLYPHS, + prepare_waffle_shop, + render_terminal_screen, + run_sqb_with_pty, +) + + +@pytest.mark.parametrize( + "test_case", + [ + TerminalBuildFailureE2ETestCase( + description="selected model with unbuilt upstreams fails during planning", + command=("build", "--select", "daily_revenue"), + model_replacements=(), + expected_screen_fragments=( + "Inspected warehouse state.", + "error[S301]: cannot build selected scope: 2 missing upstream dependencies", + ), + ), + TerminalBuildFailureE2ETestCase( + description="compile error fails before connecting", + command=("build",), + model_replacements=( + ( + "models/marts/daily_revenue.sql", + '__ref("stg_payments")', + '__ref("missing_payments")', + ), + ), + expected_screen_fragments=("error[", "missing_payments"), + ), + ], + ids=lambda case: case.description, +) +def test_given_early_build_failure_when_running_in_terminal_then_exits_with_error_only( + tmp_path: Path, test_case: TerminalBuildFailureE2ETestCase +) -> None: + project_dir: Path = prepare_waffle_shop(tmp_path) + for relative_path, old, new in test_case.model_replacements: + model_path: Path = project_dir / relative_path + model_path.write_text( + model_path.read_text(encoding="utf-8").replace(old, new), encoding="utf-8" + ) + + result: subprocess.CompletedProcess[str] = run_sqb_with_pty( + command=test_case.command, project_dir=project_dir, columns=test_case.columns + ) + screen: tuple[str, ...] = render_terminal_screen( + output=result.stdout, columns=test_case.columns + ) + rendered: str = "\n".join(screen) + + assert result.returncode == 1, rendered + assert "Traceback" not in result.stdout, rendered + assert "Exception ignored" not in result.stdout, rendered + assert not any(glyph in rendered for glyph in SPINNER_GLYPHS), 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/presentation/classes/test_transient_line_coordinator.py b/tests/unit/src/sqlbuild/presentation/classes/test_transient_line_coordinator.py index 30522b4ea5..8cf7e30bc6 100644 --- a/tests/unit/src/sqlbuild/presentation/classes/test_transient_line_coordinator.py +++ b/tests/unit/src/sqlbuild/presentation/classes/test_transient_line_coordinator.py @@ -72,5 +72,34 @@ def test_given_released_spinner_when_writing_persistent_line_then_writes_directl assert tuple(events) == test_case.expected_events +@pytest.mark.parametrize( + "test_case", + ( + TransientLineCoordinatorTestCase( + description="cleared spinner is erased once and never redrawn", + persistent_stream_is_tty=True, + expected_events=("clear", "write:error[S301]: failed\n"), + ), + ), + ids=lambda case: case.description, +) +def test_given_live_spinner_when_clearing_active_line_then_later_writes_do_not_redraw_it( + test_case: TransientLineCoordinatorTestCase, +) -> None: + events: list[str] = [] + coordinator: TransientLineCoordinator = TransientLineCoordinator() + owner: RecordingSpinnerOwner = RecordingSpinnerOwner(events=events) + coordinator.claim(stream=RecordingStream(events=[], tty=True), owner=owner) + + coordinator.clear_active() + coordinator.clear_active() + coordinator.write_persistent( + stream=RecordingStream(events=events, tty=test_case.persistent_stream_is_tty), + text="error[S301]: failed\n", + ) + + assert tuple(events) == test_case.expected_events + + if __name__ == "__main__": pytest.main([__file__, "-vv"]) diff --git a/tests/unit/src/sqlbuild/runtime/compute_logs/classes/_test_types.py b/tests/unit/src/sqlbuild/runtime/compute_logs/classes/_test_types.py index c556ced818..f34c0d42b2 100644 --- a/tests/unit/src/sqlbuild/runtime/compute_logs/classes/_test_types.py +++ b/tests/unit/src/sqlbuild/runtime/compute_logs/classes/_test_types.py @@ -85,3 +85,36 @@ class PathSafetyTestCase: description: str alias_kind: str expected_error_fragment: str + + +@dataclass(frozen=True) +class CaptureWriteFailureTestCase: + description: str + stream_name: str + expected_exit_code: int + expected_channels: tuple[str, ...] + + +@dataclass(frozen=True) +class LateStreamWriteTestCase: + description: str + captured_text: str + late_text: str + expected_sink_text: str + expected_captured_byte_count: int + + +@dataclass(frozen=True) +class TeeStorageFailureTestCase: + description: str + writes: tuple[str, ...] + expected_sink_writes: tuple[str, ...] + expected_failure_count: int + + +@dataclass(frozen=True) +class DetachedTeeTestCase: + description: str + text: str + binary: bytes + expected_append_count: int diff --git a/tests/unit/src/sqlbuild/runtime/compute_logs/classes/test_compute_log_tees.py b/tests/unit/src/sqlbuild/runtime/compute_logs/classes/test_compute_log_tees.py index 1d1e480719..c5019a2871 100644 --- a/tests/unit/src/sqlbuild/runtime/compute_logs/classes/test_compute_log_tees.py +++ b/tests/unit/src/sqlbuild/runtime/compute_logs/classes/test_compute_log_tees.py @@ -6,7 +6,9 @@ from sqlbuild.runtime.compute_logs.classes.binary_tee import BinaryComputeLogTee from sqlbuild.runtime.compute_logs.classes.text_tee import TextComputeLogTee from tests.unit.src.sqlbuild.runtime.compute_logs.classes._test_types import ( + DetachedTeeTestCase, PartialWriteTestCase, + TeeStorageFailureTestCase, ) @@ -162,3 +164,85 @@ def test_given_text_sink_exception_when_writing_then_exception_propagates_withou assert test_case.expected_result == 0 assert test_case.expected_bytes == b"" storage.append.assert_not_called() + + +@pytest.mark.parametrize( + "test_case", + ( + TeeStorageFailureTestCase( + description="storage failure reports the error once and keeps writing to the sink", + writes=("first\n", "second\n"), + expected_sink_writes=("first\n", "second\n"), + expected_failure_count=1, + ), + ), + ids=lambda case: case.description, +) +def test_given_storage_append_failure_when_writing_text_then_callback_receives_error_once( + test_case: TeeStorageFailureTestCase, +) -> None: + sink: Mock = Mock() + sink.encoding = "utf-8" + sink.errors = "strict" + sink.buffer = Mock() + sink.write.return_value = None + storage_error: OSError = OSError("controlled append failure") + storage: Mock = Mock() + storage.append.side_effect = storage_error + failures: list[Exception] = [] + tee: TextComputeLogTee = TextComputeLogTee( + sink=sink, + storage=storage, + invocation_id="text_append_failure", + stream=ComputeLogStream.STDERR, + failure_callback=failures.append, + ) + + for text in test_case.writes: + _ = tee.write(text) + + assert tuple(call.args[0] for call in sink.write.call_args_list) == ( + test_case.expected_sink_writes + ) + assert failures == [storage_error] * test_case.expected_failure_count + + +@pytest.mark.parametrize( + "test_case", + ( + DetachedTeeTestCase( + description="detached text and buffer writes reach the sink without retention", + text="late text\n", + binary=b"late bytes\n", + expected_append_count=0, + ), + ), + ids=lambda case: case.description, +) +def test_given_detached_tee_when_writing_then_sink_receives_data_without_storage_append( + test_case: DetachedTeeTestCase, +) -> None: + sink: Mock = Mock() + sink.encoding = "utf-8" + sink.errors = "strict" + sink.buffer = Mock() + sink.write.return_value = None + sink.buffer.write.return_value = None + storage: Mock = Mock() + failures: list[Exception] = [] + tee: TextComputeLogTee = TextComputeLogTee( + sink=sink, + storage=storage, + invocation_id="detached", + stream=ComputeLogStream.STDOUT, + failure_callback=failures.append, + ) + + tee.detach() + _ = tee.write(test_case.text) + _ = tee.buffer.write(test_case.binary) + + sink.write.assert_called_once_with(test_case.text) + sink.buffer.write.assert_called_once_with(test_case.binary) + assert storage.append.call_count == test_case.expected_append_count + assert failures == [] diff --git a/tests/unit/src/sqlbuild/runtime/compute_logs/classes/test_scoped_compute_log_capture.py b/tests/unit/src/sqlbuild/runtime/compute_logs/classes/test_scoped_compute_log_capture.py index 28c65c0862..086d9956e6 100644 --- a/tests/unit/src/sqlbuild/runtime/compute_logs/classes/test_scoped_compute_log_capture.py +++ b/tests/unit/src/sqlbuild/runtime/compute_logs/classes/test_scoped_compute_log_capture.py @@ -1,10 +1,11 @@ +import io import json import logging import sys from collections.abc import Callable from datetime import UTC, datetime from pathlib import Path -from typing import cast +from typing import TextIO, cast from unittest.mock import Mock, patch import pytest @@ -32,7 +33,9 @@ from sqlbuild.runtime.compute_logs.classes.text_tee import TextComputeLogTee from tests.unit.src.sqlbuild.runtime.compute_logs.classes._test_types import ( CaptureOutcomeTestCase, + CaptureWriteFailureTestCase, DiagnosticRoutingTestCase, + LateStreamWriteTestCase, ) from tests.unit.src.sqlbuild.runtime.compute_logs.classes.helpers import ( AccessFailureTextSink, @@ -530,3 +533,104 @@ def emit_records() -> int: assert diagnostic_text.count("plain internal api_key=[REDACTED]") == 1 assert diagnostic_text.count("structured internal client_secret=[REDACTED]") == 1 assert diagnostic_text.count("user debug") == 0 + + +@pytest.mark.parametrize( + "test_case", + ( + CaptureWriteFailureTestCase( + description="stdout append failure reports one capture write failure", + stream_name="stdout", + expected_exit_code=0, + expected_channels=("capture_write",), + ), + CaptureWriteFailureTestCase( + description="stderr append failure reports one capture write failure", + stream_name="stderr", + expected_exit_code=3, + expected_channels=("capture_write",), + ), + ), + ids=lambda case: case.description, +) +def test_given_storage_append_failure_when_operation_writes_then_failure_reaches_callback( + tmp_path: Path, test_case: CaptureWriteFailureTestCase +) -> None: + storage: LocalFilesystemComputeLogStorage = LocalFilesystemComputeLogStorage( + project_dir=tmp_path + ) + metadata: CaptureMetadata = build_capture_metadata( + project_dir=tmp_path, invocation_id="append_failure", started_at=datetime.now(UTC) + ) + storage.start_capture(metadata) + reported: list[tuple[type[Exception], str]] = [] + capture: ScopedComputeLogCapture = ScopedComputeLogCapture( + storage=storage, + metadata=metadata, + failure_callback=lambda error, channel: reported.append((type(error), channel)), + ) + + def write_twice() -> int: + stream: TextIO = getattr(sys, test_case.stream_name) + _ = stream.write("first line\n") + _ = stream.write("second line\n") + return test_case.expected_exit_code + + with patch.object(storage, "append", Mock(side_effect=OSError("disk full"))): + result: int = capture.run(operation=write_twice) + + assert result == test_case.expected_exit_code + assert tuple(channel for _error_type, channel in reported) == test_case.expected_channels + assert all(error_type is OSError for error_type, _channel in reported) + + +@pytest.mark.parametrize( + "test_case", + ( + LateStreamWriteTestCase( + description="stale stdout reference written after capture reaches only the terminal", + captured_text="during capture\n", + late_text="after capture\n", + expected_sink_text="during capture\nafter capture\n", + expected_captured_byte_count=len(b"during capture\n"), + ), + ), + ids=lambda case: case.description, +) +def test_given_stale_stream_reference_when_writing_after_capture_then_sink_receives_text_only( + tmp_path: Path, test_case: LateStreamWriteTestCase, monkeypatch: pytest.MonkeyPatch +) -> None: + sink: io.StringIO = io.StringIO() + monkeypatch.setattr(sys, "stdout", sink) + storage: LocalFilesystemComputeLogStorage = LocalFilesystemComputeLogStorage( + project_dir=tmp_path + ) + metadata: CaptureMetadata = build_capture_metadata( + project_dir=tmp_path, invocation_id="late_write", started_at=datetime.now(UTC) + ) + storage.start_capture(metadata) + reported: list[str] = [] + capture: ScopedComputeLogCapture = ScopedComputeLogCapture( + storage=storage, + metadata=metadata, + failure_callback=lambda _error, channel: reported.append(channel), + ) + stale_streams: list[TextIO] = [] + + def keep_stream_reference() -> int: + stale_streams.append(sys.stdout) + _ = sys.stdout.write(test_case.captured_text) + return 0 + + result: int = capture.run(operation=keep_stream_reference) + _ = stale_streams[0].write(test_case.late_text) + stale_streams[0].flush() + + assert result == 0 + assert reported == [] + assert sink.getvalue() == test_case.expected_sink_text + assert ( + storage.get_byte_count(invocation_id=metadata.invocation_id, stream=ComputeLogStream.STDOUT) + == test_case.expected_captured_byte_count + ) + assert storage.is_complete(invocation_id=metadata.invocation_id) is True From 5b5b97f469365f4dbb47f197f059261e383ec526 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Tue, 6 Oct 2026 07:53:53 +0100 Subject: [PATCH 2/2] fix: align test and expect status columns in build and test output Expectation rows under a failing test are indented further than their parent row but padded to the same name width, so their PASS/FAIL sat four columns to the right. Pad tree child names by their extra indentation in sqb build and sqb test, include that indentation when sizing the name column so long expectation names still align, and align the nested error label with the expectation names. --- .../commands/_helpers/test/sql_progress.py | 14 +- .../classes/build_progress_callbacks.py | 21 +- .../nested_command_progress_callbacks.py | 4 +- src/sqlbuild/cli/progress/constants.py | 3 + .../commands/shared/_helpers/_test_types.py | 9 + .../cli/commands/shared/_helpers/helpers.py | 66 +++++- .../commands/shared/_helpers/test_progress.py | 193 +++++++++++++++++- 7 files changed, 298 insertions(+), 12 deletions(-) create mode 100644 src/sqlbuild/cli/progress/constants.py diff --git a/src/sqlbuild/cli/commands/_helpers/test/sql_progress.py b/src/sqlbuild/cli/commands/_helpers/test/sql_progress.py index a3f4a213aa..ca177d6577 100644 --- a/src/sqlbuild/cli/commands/_helpers/test/sql_progress.py +++ b/src/sqlbuild/cli/commands/_helpers/test/sql_progress.py @@ -5,6 +5,7 @@ import json from pathlib import Path +from sqlbuild.cli.progress.constants import NESTED_CHILD_NAME_INDENT from sqlbuild.cli.progress.main._expectation_detail import format_expectation_detail from sqlbuild.cli.progress.main._expectation_name import format_expectation_name from sqlbuild.cli.progress.models import NestedProgressChildRow @@ -21,6 +22,7 @@ def resolve_test_name_width(test_entries: tuple[SqlTestPlanEntry, ...]) -> int: """Resolve one shared width for test rows and nested expectation rows.""" names: list[str] = [] + child_names: list[str] = [] entry: SqlTestPlanEntry for entry in test_entries: names.append( @@ -31,9 +33,15 @@ def resolve_test_name_width(test_entries: tuple[SqlTestPlanEntry, ...]) -> int: parameter_values=entry.parameter_values, ) ) - names.extend(f"expected {step.model_name}" for step in entry.chain if step.expected_cte_sql) - names.extend(f"assertion {assertion.name}" for assertion in entry.assertions) - return resolve_name_column_width(names=names, min_width=50) + child_names.extend( + f"expected {step.model_name}" for step in entry.chain if step.expected_cte_sql + ) + child_names.extend(f"assertion {assertion.name}" for assertion in entry.assertions) + child_name_length: int = max((len(name) for name in child_names), default=0) + return max( + resolve_name_column_width(names=names, min_width=50), + child_name_length + NESTED_CHILD_NAME_INDENT, + ) def build_test_expectation_rows( diff --git a/src/sqlbuild/cli/commands/classes/build_progress_callbacks.py b/src/sqlbuild/cli/commands/classes/build_progress_callbacks.py index 0fa259a237..6d0f3d9417 100644 --- a/src/sqlbuild/cli/commands/classes/build_progress_callbacks.py +++ b/src/sqlbuild/cli/commands/classes/build_progress_callbacks.py @@ -67,6 +67,7 @@ _MIN_NAME_WIDTH: int = 20 _NAME_PADDING: int = 2 _SUB_INDENT: int = 2 +_TREE_CHILD_NAME_INDENT: int = _SUB_INDENT + 4 _HOOK_PHASE_WIDTH: int = 10 _HOOK_TYPE_WIDTH: int = 8 _HOOK_MIN_LABEL_WIDTH: int = 24 @@ -242,13 +243,17 @@ def __init__( if getattr(chain_step, "expected_cte_sql", None): max_name_len = max( max_name_len, - len(format_expectation_name(str(getattr(chain_step, "model_name", "")))), + _tree_child_name_length( + format_expectation_name(str(getattr(chain_step, "model_name", ""))) + ), ) assertion: object for assertion in getattr(test_entry, "assertions", ()): max_name_len = max( max_name_len, - len(format_expectation_name(f"assertion {getattr(assertion, 'name', '')}")), + _tree_child_name_length( + format_expectation_name(f"assertion {getattr(assertion, 'name', '')}") + ), ) self._name_width: int = max(max_name_len + _NAME_PADDING, _MIN_NAME_WIDTH) @@ -704,7 +709,7 @@ def _write_failed_test_result(self, *, test_result: SqlTestExecutionResult) -> N ) expectation_pad: str = " " * (self._prefix_width + _SUB_INDENT + 2) expectation_type_width: int = _TYPE_WIDTH - 2 - sub_nw: int = self._name_width - _SUB_INDENT + tree_child_nw: int = self._name_width - _TREE_CHILD_NAME_INDENT step_result: StepResult for index, step_result in enumerate(test_result.step_results): expectation_status: str = self._style.status( @@ -719,7 +724,7 @@ def _write_failed_test_result(self, *, test_result: SqlTestExecutionResult) -> N self._stream.write( f"{expectation_pad}{connector} " f"{self._style.muted(f'{"expect":<{expectation_type_width}}')}" - f"{expectation_name:<{sub_nw}} {expectation_status}{expectation_detail}\n" + f"{expectation_name:<{tree_child_nw}} {expectation_status}{expectation_detail}\n" ) if test_result.error_message: self._write_error_detail( @@ -784,7 +789,7 @@ def _write_error_detail( pad: str = " " * (self._prefix_width + child_offset) connector: str = tree_connector(style=self._style, last=True) if tree_child else "" connector_suffix: str = f"{connector} " if tree_child else "" - label_width: int = _TYPE_WIDTH - 4 if tree_child else _TYPE_WIDTH - 1 + label_width: int = _TYPE_WIDTH - 3 if tree_child else _TYPE_WIDTH - 1 label_padding: str = " " * max(0, label_width - len("error")) label: str = f"{self._style.error_muted('error')}{label_padding}" plain_prefix_width: int = len(pad) + (4 if tree_child else 0) + label_width + 1 @@ -1495,6 +1500,12 @@ def _format_display_sql(sql: str) -> str: return f"{stripped};" +def _tree_child_name_length(name: str) -> int: + """Measure a tree child name so its status still fits the shared status column.""" + + return len(name) + _TREE_CHILD_NAME_INDENT - _NAME_PADDING + + def _truncate_name(*, name: str, width: int) -> str: """Truncate a display name to fit within the given width.""" diff --git a/src/sqlbuild/cli/progress/classes/nested_command_progress_callbacks.py b/src/sqlbuild/cli/progress/classes/nested_command_progress_callbacks.py index bda2f92999..e39285e72b 100644 --- a/src/sqlbuild/cli/progress/classes/nested_command_progress_callbacks.py +++ b/src/sqlbuild/cli/progress/classes/nested_command_progress_callbacks.py @@ -9,6 +9,7 @@ NativeProgressProjector, current_native_progress_projector, ) +from sqlbuild.cli.progress.constants import NESTED_CHILD_NAME_INDENT from sqlbuild.cli.progress.models import NestedProgressChildRow from sqlbuild.presentation.classes.cli_style import CliStyle from sqlbuild.presentation.classes.transient_line_coordinator import TransientLineCoordinator @@ -155,6 +156,7 @@ def _write_completed_item( self._stream.write( f" {self._label:<{_LABEL_WIDTH}}{item_name:<{self._name_width}} {status}{detail}\n" ) + child_name_width: int = self._name_width - NESTED_CHILD_NAME_INDENT child_row: NestedProgressChildRow for index, child_row in enumerate(child_rows): child_status: str = self._style.status(status=child_row.status_text) @@ -162,7 +164,7 @@ def _write_completed_item( connector: str = tree_connector(style=self._style, last=last) self._stream.write( f" {connector} {child_row.label:<{_LABEL_WIDTH - 2}}" - f"{child_row.name:<{self._name_width}} " + f"{child_row.name:<{child_name_width}} " f"{child_status}{child_row.detail}\n" ) if error_message is not None: diff --git a/src/sqlbuild/cli/progress/constants.py b/src/sqlbuild/cli/progress/constants.py new file mode 100644 index 0000000000..fb554beaa8 --- /dev/null +++ b/src/sqlbuild/cli/progress/constants.py @@ -0,0 +1,3 @@ +"""Shared layout constants for CLI progress rendering.""" + +NESTED_CHILD_NAME_INDENT: int = 4 diff --git a/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/_test_types.py b/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/_test_types.py index c43bbbb75e..a5f1fd369f 100644 --- a/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/_test_types.py @@ -176,3 +176,12 @@ class NestedProgressConcurrentTestCase: description: str concurrency: int expected_output: str + + +@dataclass(frozen=True) +class SqlTestStatusAlignmentTestCase: + description: str + test_name: str + expected_models: tuple[str, ...] + assertion_names: tuple[str, ...] + expected_row_count: int diff --git a/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/helpers.py b/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/helpers.py index 505794157f..fd42c3c6c9 100644 --- a/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/helpers.py +++ b/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/helpers.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re import threading from collections.abc import Callable from pathlib import Path @@ -14,9 +15,21 @@ ) from sqlbuild.compiler.compile.models import CompiledObjectKey, CompiledRelationLocation from sqlbuild.compiler.compile.types import CompiledResourceType -from sqlbuild.compiler.planner.models import ModelPlanEntry, PlanOutput +from sqlbuild.compiler.planner.models import ( + ChainStep, + ModelPlanEntry, + PlanOutput, + SqlTestAssertionStep, + SqlTestPlanEntry, +) from sqlbuild.compiler.planner.types import MaterializationType, PlanAction, PlanReason from sqlbuild.executor.auditing.models import AuditExecutionResult +from sqlbuild.executor.testing.models import SqlTestExecutionResult, StepResult +from sqlbuild.executor.testing.types import SqlTestOutcome + +_STATUS_TOKEN_PATTERN: re.Pattern[str] = re.compile( + r"^.*? (PASS|FAIL|ERROR|OK)(?= |$)", flags=re.MULTILINE +) def write_spinner_line_and_release( @@ -85,3 +98,54 @@ def build_progress_snapshot_plan_output( model_entries=(entry,), selected_keys=frozenset((entry.key,)), ) + + +def build_sql_test_plan_entry( + *, test_name: str, expected_models: tuple[str, ...], assertion_names: tuple[str, ...] +) -> SqlTestPlanEntry: + return SqlTestPlanEntry( + key=CompiledObjectKey(resource_type=CompiledResourceType.SQL_TEST, name=test_name), + name=test_name, + chain=tuple( + ChainStep( + model_name=model_name, + resolved_sql="SELECT 1 AS order_id", + expected_cte_sql="SELECT 1 AS order_id", + ) + for model_name in expected_models + ), + assertions=tuple( + SqlTestAssertionStep(name=assertion_name, resolved_sql="SELECT 1 AS order_id") + for assertion_name in assertion_names + ), + ) + + +def build_sql_test_result( + *, + test_name: str, + expected_models: tuple[str, ...], + assertion_names: tuple[str, ...], + outcome: SqlTestOutcome, +) -> SqlTestExecutionResult: + first_step, *later_steps = ( + *expected_models, + *(f"assertion {assertion_name}" for assertion_name in assertion_names), + ) + return SqlTestExecutionResult( + test_name=test_name, + outcome=outcome, + step_results=( + StepResult(model_name=first_step, outcome=outcome), + *( + StepResult(model_name=step_name, outcome=SqlTestOutcome.PASS) + for step_name in later_steps + ), + ), + ) + + +def status_columns(output: str) -> tuple[int, ...]: + """Return the column of the first status token on every row that has one.""" + + return tuple(match.start(1) - match.start() for match in _STATUS_TOKEN_PATTERN.finditer(output)) diff --git a/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/test_progress.py b/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/test_progress.py index d76c121ae8..4ad8ceea79 100644 --- a/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/test_progress.py +++ b/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/test_progress.py @@ -11,6 +11,10 @@ from sqlbuild.adapter.contract.models import LifeCycleEvent from sqlbuild.adapter.contract.types import LifeCycleEventKind +from sqlbuild.cli.commands._helpers.test.sql_progress import ( + build_test_expectation_rows, + resolve_test_name_width, +) from sqlbuild.cli.commands.classes import build_progress_callbacks from sqlbuild.cli.commands.classes.build_progress_callbacks import ( BuildProgressCallbacks, @@ -55,11 +59,15 @@ NestedProgressChildRowsTestCase, NestedProgressConcurrentTestCase, RuntimeDiagnosticsTestCase, + SqlTestStatusAlignmentTestCase, TruncateNameTestCase, ) from tests.unit.src.sqlbuild.cli.commands.shared._helpers.helpers import ( build_audit_result, build_progress_snapshot_plan_output, + build_sql_test_plan_entry, + build_sql_test_result, + status_columns, write_spinner_line_and_release, ) @@ -242,7 +250,7 @@ def test_given_name_and_width_when_truncating_then_returns_expected_result( "test test_fact_orders", "expect expected fact_orders", "expect assertion line_totals_are_non_negative", - "expected fact_orders PASS", + "expected fact_orders PASS", ), unexpected_fragments=("...",), ), @@ -910,7 +918,7 @@ def test_given_execution_context_when_writing_header_then_renders_expected_outpu "test test_daily_revenu...", "ERROR", "├── expect expected stg_orders", - "└── error [T002] execution error while running 'stg_orders'", + "└── error [T002] execution error while running 'stg_orders'", ), unexpected_fragments=("0 mismatched", "\033["), ), @@ -1317,3 +1325,184 @@ def test_given_active_top_level_node_when_waiting_then_spinner_advances_frames( spinner_frame: str for spinner_frame in test_case.expected_spinner_frames: assert spinner_frame in output + + +@pytest.mark.parametrize( + "test_case", + ( + SqlTestStatusAlignmentTestCase( + description="short names share one status column", + test_name="test_daily_revenue_chain", + expected_models=("stg_orders", "daily_revenue"), + assertion_names=("revenue_is_non_negative",), + expected_row_count=4, + ), + SqlTestStatusAlignmentTestCase( + description="longest name is a child assertion", + test_name="test_daily_revenue_chain", + expected_models=("stg_orders",), + assertion_names=("daily_revenue_is_non_negative_for_every_completed_order_day",), + expected_row_count=3, + ), + SqlTestStatusAlignmentTestCase( + description="longest name is the parent test", + test_name="test_daily_revenue_chain_with_refunds_and_partial_payment_retries", + expected_models=("stg_orders", "daily_revenue"), + assertion_names=(), + expected_row_count=3, + ), + ), + ids=lambda case: case.description, +) +def test_given_failed_sql_test_when_reporting_build_progress_then_child_statuses_align_with_parent( + test_case: SqlTestStatusAlignmentTestCase, monkeypatch: pytest.MonkeyPatch +) -> None: + stream: StringIO = StringIO() + monkeypatch.setattr("sys.stdout", stream) + callbacks: BuildProgressCallbacks = BuildProgressCallbacks( + plan=PlanOutput( + test_entries=( + build_sql_test_plan_entry( + test_name=test_case.test_name, + expected_models=test_case.expected_models, + assertion_names=test_case.assertion_names, + ), + ) + ), + use_color=False, + ) + + callbacks.on_node_complete( + build_sql_test_result( + test_name=test_case.test_name, + expected_models=test_case.expected_models, + assertion_names=test_case.assertion_names, + outcome=SqlTestOutcome.FAIL, + ) + ) + columns: tuple[int, ...] = status_columns(stream.getvalue()) + + assert len(columns) == test_case.expected_row_count, stream.getvalue() + assert len(set(columns)) == 1, stream.getvalue() + + +@pytest.mark.parametrize( + "test_case", + ( + SqlTestStatusAlignmentTestCase( + description="longest name is a child assertion", + test_name="test_daily_revenue_chain", + expected_models=("stg_orders",), + assertion_names=("daily_revenue_is_non_negative_for_every_completed_order_day",), + expected_row_count=4, + ), + ), + ids=lambda case: case.description, +) +def test_given_passing_sql_test_when_reporting_host_model_then_test_rows_align_with_model_status( + test_case: SqlTestStatusAlignmentTestCase, monkeypatch: pytest.MonkeyPatch +) -> None: + stream: StringIO = StringIO() + monkeypatch.setattr("sys.stdout", stream) + callbacks: BuildProgressCallbacks = BuildProgressCallbacks( + plan=PlanOutput( + test_entries=( + build_sql_test_plan_entry( + test_name=test_case.test_name, + expected_models=test_case.expected_models, + assertion_names=test_case.assertion_names, + ), + ) + ), + use_color=False, + ) + + callbacks.on_node_complete( + build_sql_test_result( + test_name=test_case.test_name, + expected_models=test_case.expected_models, + assertion_names=test_case.assertion_names, + outcome=SqlTestOutcome.PASS, + ) + ) + callbacks.on_node_complete( + ModelExecutionResult( + model_name=test_case.expected_models[0], + status=ExecutionStatus.SUCCESS, + duration_ms=100, + ) + ) + columns: tuple[int, ...] = status_columns(stream.getvalue()) + + assert len(columns) == test_case.expected_row_count, stream.getvalue() + assert len(set(columns)) == 1, stream.getvalue() + + +@pytest.mark.parametrize( + "test_case", + ( + SqlTestStatusAlignmentTestCase( + description="short names share one status column", + test_name="test_daily_revenue_chain", + expected_models=("stg_orders", "daily_revenue"), + assertion_names=("revenue_is_non_negative",), + expected_row_count=4, + ), + SqlTestStatusAlignmentTestCase( + description="longest name is a child assertion", + test_name="test_daily_revenue_chain", + expected_models=("stg_orders",), + assertion_names=("daily_revenue_is_non_negative_for_every_completed_order_day",), + expected_row_count=3, + ), + SqlTestStatusAlignmentTestCase( + description="longest name is the parent test", + test_name="test_daily_revenue_chain_with_refunds_and_partial_payment_retries", + expected_models=("stg_orders", "daily_revenue"), + assertion_names=(), + expected_row_count=3, + ), + ), + ids=lambda case: case.description, +) +def test_given_failed_sql_test_when_reporting_test_command_progress_then_child_statuses_align( + test_case: SqlTestStatusAlignmentTestCase, +) -> None: + stream: StringIO = StringIO() + callbacks: NestedCommandProgressCallbacks = NestedCommandProgressCallbacks( + total=1, + label="test", + stream=stream, + use_color=False, + name_width=resolve_test_name_width( + ( + build_sql_test_plan_entry( + test_name=test_case.test_name, + expected_models=test_case.expected_models, + assertion_names=test_case.assertion_names, + ), + ) + ), + ) + result: SqlTestExecutionResult = build_sql_test_result( + test_name=test_case.test_name, + expected_models=test_case.expected_models, + assertion_names=test_case.assertion_names, + outcome=SqlTestOutcome.FAIL, + ) + + callbacks.on_item_start(group_name="daily_revenue", item_name=test_case.test_name) + callbacks.on_item_complete( + group_name="daily_revenue", + item_name=test_case.test_name, + status_text="FAIL", + child_rows=build_test_expectation_rows(result), + ) + columns: tuple[int, ...] = status_columns(stream.getvalue()) + + assert len(columns) == test_case.expected_row_count, stream.getvalue() + assert len(set(columns)) == 1, stream.getvalue() + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"])