From ce1935fabee0c82a760586b91c155d7487af3c07 Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Tue, 10 Nov 2026 04:19:01 +0000 Subject: [PATCH] feat(cli): render attached and standalone audit SQL Signed-off-by: Vamsi-klu --- docs/concepts/audits.md | 30 ++++ docs/reference/cli.md | 10 +- sqlmesh/cli/main.py | 64 ++++++- sqlmesh/core/context.py | 83 ++++++++- tests/cli/test_cli.py | 272 ++++++++++++++++++++++++++++ tests/core/test_context.py | 354 ++++++++++++++++++++++++++++++++++++- 6 files changed, 803 insertions(+), 10 deletions(-) diff --git a/docs/concepts/audits.md b/docs/concepts/audits.md index 18368cd103..e2464b8626 100644 --- a/docs/concepts/audits.md +++ b/docs/concepts/audits.md @@ -190,6 +190,36 @@ Standalone audits can declare dependencies using the `depends_on` property. SQLM !!! note Standalone audits are non-blocking only. Because they are not associated with a single model, SQLMesh cannot determine which model should be blocked if the audit fails. + +## Render audit SQL + +Use [`sqlmesh render --audit`](../reference/cli.md#render) to inspect an audit's rendered failure query. Pass the audit's metadata `name`, rather than its SQL filename. + +For an audit attached to a model, supply the model name. SQLMesh uses the audit definition and arguments configured on that model, including built-in, user-defined, and inline audits. For the parameterized example above: + +```bash +sqlmesh render sushi.items --audit does_not_exceed_threshold \ + --start 2024-01-01 --end 2024-01-31 --execution-time 2024-02-01 +``` + +This renders both invocations in their configured order, keeping their distinct `column` and `threshold` arguments. Each statement has a SQL comment identifying the audit and invocation number, and ends with a semicolon. Blank lines separate the statements. The audit must be attached to the supplied model; SQLMesh does not infer arguments for an unattached audit. + +For a standalone audit, omit the model name. For the standalone example above, use `--no-format` to write plain SQL to a file: + +```bash +sqlmesh render --audit assert_item_price_is_not_null \ + --start 2024-01-01 --end 2024-01-31 --execution-time 2024-02-01 \ + --no-format > audit.sql +``` + +Without a model name, SQLMesh resolves only standalone audits. If an attached and a standalone audit share a name, supplying the model selects the attached audit; omitting the model selects the standalone audit. Explicitly requested audits render even if they have `skip TRUE`, so you can inspect queries that normal execution would skip. + +The output uses the audit's declared dialect, falling back to the attached model's dialect or the owning project's dialect. Pass `--dialect` to choose a different output dialect. The audit's input dialect still controls how SQLMesh parses its definition. Formatting options such as `--indent` override the owning project's formatting defaults. + +Audit rendering does not support `--expand`. It uses the audit's normal table references and interval filters. Model rendering without `--audit` keeps its existing behavior and requires a model name. + +The command does not execute the audit's failure query. Project loading can read state, rendering can inspect column metadata, and user-defined macros can execute database queries as usual. + ## Built-in audits SQLMesh comes with a suite of built-in generic audits that cover a broad set of common use cases. Built-in audits are blocking by default, but they all have non-blocking counterparts which you can use by appending `_non_blocking` - see [Non-blocking audits](#non-blocking-audits). diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 460b6e520b..fb83b64953 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -432,11 +432,15 @@ Options: ## render ``` -Usage: sqlmesh render [OPTIONS] MODEL +Usage: sqlmesh render [OPTIONS] [MODEL] - Render a model's query, optionally expanding referenced models. + Render a model's query, optionally expanding referenced models, or render + audit queries. Options: + --audit TEXT Render all invocations of this audit on MODEL, + or a standalone audit without MODEL. Cannot be + used with --expand. -s, --start TEXT The start datetime of the interval for which this command will be applied. -e, --end TEXT The end datetime of the interval for which this @@ -464,6 +468,8 @@ Options: --help Show this message and exit. ``` +`MODEL` is required when rendering a model or an attached audit. Omit it only when rendering a standalone audit with `--audit`. Use the audit's metadata name, rather than its filename. See [Render audit SQL](../concepts/audits.md#render-audit-sql) for examples, repeated invocations, output dialect selection, and rendering behavior. + ## rewrite ``` diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py index fc6cad8011..2a847850a6 100644 --- a/sqlmesh/cli/main.py +++ b/sqlmesh/cli/main.py @@ -4,9 +4,11 @@ import os import sys import typing as t +from contextlib import redirect_stdout from pathlib import Path import click +from sqlglot.errors import SqlglotError from sqlmesh import configure_logging, remove_excess_logs from sqlmesh.cli import error_handler @@ -169,7 +171,11 @@ def cli( logger.exception("Failed to initialize SQLMesh context") raise - if load and not context.models: + if ( + load + and not context.models + and not (ctx.invoked_subcommand == "render" and context.standalone_audits) + ): raise click.ClickException( f"`{paths}` doesn't seem to have any models... cd into the proper directory or specify the path(s) with -p." ) @@ -279,7 +285,11 @@ def init( @cli.command("render") -@click.argument("model") +@click.argument("model", required=False) +@click.option( + "--audit", + help="Render all invocations of this audit on MODEL, or a standalone audit without MODEL. Cannot be used with --expand.", +) @opt.start_time @opt.end_time @opt.execution_time @@ -296,16 +306,62 @@ def init( @cli_analytics def render( ctx: click.Context, - model: str, + model: t.Optional[str], start: TimeLike, end: TimeLike, execution_time: t.Optional[TimeLike] = None, expand: t.Optional[t.Union[bool, t.Iterable[str]]] = None, dialect: t.Optional[str] = None, no_format: bool = False, + audit: t.Optional[str] = None, **format_kwargs: t.Any, ) -> None: - """Render a model's query, optionally expanding referenced models.""" + """Render a model's query, optionally expanding referenced models, or render audit queries.""" + if audit is not None: + if expand: + raise click.UsageError("--audit cannot be used with --expand.", ctx) + try: + with redirect_stdout(sys.stderr): + queries = ctx.obj.render_audit( + model, + audit_name=audit, + start=start, + end=end, + execution_time=execution_time, + ) + owner = ( + ctx.obj.get_model(model, raise_if_missing=True) + if model is not None + else ctx.obj.standalone_audits[audit.lower()] + ) + definition = next(a for a, _ in owner.audits_with_args if a.name == audit.lower()) + config = ctx.obj.config_for_node(owner) + options = { + **config.format.generator_options, + **{k: v for k, v in format_kwargs.items() if v is not None}, + } + output_dialect = ( + dialect + if dialect is not None + else definition.dialect or owner.dialect or config.dialect + ) + label = " ".join(definition.name.splitlines()) + statements = [ + f"-- Audit: {label}, invocation {index}\n{query.sql(pretty=True, dialect=output_dialect, **options)};" + for index, query in enumerate(queries, 1) + ] + except (SQLMeshError, SqlglotError, ValueError) as ex: + # Click reports these errors on stderr, preserving redirectable SQL on stdout. + raise click.ClickException(str(ex)) from ex + sql = "\n\n".join(statements) + if no_format: + print(sql) + else: + ctx.obj.console.show_sql(sql) + return + + if model is None: + raise click.MissingParameter(param_hint="'MODEL'", param_type="argument", ctx=ctx) model = ctx.obj.get_model(model, raise_if_missing=True) rendered = ctx.obj.render( diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index fd7f8ad071..31742ca1c0 100644 --- a/sqlmesh/core/context.py +++ b/sqlmesh/core/context.py @@ -80,7 +80,7 @@ from sqlmesh.core.loader import Loader from sqlmesh.core.linter.definition import AnnotatedRuleViolation, Linter from sqlmesh.core.linter.rules import BUILTIN_RULES -from sqlmesh.core.macros import ExecutableOrMacro, macro +from sqlmesh.core.macros import ExecutableOrMacro, RuntimeStage, macro from sqlmesh.core.metric import Metric, rewrite from sqlmesh.core.model import Model, update_model_schemas from sqlmesh.core.config.model import ModelDefaultsConfig @@ -1286,6 +1286,87 @@ def render( **kwargs, ) + @python_api_analytics + def render_audit( + self, + model_or_snapshot: t.Optional[ModelOrSnapshot] = None, + *, + audit_name: str, + start: t.Optional[TimeLike] = None, + end: t.Optional[TimeLike] = None, + execution_time: t.Optional[TimeLike] = None, + ) -> t.List[exp.Query]: + """Render an audit's failure queries without running them. + + Args: + model_or_snapshot: The audited model, name, or snapshot. Omit for a standalone audit. + audit_name: The audit metadata name. All effective invocations on the model are rendered. + start: The start of the interval to render. + end: The end of the interval to render. + execution_time: The execution time reference, defaulting to now for the entire call. + + Returns: + Failure queries in invocation order, including explicitly requested skipped audits. + + Rendering can read column metadata, and user macros retain their normal execution behavior. + """ + audit_name = audit_name.lower() + model = ( + self.get_model(model_or_snapshot, raise_if_missing=True) + if model_or_snapshot is not None + else None + ) + if model is not None: + audits = [ + (audit, args) for audit, args in model.audits_with_args if audit.name == audit_name + ] + if not audits: + raise SQLMeshError(f"Audit '{audit_name}' is not attached to model '{model.name}'.") + else: + standalone_audit = self.standalone_audits.get(audit_name) + if standalone_audit is None: + from sqlmesh.core.audit.builtin import BUILT_IN_AUDITS + + if audit_name in BUILT_IN_AUDITS or any( + audit.name == audit_name + for model in self.models.values() + for audit, _ in model.audits_with_args + ): + raise SQLMeshError(f"Audit '{audit_name}' requires a model. Supply MODEL.") + raise SQLMeshError( + f"Cannot find standalone audit '{audit_name}'. Model-attached audits require MODEL." + ) + audits = [(standalone_audit, {})] + + snapshots = self.snapshots + deployability_index = DeployabilityIndex.create(snapshots.values(), start=start) + execution_time = now() if execution_time is None else execution_time + adapter = self._get_engine_adapter(model.gateway if model is not None else None) + queries = [] + for audit, audit_args in audits: + audit_args = audit_args.copy() + audit_args.pop("blocking", None) + kwargs: t.Dict[str, t.Any] = { + "start": start, + "end": end, + "execution_time": execution_time, + "snapshots": snapshots, + "deployability_index": deployability_index, + "engine_adapter": adapter, + "runtime_stage": RuntimeStage.AUDITING, + **audit_args, + } + if model is not None: + snapshot = snapshots.get(model.fqn) + # Before the first plan there is no physical table to map to, as in model rendering. + if snapshot is not None and snapshot.version is None: + kwargs.setdefault("this_model", model.fqn) + query = model.render_audit_query(audit, **kwargs) + else: + query = t.cast(StandaloneAudit, audit).render_audit_query(**kwargs) + queries.append(query) + return queries + @python_api_analytics def evaluate( self, diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py index f1540727b1..e02d6d7809 100644 --- a/tests/cli/test_cli.py +++ b/tests/cli/test_cli.py @@ -11,6 +11,7 @@ from click import ClickException from click.testing import CliRunner from sqlglot import __version__ as SQLGLOT_VERSION +from sqlglot import exp, parse as parse_sql from sqlmesh import RuntimeEnv from sqlmesh._version import __version__ as SQLMESH_VERSION from sqlmesh.cli.project_init import ProjectTemplate, init_example_project @@ -2341,6 +2342,277 @@ def test_render(runner: CliRunner, tmp_path: Path): assert expected in cleaned_output +@pytest.fixture +def render_audit_project(tmp_path: Path) -> Path: + create_example_project(tmp_path, template=ProjectTemplate.EMPTY) + (tmp_path / "models" / "orders.sql").write_text( + """MODEL ( + name demo.orders, + kind INCREMENTAL_BY_TIME_RANGE (time_column ds), + columns (id INT, region TEXT, ds DATE), + audits ( + minimum_id(minimum := 0, blocking := false), + minimum_id(minimum := 10), + not_null(columns := (id), condition := (region = 'us')) + ) + ); + SELECT 1 AS id, 'us' AS region, CAST('2026-11-10' AS DATE) AS ds; + """ + ) + (tmp_path / "audits").mkdir(exist_ok=True) + (tmp_path / "audits" / "minimum_id.sql").write_text( + """AUDIT (name minimum_id, defaults (minimum = -1)); + SELECT * FROM @this_model WHERE id <= @minimum; + """ + ) + (tmp_path / "audits" / "check_orders.sql").write_text( + """AUDIT (name check_orders, standalone TRUE); + SELECT id, @execution_ds AS rendered_on FROM demo.orders WHERE id < 0; + """ + ) + return tmp_path + + +@pytest.mark.parametrize("audit_name,count", [("MINIMUM_ID", 2), ("not_null", 1)]) +def test_render_audit_attached(runner: CliRunner, render_audit_project: Path, audit_name, count): + result = runner.invoke( + cli, + [ + "--paths", + str(render_audit_project), + "render", + "demo.orders", + "--audit", + audit_name, + "--start", + "2026-11-10", + "--end", + "2026-11-10", + "--execution-time", + "2026-11-11", + "--dialect", + "postgres", + "--no-format", + ], + ) + assert result.exit_code == 0, result.output + queries = parse_sql(result.stdout, read="postgres") + assert len(queries) == count + assert all(isinstance(query, exp.Query) for query in queries) + assert result.stdout.count("-- Audit:") == count + assert "2026-11-10" in result.stdout + if count == 2: + for query, minimum in zip(queries, (0, 10)): + assert query is not None + assert f"<= {minimum}" in query.sql() + else: + assert "'us'" in result.stdout + assert "IS NULL" in result.stdout + + +def test_render_audit_standalone(runner: CliRunner, render_audit_project: Path): + result = runner.invoke( + cli, + [ + "--paths", + str(render_audit_project), + "render", + "--audit", + "CHECK_ORDERS", + "--execution-time", + "2026-11-11", + "--no-format", + ], + ) + assert result.exit_code == 0, result.output + assert len(parse_sql(result.stdout, read="duckdb")) == 1 + assert "2026-11-11" in result.stdout + + +@pytest.mark.parametrize("audit_option", [["--audit", "check_orders"], ["--audit=check_orders"]]) +def test_render_audit_project_without_models( + runner: CliRunner, render_audit_project: Path, audit_option +): + (render_audit_project / "models" / "orders.sql").unlink() + result = runner.invoke( + cli, + ["--paths", str(render_audit_project), "render", *audit_option, "--no-format"], + ) + assert result.exit_code == 0, result.output + (query,) = parse_sql(result.stdout, read="duckdb") + assert isinstance(query, exp.Query) + + +def test_render_audit_unattached_custom_audit_requires_model( + runner: CliRunner, render_audit_project: Path +): + (render_audit_project / "audits" / "unattached.sql").write_text( + "AUDIT (name unattached); SELECT * FROM @this_model WHERE id < 0;" + ) + result = runner.invoke( + cli, + ["--paths", str(render_audit_project), "render", "--audit", "unattached", "--no-format"], + ) + assert result.exit_code != 0 + assert result.stdout == "" + assert "Model-attached audits require MODEL" in result.stderr + + +@pytest.mark.parametrize( + "arguments,message", + [ + ([], "Missing argument 'MODEL'"), + (["--audit", "minimum_id"], "requires a model"), + (["--audit", "missing"], "standalone audit"), + (["demo.orders", "--audit", "valid_http_method"], "not attached"), + (["missing", "--audit", "minimum_id"], "Cannot find model"), + (["demo.orders", "--audit", "minimum_id", "--expand", "True"], "--expand"), + (["demo.orders", "--audit", "minimum_id", "--expand", "demo.orders"], "--expand"), + ], +) +def test_render_audit_errors(runner: CliRunner, render_audit_project: Path, arguments, message): + result = runner.invoke(cli, ["--paths", str(render_audit_project), "render", *arguments]) + assert result.exit_code != 0 + assert result.stdout == "" + assert message in result.stderr + + +def test_render_audit_later_render_failure_has_no_output( + runner: CliRunner, render_audit_project: Path +): + (render_audit_project / "audits" / "minimum_id.sql").write_text( + "AUDIT (name minimum_id); SELECT * FROM @this_model WHERE id <= @minimum;" + ) + model_path = render_audit_project / "models" / "orders.sql" + model_path.write_text( + model_path.read_text().replace("minimum_id(minimum := 10)", "minimum_id()") + ) + result = runner.invoke( + cli, + [ + "--paths", + str(render_audit_project), + "render", + "demo.orders", + "--audit", + "minimum_id", + "--no-format", + ], + ) + assert result.exit_code != 0 + assert result.stdout == "" + assert "minimum" in result.stderr + + +def test_render_audit_later_serialization_failure_has_no_output( + runner: CliRunner, render_audit_project: Path, mocker +): + query = exp.select("1") + invalid_query = mocker.Mock(spec=exp.Query) + invalid_query.sql.side_effect = ValueError("cannot serialize second invocation") + mocker.patch.object(Context, "render_audit", return_value=[query, invalid_query]) + result = runner.invoke( + cli, + [ + "--paths", + str(render_audit_project), + "render", + "demo.orders", + "--audit", + "minimum_id", + "--no-format", + ], + ) + assert result.exit_code != 0 + assert result.stdout == "" + assert "second invocation" in result.stderr + + +@pytest.mark.parametrize("indent", [None, 2]) +def test_render_audit_uses_owning_project_generator_options( + runner: CliRunner, render_audit_project: Path, indent +): + other = render_audit_project.with_name(render_audit_project.name + "_other") + other.mkdir() + create_example_project(other, template=ProjectTemplate.EMPTY) + (other / "models" / "orders.sql").write_text( + (render_audit_project / "models" / "orders.sql") + .read_text() + .replace("demo.orders", "other.orders") + ) + (other / "audits").mkdir(exist_ok=True) + (other / "audits" / "minimum_id.sql").write_text( + (render_audit_project / "audits" / "minimum_id.sql").read_text() + ) + config = other / "config.yaml" + config.write_text(config.read_text() + "\nformat:\n indent: 6\n") + arguments = [ + "--paths", + str(render_audit_project), + "--paths", + str(other), + "render", + "other.orders", + "--audit", + "minimum_id", + "--no-format", + ] + if indent is not None: + arguments.extend(["--indent", str(indent)]) + result = runner.invoke(cli, arguments) + assert result.exit_code == 0, result.output + assert f"FROM (\n{' ' * (indent or 6)}SELECT" in result.stdout + assert len(parse_sql(result.stdout, read="duckdb")) == 2 + + +def test_render_audit_labels_are_valid_comments(runner: CliRunner, render_audit_project: Path): + name = "odd\n*/ SELECT 7; --" + model_path = render_audit_project / "models" / "orders.sql" + model_path.write_text(model_path.read_text().replace("minimum_id", f'"{name}"')) + audit_path = render_audit_project / "audits" / "minimum_id.sql" + audit_path.write_text(audit_path.read_text().replace("name minimum_id", f'name "{name}"')) + result = runner.invoke( + cli, + [ + "--paths", + str(render_audit_project), + "render", + "demo.orders", + "--audit", + name, + "--no-format", + ], + ) + assert result.exit_code == 0, result.output + assert len(parse_sql(result.stdout, read="duckdb")) == 2 + assert "-- Audit: odd */ select 7; --, invocation 1" in result.stdout + + +@pytest.mark.parametrize("output_dialect", [None, "duckdb"]) +def test_render_audit_input_and_output_dialects( + runner: CliRunner, render_audit_project: Path, output_dialect +): + audit_path = render_audit_project / "audits" / "check_orders.sql" + audit_path.write_text("""AUDIT (name check_orders, standalone TRUE, dialect spark); + SELECT DATE_FORMAT(CAST('2026-11-10' AS DATE), 'yyyy-MM-dd') AS rendered_on; + """) + arguments = [ + "--paths", + str(render_audit_project), + "render", + "--audit", + "check_orders", + "--no-format", + ] + if output_dialect is not None: + arguments.extend(["--dialect", output_dialect]) + result = runner.invoke(cli, arguments) + assert result.exit_code == 0, result.output + (query,) = parse_sql(result.stdout, read=output_dialect or "spark") + assert isinstance(query, exp.Query) + assert ("STRFTIME" if output_dialect else "DATE_FORMAT") in result.stdout + + @time_machine.travel(FREEZE_TIME) def test_signals(runner: CliRunner, tmp_path: Path): create_example_project(tmp_path, template=ProjectTemplate.EMPTY) diff --git a/tests/core/test_context.py b/tests/core/test_context.py index 71276b7396..2b7e451a87 100644 --- a/tests/core/test_context.py +++ b/tests/core/test_context.py @@ -17,7 +17,7 @@ import sqlmesh.core.constants from sqlmesh.cli.project_init import init_example_project -from sqlmesh.core.audit import StandaloneAudit +from sqlmesh.core.audit import ModelAudit, StandaloneAudit, load_audit from sqlmesh.core.console import TerminalConsole from sqlmesh.core import dialect as d, constants as c from sqlmesh.core.config import ( @@ -42,11 +42,12 @@ from sqlmesh.core.plan.definition import Plan from sqlmesh.core.macros import MacroEvaluator, RuntimeStage from sqlmesh.core.model import load_sql_based_model, model, SqlModel, Model, update_model_schemas +from sqlmesh.core.model import create_external_model, create_python_model, create_seed_model from sqlmesh.core.model.common import ParsableSql from sqlmesh.core.model.cache import OptimizedQueryCache -from sqlmesh.core.snapshot import SnapshotChangeCategory +from sqlmesh.core.snapshot import DeployabilityIndex, SnapshotChangeCategory from sqlmesh.core.renderer import render_statements -from sqlmesh.core.model.kind import ModelKindName +from sqlmesh.core.model.kind import ModelKindName, SeedKind from sqlmesh.core.state_sync.cache import CachingStateSync from sqlmesh.core.state_sync.db import EngineAdapterStateSync from sqlmesh.utils.connection_pool import SingletonConnectionPool, ThreadLocalSharedConnectionPool @@ -66,6 +67,7 @@ NoChangesPlanError, ) from sqlmesh.utils.metaprogramming import Executable +from sqlmesh.utils.jinja import JinjaMacroRegistry, MacroExtractor from sqlmesh.utils.windows import IS_WINDOWS, fix_windows_path from tests.utils.test_helpers import use_terminal_console from tests.utils.test_filesystem import create_temp_file @@ -277,6 +279,352 @@ def test_render_seed_model(sushi_context, assert_exp_eq): ) +@pytest.fixture +def audit_render_context(): + context = Context(config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))) + audit = ModelAudit( + name="minimum_id", + query="SELECT id FROM @this_model WHERE id <= @minimum", + defaults={"minimum": exp.Literal.number(-1)}, + ) + context.upsert_model( + load_sql_based_model( + parse( + """MODEL ( + name demo.orders, + kind INCREMENTAL_BY_TIME_RANGE (time_column ds), + columns (id INT, region TEXT, ds DATE), + audits ( + minimum_id(minimum := 0, blocking := false), + minimum_id(minimum := 10), + not_null(columns := (id), condition := (region = 'us')) + ) + ); + SELECT 1 AS id, 'us' AS region, CAST('2026-11-10' AS DATE) AS ds; + """, + default_dialect="duckdb", + ), + dialect="duckdb", + default_catalog=context.default_catalog, + audit_definitions={audit.name: audit}, + ) + ) + context._standalone_audits["minimum_id"] = StandaloneAudit( + name="minimum_id", dialect="duckdb", query="SELECT @execution_ds AS rendered_on" + ) + yield context + context.close() + + +def test_render_audit_arguments_and_failure_queries(audit_render_context: Context): + context = audit_render_context + model = context.get_model("demo.orders", raise_if_missing=True) + original_audits = model.json() + adapter = context.engine_adapter + adapter.execute("CREATE SCHEMA demo") + adapter.execute("CREATE TABLE demo.orders (id INT, region TEXT, ds DATE)") + adapter.execute("""INSERT INTO demo.orders VALUES + (-1, 'us', '2026-11-10'), (5, 'eu', '2026-11-10'), + (-2, 'us', '2026-11-09'), (NULL, 'us', '2026-11-10'), + (NULL, 'eu', '2026-11-10'), (NULL, NULL, '2026-11-10')""") + kwargs = dict(start="2026-11-10", end="2026-11-10", execution_time="2026-11-11") + + queries = context.render_audit("demo.orders", audit_name="MINIMUM_ID", **kwargs) + assert [adapter.fetchall(query) for query in queries] == [[(-1,)], [(-1,), (5,)]] + assert context.render_audit(model, audit_name="minimum_id", **kwargs) == queries + assert ( + context.render_audit( + context.get_snapshot(model, raise_if_missing=True), audit_name="minimum_id", **kwargs + ) + == queries + ) + null_queries = context.render_audit("demo.orders", audit_name="not_null", **kwargs) + assert adapter.fetchall(null_queries[0]) == [(None, "us", date(2026, 11, 10))] + assert model.json() == original_audits + assert "blocking" in model.audits[0][1] + + +def test_render_audit_standalone_name_collision(audit_render_context: Context): + context = audit_render_context + (query,) = context.render_audit(audit_name="MINIMUM_ID", execution_time="2026-11-11") + assert context.engine_adapter.fetchall(query) == [("2026-11-11",)] + assert len(context.render_audit("demo.orders", audit_name="minimum_id")) == 2 + + +@pytest.mark.parametrize("deployable", [True, False]) +def test_render_audit_snapshot_relation(audit_render_context: Context, mocker, deployable): + context = audit_render_context + model = context.get_model("demo.orders", raise_if_missing=True) + snapshots = context.snapshots + snapshot = snapshots[model.fqn] + assert snapshot.version is None + unplanned = context.render_audit(model, audit_name="minimum_id") + assert model.fqn in unplanned[0].sql() + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + index = ( + DeployabilityIndex.all_deployable() if deployable else DeployabilityIndex.none_deployable() + ) + mocker.patch.object(Context, "snapshots", new_callable=PropertyMock, return_value=snapshots) + mocker.patch.object(DeployabilityIndex, "create", return_value=index) + query, _ = context.render_audit(snapshot, audit_name="minimum_id") + expected_table = exp.to_table(snapshot.table_name(deployable)) + assert any( + (table.catalog, table.db, table.name) + == (expected_table.catalog, expected_table.db, expected_table.name) + for table in query.find_all(exp.Table) + ) + assert query == model.render_audit_query( + model.audits_with_args[0][0], + snapshots=snapshots, + deployability_index=index, + engine_adapter=context.engine_adapter, + runtime_stage=RuntimeStage.AUDITING, + minimum=exp.Literal.number(0), + ) + + +def test_render_audit_execution_boundary_and_gateway(audit_render_context: Context, mocker): + context = audit_render_context + context.upsert_model("demo.orders", gateway="other") + adapter = mocker.Mock(spec=DuckDBEngineAdapter) + adapter.columns.return_value = {"ds": exp.DataType.build("DATE")} + context.engine_adapters["other"] = adapter + forbidden = [ + mocker.patch.object(context, "plan"), + mocker.patch.object(context, "apply"), + mocker.patch.object(context, "evaluate"), + mocker.patch.object(context.snapshot_evaluator, "_audit"), + mocker.patch.object(context.snapshot_evaluator, "evaluate"), + ] + model = context.get_model("demo.orders", raise_if_missing=True) + for method in ("render_pre_statements", "render_post_statements"): + forbidden.append(mocker.patch.object(type(model), method)) + queries = context.render_audit(model, audit_name="minimum_id") + assert len(queries) == 2 + assert adapter.columns.call_count == 2 # Metadata reads are allowed. + adapter.fetchone.assert_not_called() + adapter.fetchall.assert_not_called() + adapter.execute.assert_not_called() + for operation in forbidden: + operation.assert_not_called() + + +@pytest.mark.parametrize("skip,blocking", [(False, True), (True, True), (True, False)]) +def test_render_audit_skip_stage_defaults_and_overrides( + audit_render_context: Context, skip, blocking +): + context = audit_render_context + audit = ModelAudit( + name="minimum_id", + skip=skip, + blocking=blocking, + defaults={"minimum": exp.Literal.number(-1)}, + query="SELECT @minimum AS threshold, @runtime_stage AS stage, @execution_ds AS rendered_on", + ) + context.upsert_model("demo.orders", audit_definitions={audit.name: audit}) + queries = context.render_audit( + "demo.orders", audit_name="minimum_id", execution_time="2026-11-11" + ) + assert [context.engine_adapter.fetchall(query) for query in queries] == [ + [(0, "auditing", "2026-11-11")], + [(10, "auditing", "2026-11-11")], + ] + # An inline audit without an explicit invocation still uses its defaults. + context.upsert_model("demo.orders", audits=[]) + (query,) = context.render_audit( + "demo.orders", audit_name="minimum_id", execution_time="2026-11-11" + ) + assert context.engine_adapter.fetchall(query) == [(-1, "auditing", "2026-11-11")] + + +def test_render_audit_resolves_one_execution_time(audit_render_context: Context, mocker): + context = audit_render_context + audit = ModelAudit(name="minimum_id", query="SELECT @execution_ds AS rendered_on") + context.upsert_model("demo.orders", audit_definitions={audit.name: audit}) + clock = mocker.patch( + "sqlmesh.core.context.now", + side_effect=[to_datetime("2026-11-10"), to_datetime("2026-11-11")], + ) + queries = context.render_audit("demo.orders", audit_name="minimum_id") + assert [context.engine_adapter.fetchall(query) for query in queries] == [ + [("2026-11-10",)], + [("2026-11-10",)], + ] + clock.assert_called_once() + + +def test_render_audit_builtin_override(audit_render_context: Context): + context = audit_render_context + override = ModelAudit(name="not_null", query="SELECT 'custom override' AS failure") + model = context.get_model("demo.orders", raise_if_missing=True) + context.upsert_model( + "demo.orders", audit_definitions={**model.audit_definitions, "not_null": override} + ) + (query,) = context.render_audit("demo.orders", audit_name="not_null") + assert context.engine_adapter.fetchall(query) == [("custom override",)] + + +@pytest.mark.parametrize( + "model_name,audit_name,message", + [ + ("missing", "minimum_id", "Cannot find model"), + ("demo.orders", "missing", "not attached"), + (None, "not_null", "requires a model"), + (None, "missing", "Cannot find standalone audit"), + ], +) +def test_render_audit_resolution_errors( + audit_render_context: Context, model_name, audit_name, message +): + with pytest.raises(SQLMeshError, match=message): + audit_render_context.render_audit(model_name, audit_name=audit_name) + + +@pytest.mark.parametrize("kind", ["full", "external", "seed", "python"]) +def test_render_audit_does_not_evaluate_model( + audit_render_context: Context, tmp_path: Path, mocker, kind +): + context = audit_render_context + kwargs: t.Dict[str, t.Any] = dict( + dialect="duckdb", + default_catalog=context.default_catalog, + columns={"id": exp.DataType.build("INT")}, + audits=[("not_null", {"columns": exp.Tuple(expressions=[exp.column("id")])})], + ) + name = f"demo.{kind}" + audited_model: Model + if kind == "external": + audited_model = create_external_model(name, **kwargs) + elif kind == "seed": + seed = tmp_path / "ids.csv" + seed.write_text("id\n1\n") + audited_model = create_seed_model(name, SeedKind(path=str(seed)), **kwargs) + elif kind == "python": + audited_model = create_python_model( + name, + "data", + { + "data": Executable( + payload="def data(context, **kwargs):\n raise AssertionError('must not evaluate model')" + ) + }, + depends_on=set(), + kind="FULL", + **kwargs, + ) + else: + audited_model = load_sql_based_model( + parse( + f"MODEL (name {name}, kind FULL, audits (not_null(columns := (id)))); SELECT 1 AS id" + ), + **kwargs, + ) + context.upsert_model(audited_model) + evaluate = mocker.patch.object( + type(audited_model), "render", side_effect=AssertionError("must not evaluate model") + ) + (query,) = context.render_audit(name, audit_name="not_null") + assert '"id" IS NULL' in query.sql() + evaluate.assert_not_called() + + +def test_render_audit_quoted_identifiers(audit_render_context: Context): + context = audit_render_context + context.upsert_model( + load_sql_based_model( + parse( + '''MODEL (name "CaseSchema"."Orders With Space", kind FULL, + audits (not_null(columns := ("MixedCase")))); SELECT 1 AS "MixedCase"''', + default_dialect="postgres", + ), + dialect="postgres", + default_catalog=context.default_catalog, + ) + ) + (query,) = context.render_audit('"CaseSchema"."Orders With Space"', audit_name="not_null") + table = next(query.find_all(exp.Table)) + assert (table.catalog, table.db, table.name) == ("memory", "CaseSchema", "Orders With Space") + assert '"MixedCase" IS NULL' in query.sql() + + +@pytest.mark.parametrize("macro_type", ["sql", "python", "jinja"]) +def test_render_audit_macros(audit_render_context: Context, macro_type): + context = audit_render_context + python_env = {} + if macro_type == "sql": + audit = load_audit( + parse("AUDIT (name macro_audit); @DEF(value, 7); SELECT @value AS result") + ) + elif macro_type == "python": + audit = ModelAudit(name="macro_audit", query="SELECT @audit_value() AS result") + python_env = { + "audit_value": Executable( + payload="def audit_value(evaluator):\n return 7 if evaluator.locals['runtime_stage'] == 'auditing' else 0" + ) + } + else: + registry = JinjaMacroRegistry() + registry.add_macros(MacroExtractor().extract("{% macro audit_value() %}7{% endmacro %}")) + audit = load_audit( + parse( + "AUDIT (name macro_audit);\nJINJA_QUERY_BEGIN;\nSELECT {{ audit_value() }} AS result\nJINJA_END;" + ), + jinja_macros=registry, + ) + audit = audit.copy(update={"jinja_macros": registry}) + context.upsert_model( + "demo.orders", audits=[], audit_definitions={audit.name: audit}, python_env=python_env + ) + (query,) = context.render_audit("demo.orders", audit_name="macro_audit") + assert context.engine_adapter.fetchall(query) == [(7,)] + + +def test_render_audit_timestamp_interval(audit_render_context: Context): + context = audit_render_context + context.upsert_model( + "demo.orders", + columns_to_types_={ + "id": exp.DataType.build("INT"), + "region": exp.DataType.build("TEXT"), + "ds": exp.DataType.build("TIMESTAMP"), + }, + ) + query, _ = context.render_audit( + "demo.orders", + audit_name="minimum_id", + start="2026-11-10", + end="2026-11-10", + ) + interval = next(query.find_all(exp.Between)) + assert "2026-11-10 00:00:00" in interval.args["low"].sql() + assert "2026-11-10 23:59:59.999999" in interval.args["high"].sql() + + +@pytest.mark.parametrize( + "query", ["SELECT @missing AS result", "SELECT @missing_function() AS result"] +) +def test_render_audit_missing_macros_keep_file_context(audit_render_context: Context, query): + context = audit_render_context + audit = ModelAudit(name="broken", query=query) + audit._path = Path("audits/broken.sql") + context.upsert_model("demo.orders", audits=[], audit_definitions={"broken": audit}) + with pytest.raises(SQLMeshError, match="audits/broken.sql"): + context.render_audit("demo.orders", audit_name="broken") + + +def test_render_audit_empty_render_keeps_audit_context(audit_render_context: Context): + context = audit_render_context + audit = load_audit( + parse( + "AUDIT (name empty_audit);\nJINJA_QUERY_BEGIN;\n{% if false %}SELECT 1{% endif %}\nJINJA_END;" + ) + ) + audit._path = Path("audits/empty_audit.sql") + context.upsert_model("demo.orders", audits=[], audit_definitions={audit.name: audit}) + with pytest.raises(SQLMeshError, match="audit 'empty_audit'.*model 'demo.orders'"): + context.render_audit("demo.orders", audit_name="empty_audit") + + @pytest.mark.slow def test_load_only_hydrates_remote_snapshots_missing_locally(sushi_context: Context) -> None: prod = sushi_context.state_reader.get_environment("prod")