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/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/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/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"]) 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