From 0a5ce603bd98797674881bbb727c4cf91fa39d59 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 10:39:31 +0000 Subject: [PATCH 1/3] fix: stop init discarding the reference state it compared against init overwrote the cached reference manifest with the target manifest whenever --reference-target was unset or matched --target, on the reasoning that "reference and target are the same". They are not the same thing: the reference *target* selects which warehouse profile to compile against, while the reference *state* is the baseline to diff against. Conflating them threw away the baseline. Because deleted nodes exist only in that baseline, `delete` could no longer look up what it was meant to drop. Reproduced against a real project: cache.json correctly recorded model.demo.stg_orders as deleted, and generate_delete_map then exited having found nothing to delete. So `dbt-ci delete` was a silent no-op for anyone not passing --reference-target, which is also the default in every README example that uses a local --state directory. A local --state directory is now cached as the baseline just as a downloaded one already was, and the target manifest is only used as the reference when there is genuinely no baseline. Caching it is best-effort: a missing file leaves the clearer error to the comparison that follows rather than aborting init. With the fix the same scenario resolves the deletion, table id included: {'model.demo.stg_orders': {'type': 'model', 'name': 'stg_orders', 'table_id': 'demo.main.stg_orders'}} --- dbt_ci/commands/init/index.py | 21 ++++++- tests/unit/commands/test_init.py | 95 ++++++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 3 deletions(-) diff --git a/dbt_ci/commands/init/index.py b/dbt_ci/commands/init/index.py index 40eae4a..d1c462f 100644 --- a/dbt_ci/commands/init/index.py +++ b/dbt_ci/commands/init/index.py @@ -64,6 +64,16 @@ def init(args: Namespace): # Reload reference manifest file after downloading from storage cache.write_reference_manifest(get_reference_manifest_file(str(local_state_dir))) + elif reference_state_path: + # A local --state directory is just as much a baseline as a downloaded one. + # Caching it is what lets later commands resolve nodes that no longer exist + # in the target - without it, `delete` cannot look up what it should drop. + # Best-effort: if the state isn't there, the comparison below raises a + # clearer error than failing here would. + try: + cache.write_reference_manifest(get_reference_manifest_file(reference_state_path)) + except FileNotFoundError: + logger.debug(f"No reference manifest at '{reference_state_path}' to cache yet.") # Compile dbt and generate reference manifest.json file #run_multiprocessed @@ -77,12 +87,17 @@ def init(args: Namespace): # Different targets - will compile again later with actual target cache.write_target_manifest(target_manifest_file) else: - # Same target or no reference target specified - reference and target are the same + # Same target, or none specified. The reference *target* (which warehouse to + # compile against) is independent of the reference *state* (the baseline to + # compare with), so an already-cached baseline must not be replaced here - + # doing so discarded the only copy of the deleted nodes' metadata. logger.debug("Reference target is the same as current target!") - logger.debug("Using the same manifest for both reference and target state.") - cache.write_reference_manifest(target_manifest_file) cache.write_target_manifest(target_manifest_file) + if cache.get_cache("reference_manifest.json") is None: + logger.debug("No reference state available. Using the target manifest as the reference.") + cache.write_reference_manifest(target_manifest_file) + # Will generate summary and output it in the logs. It also covers: # 1. Migration plan for partitioning changes # 2. Ephemeral plan diff --git a/tests/unit/commands/test_init.py b/tests/unit/commands/test_init.py index 8818e39..970147a 100644 --- a/tests/unit/commands/test_init.py +++ b/tests/unit/commands/test_init.py @@ -357,3 +357,98 @@ def test_resolve_manifest_creates_directory(self, mock_logger, mock_path, mock_o # Verify download was called storage_connector["download"].assert_called_once_with(state_uri) + + +class TestReferenceManifestCaching: + """Test that init preserves the reference baseline it compared against. + + The reference *target* (which warehouse profile to compile against) is independent + of the reference *state* (the baseline to diff against). Conflating them overwrote + the cached baseline with the target manifest, which left `delete` unable to look up + the nodes it was supposed to drop. + """ + + def _args(self, tmp_path, **overrides): + """Build an init args namespace pointing at a local state directory.""" + defaults = { + "dbt_project_dir": str(tmp_path), + "reference_state": str(tmp_path / "state"), + "reference_target": None, + "target": "dev", + "state_uri": None, + "comparison_strategy": "dbt", + "runner": "dbt", + "reference_path": "reference", + } + defaults.update(overrides) + return Namespace(**defaults) + + @patch("dbt_ci.commands.init.index.init_summary") + @patch("dbt_ci.commands.init.index.StateModified") + @patch("dbt_ci.commands.init.index.DbtCommands") + @patch("dbt_ci.commands.init.index.get_manifest_file") + @patch("dbt_ci.commands.init.index.get_reference_manifest_file") + @patch("dbt_ci.commands.init.index.CacheManager") + @patch("dbt_ci.commands.init.index.click.secho") + def test_local_state_is_cached_as_the_reference( + self, _secho, mock_cache, mock_ref_manifest, mock_target_manifest, + _dbt_commands, mock_state_modified, _summary, tmp_path, + ): + """A --state directory is cached as the baseline, not discarded.""" + mock_cache.return_value.get_cache.return_value = {"nodes": {}} + mock_ref_manifest.return_value = {"nodes": {"model.p.gone": {}}} + mock_target_manifest.return_value = {"nodes": {}} + mock_state_modified.return_value.get_state_modified.return_value = {} + + index(self._args(tmp_path)) + + # The baseline is written from the state directory... + mock_cache.return_value.write_reference_manifest.assert_called_once_with( + {"nodes": {"model.p.gone": {}}} + ) + # ...and is not replaced by the target manifest afterwards. + assert mock_cache.return_value.write_reference_manifest.call_count == 1 + + @patch("dbt_ci.commands.init.index.init_summary") + @patch("dbt_ci.commands.init.index.StateModified") + @patch("dbt_ci.commands.init.index.DbtCommands") + @patch("dbt_ci.commands.init.index.get_manifest_file") + @patch("dbt_ci.commands.init.index.get_reference_manifest_file") + @patch("dbt_ci.commands.init.index.CacheManager") + @patch("dbt_ci.commands.init.index.click.secho") + def test_target_manifest_used_when_no_baseline_exists( + self, _secho, mock_cache, mock_ref_manifest, mock_target_manifest, + _dbt_commands, mock_state_modified, _summary, tmp_path, + ): + """With no reference state at all, the target manifest stands in as the baseline.""" + mock_cache.return_value.get_cache.return_value = None + mock_ref_manifest.side_effect = FileNotFoundError("no manifest") + mock_target_manifest.return_value = {"nodes": {"model.p.a": {}}} + mock_state_modified.return_value.get_state_modified.return_value = {} + + index(self._args(tmp_path)) + + mock_cache.return_value.write_reference_manifest.assert_called_once_with( + {"nodes": {"model.p.a": {}}} + ) + + @patch("dbt_ci.commands.init.index.init_summary") + @patch("dbt_ci.commands.init.index.StateModified") + @patch("dbt_ci.commands.init.index.DbtCommands") + @patch("dbt_ci.commands.init.index.get_manifest_file") + @patch("dbt_ci.commands.init.index.get_reference_manifest_file") + @patch("dbt_ci.commands.init.index.CacheManager") + @patch("dbt_ci.commands.init.index.click.secho") + def test_missing_state_does_not_abort_init( + self, _secho, mock_cache, mock_ref_manifest, mock_target_manifest, + _dbt_commands, mock_state_modified, _summary, tmp_path, + ): + """Caching the baseline is best-effort; a missing file must not fail init.""" + mock_cache.return_value.get_cache.return_value = {"already": "cached"} + mock_ref_manifest.side_effect = FileNotFoundError("no manifest") + mock_target_manifest.return_value = {"nodes": {}} + mock_state_modified.return_value.get_state_modified.return_value = {} + + index(self._args(tmp_path)) + + mock_state_modified.return_value.get_state_modified.assert_called_once() From 224a2c7246b86226cd84157dac493daa6ce1a969 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 10:39:31 +0000 Subject: [PATCH 2/3] feat: add depth-limited selection, run reports and config reuse Three capabilities that were half-built in the codebase but never connected. **--downstream-depth** implements the `levels` parameter that was stubbed with "To be implemented in the future" in three places. Until now a change always selected its entire downstream closure, so touching a core staging model rebuilt nearly the whole project - the opposite of what change-based CI is for. Selection now walks a bounded breadth-first traversal instead. Verified against a real project with a customers -> l1 -> l2 -> l3 chain: depth 0 runs `customers`, depth 1 adds `l1`, depth 2 adds `l2`, omitting the flag runs all four. New and deleted nodes are always included, since a new model has to run whether or not anything depends on it yet. The old code path would have raised KeyError if anyone had passed `levels`: it looked up a `downstream_dependencies_level_N` key that the graph never contains. **dbt-ci report** renders the run report that every command already writes and nothing ever read - init, run, delete, ephemeral and migration record status, timings and variables into report.json, which finalize then deleted. The report covers change counts and node names, command status with durations, and the exposures downstream of the change set. It appends to $GITHUB_STEP_SUMMARY when set, so in GitHub Actions it needs no workflow wiring. Exposures are resolved against the target graph for modified and new nodes, and against the reference graph for deleted ones - an exposure downstream of a model that was just deleted is the case most worth catching. **Cached run configuration** wires up dbt/flags.py, a module written for exactly this and never imported. init records the target and vars it ran with, and the README promises "specify state once in init, reuse everywhere", but that was only true of state paths: every later command made you repeat --target and --vars. They are now inherited from the cache, with an explicitly passed flag still winning. --- README.md | 77 +++++++++ dbt_ci/cli/config/schema.py | 5 + dbt_ci/commands/delete/index.py | 2 + dbt_ci/commands/ephemeral/index.py | 2 + dbt_ci/commands/migration/index.py | 2 + dbt_ci/commands/report/__init__.py | 0 dbt_ci/commands/report/cli.py | 48 ++++++ dbt_ci/commands/report/index.py | 250 +++++++++++++++++++++++++++++ dbt_ci/commands/run/cli.py | 11 ++ dbt_ci/commands/run/index.py | 2 + dbt_ci/commands/run/run.py | 26 ++- dbt_ci/dbt/flags.py | 126 +++++++++------ dbt_ci/graph/graph_utils.py | 78 +++++++-- dbt_ci/main.py | 1 + tests/unit/commands/test_report.py | 157 ++++++++++++++++++ tests/unit/graph/test_depth.py | 121 ++++++++++++++ 16 files changed, 841 insertions(+), 67 deletions(-) create mode 100644 dbt_ci/commands/report/__init__.py create mode 100644 dbt_ci/commands/report/cli.py create mode 100644 dbt_ci/commands/report/index.py create mode 100644 tests/unit/commands/test_report.py create mode 100644 tests/unit/graph/test_depth.py diff --git a/README.md b/README.md index 1bb1ecc..95cf885 100644 --- a/README.md +++ b/README.md @@ -173,8 +173,34 @@ dbt-ci run --dbt-project-dir dbt --mode models | Flag | Aliases | Env Var(s) | Default | Description | |------|---------|-----------|---------|-------------| | `--mode` | `-m`, `--nodes`, `-n` | `DBT_NODES` | `all` | What to run: `all`, `models`, `seeds`, `snapshots`, `tests` | +| `--downstream-depth` | | `DBT_DOWNSTREAM_DEPTH` | Full graph | How many levels of downstream dependencies to include (dbt's `model+N`) | | `--filters` | `-f` | | `None` | Extra resource-type filter (repeatable, choices: `models`, `seeds`, `snapshots`, `tests`). E.g. `--mode tests -f snapshots` to run only tests that have a snapshot dependency | +#### Limiting blast radius + +By default a change selects its **entire** downstream graph. On a large project a change +to a core staging model therefore rebuilds almost everything, which is the opposite of +what change-based CI is for. `--downstream-depth` caps how far a change propagates: + +```bash +dbt-ci run --downstream-depth 0 # only what changed +dbt-ci run --downstream-depth 1 # changed models and their direct children +dbt-ci run --downstream-depth 2 # two levels out +dbt-ci run # unlimited (default) +``` + +For a chain `customers → l1 → l2 → l3` where only `customers` changed: + +| Flag | Models run | +|------|------------| +| `--downstream-depth 0` | `customers` | +| `--downstream-depth 1` | `customers`, `l1` | +| `--downstream-depth 2` | `customers`, `l1`, `l2` | +| *(omitted)* | `customers`, `l1`, `l2`, `l3` | + +New and deleted nodes are always included regardless of depth — a new model has to run +whether or not anything depends on it yet. + > All [common options](#common-options) also apply. **Examples:** @@ -262,6 +288,38 @@ dbt-ci migration # apply the changes --- +### `report` - Summarise the Run + +Renders the change set detected by `init` and the status of every command that has run +so far. In GitHub Actions the report is appended to the job summary automatically, so no +workflow wiring is needed beyond calling it. + +```bash +dbt-ci report # → $GITHUB_STEP_SUMMARY, or stdout locally +dbt-ci report --output report.md # → a file, e.g. to post as a PR comment +dbt-ci report --format json # → machine-readable +``` + +The report covers: + +- **Change counts and node names**, grouped as modified / new / deleted. Long lists fold + into a collapsible block. +- **Exposure impact** — the exposures downstream of the change set, including those + downstream of *deleted* nodes, which is usually the case worth catching. +- **Command status and duration** for each of `init`, `run`, `delete`, `ephemeral` and + `migration` that has run. + +**Flags:** + +| Flag | Aliases | Env Var(s) | Default | Description | +|------|---------|-----------|---------|-------------| +| `--output` | `-o` | `DBT_REPORT_OUTPUT` | `$GITHUB_STEP_SUMMARY` or stdout | Where to write the report | +| `--format` | `-F` | `DBT_REPORT_FORMAT` | `markdown` | `markdown` or `json` | + +> All [common options](#common-options) also apply. + +--- + ### `config` - Generate a Config File Writes a commented `dbt-ci.config.yaml` skeleton with the common options pre-filled, so @@ -460,6 +518,10 @@ init: comparison-strategy: hybrid base-ref: main +run: + nodes: models + downstream-depth: 2 + finalize: artifacts-uri: s3://my-bucket/dbt-artifacts/ files: @@ -642,6 +704,19 @@ dbt-ci run **Note:** State management is cache-based. Run `init` once, then subsequent commands automatically use the cached state. +### Settings Inherited From `init` + +`init` records the `--target` and `--vars` it ran with, and later commands reuse them, so +they only need to be given once: + +```bash +dbt-ci init --target ci --vars '{"use_production_data": false}' --state dbt/.dbtstate +dbt-ci run # runs against target 'ci' with the same vars +dbt-ci delete # likewise +``` + +Passing the flag explicitly still wins — the cache only fills in what was left unset. + ### Cache Location `init` writes its cache (state comparison, manifests, run report and log file) to @@ -729,6 +804,8 @@ dbt-ci: - **💬 Notifications**: Slack webhook integration for CI/CD alerts - **♻️ Ephemeral Environments**: Test changes in isolated environments - **🧹 Cleanup**: Automatically remove deleted models from target warehouse +- **🎯 Blast-Radius Control**: Cap how far a change propagates with `--downstream-depth` +- **📝 Run Reports**: Markdown summary of the change set, exposure impact and command status - **🔀 Partition Migrations**: Rebuild tables whose partitioning configuration changed (BigQuery) ## Use Cases diff --git a/dbt_ci/cli/config/schema.py b/dbt_ci/cli/config/schema.py index c13af99..b1059db 100644 --- a/dbt_ci/cli/config/schema.py +++ b/dbt_ci/cli/config/schema.py @@ -53,6 +53,7 @@ "type": "section", "fields": { "nodes": {"type": "enum", "choices": ["all", "models", "seeds", "snapshots", "tests"]}, + "downstream-depth": {"aliases": ["downstream_depth"], "type": "int"}, }, }, "finalize": { @@ -95,6 +96,10 @@ def _validate_value(path: str, value: Any, rule: dict, errors: list[str]) -> Non elif kind == "bool": if not isinstance(value, bool): errors.append(f"'{path}' must be a boolean (true/false), got {value!r}") + elif kind == "int": + # bool is a subclass of int, so reject it explicitly. + if isinstance(value, bool) or not isinstance(value, int): + errors.append(f"'{path}' must be an integer, got {value!r}") elif kind == "enum": choices = rule["choices"] if isinstance(value, str): diff --git a/dbt_ci/commands/delete/index.py b/dbt_ci/commands/delete/index.py index 6c72e90..6432d76 100644 --- a/dbt_ci/commands/delete/index.py +++ b/dbt_ci/commands/delete/index.py @@ -8,6 +8,7 @@ from argparse import Namespace import click from dbt_ci.utilities.cache import CacheManager +from dbt_ci.dbt.flags import apply_cached_config from dbt_ci.utilities.logging import redact_namespace from dbt_ci.connectors import get_connector from dbt_ci.graph.dependency_graph import DbtGraph @@ -24,6 +25,7 @@ def delete(args: Namespace): logger.debug(f"Running with the following arguments: {redact_namespace(args)}") cache = CacheManager(args) cache.start_report("delete", args) + apply_cached_config(args) connector_type = cast(SupportedConnectors, get_profile(args)["type"]) delete_connector = get_connector(connector_type) delete_map = generate_delete_map(args, cache) diff --git a/dbt_ci/commands/ephemeral/index.py b/dbt_ci/commands/ephemeral/index.py index 39ba49f..d9db689 100644 --- a/dbt_ci/commands/ephemeral/index.py +++ b/dbt_ci/commands/ephemeral/index.py @@ -13,6 +13,7 @@ from typing import Optional, cast import click from dbt_ci.utilities.cache import CacheManager +from dbt_ci.dbt.flags import apply_cached_config from dbt_ci.connectors import DB_CONNECTORS from dbt_ci.graph.dependency_graph import DbtGraph from dbt_ci.utilities.logging import print_exception, redact_namespace @@ -56,6 +57,7 @@ def ephemeral(args: Namespace): ) cache = CacheManager(args) cache.start_report("ephemeral", args) + apply_cached_config(args) cache_dict: StateChangeSummary = cast(StateChangeSummary, cache.get_cache()) if cache_dict is None: diff --git a/dbt_ci/commands/migration/index.py b/dbt_ci/commands/migration/index.py index cd9e302..95f820e 100644 --- a/dbt_ci/commands/migration/index.py +++ b/dbt_ci/commands/migration/index.py @@ -5,6 +5,7 @@ from typing import cast import click from dbt_ci.utilities.cache import CacheManager +from dbt_ci.dbt.flags import apply_cached_config from dbt_ci.graph.dependency_graph import DbtGraph from dbt_ci.utilities.logging import print_exception, redact_namespace from dbt_ci.schema import MigrationMap, SupportedConnectors @@ -33,6 +34,7 @@ def migration(args: Namespace): logger.debug(f"Running with the following arguments: {redact_namespace(args)}") cache = CacheManager(args) cache.start_report("migrate", args) + apply_cached_config(args) connector_type = cast(SupportedConnectors, get_profile(args)["type"]) connector = get_connector(connector_type) diff --git a/dbt_ci/commands/report/__init__.py b/dbt_ci/commands/report/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/dbt_ci/commands/report/cli.py b/dbt_ci/commands/report/cli.py new file mode 100644 index 0000000..c387886 --- /dev/null +++ b/dbt_ci/commands/report/cli.py @@ -0,0 +1,48 @@ +import click +from dbt_ci.main import cli +from dbt_ci.cli.common_options import common_options +from dbt_ci.cli.config import make_config_callback +from dbt_ci.cli.namespace import to_namespace +from dbt_ci.utilities.logging import setup_logging +from dbt_ci.commands.report.index import report + + +@cli.command(name="report") +@common_options +@click.option( + "--output", "-o", + envvar=["DBT_REPORT_OUTPUT"], + default=None, + type=str, + callback=make_config_callback("DBT_REPORT_OUTPUT"), + help=( + "Where to write the report. Defaults to $GITHUB_STEP_SUMMARY when set, " + "otherwise stdout." + ), +) +@click.option( + "--format", "-F", "format", + envvar=["DBT_REPORT_FORMAT"], + type=click.Choice(["markdown", "json"], case_sensitive=False), + default="markdown", + callback=make_config_callback("DBT_REPORT_FORMAT", then=str.lower), + help="Output format (default: markdown).", +) +def report_cmd(**kwargs): + """Summarise the current dbt-ci run + + Renders the change set from 'init' and the status of each command that has run so + far. In GitHub Actions the report is appended to the job summary automatically. + + Examples: + # Append to the GitHub Actions job summary + dbt-ci report + + # Write markdown to a file, e.g. to post as a pull request comment + dbt-ci report --output report.md + + # Machine-readable output + dbt-ci report --format json + """ + setup_logging(to_namespace(kwargs).log_level) + return report(to_namespace(kwargs, command="report")) diff --git a/dbt_ci/commands/report/index.py b/dbt_ci/commands/report/index.py new file mode 100644 index 0000000..e8f9055 --- /dev/null +++ b/dbt_ci/commands/report/index.py @@ -0,0 +1,250 @@ +"""Render the run report that init/run/delete/ephemeral/migration already record. + +Every command writes its status, timings and resolved variables into report.json, and +the change set into cache.json, but nothing ever read them back - the files were written +and then deleted by finalize. This turns them into something a reviewer sees. +""" +import json +import logging +import os +import sys +from argparse import Namespace +from datetime import datetime +from typing import Any, cast + +import click + +from dbt_ci.graph.dependency_graph import DbtGraph +from dbt_ci.graph.graph_utils import ( + get_display_name, + get_downstream_dependencies, + get_node_ids_from_structured_nodes, +) +from dbt_ci.schema import DependencyGraph, StateChangeSummary +from dbt_ci.utilities.cache import CacheManager +from dbt_ci.utilities.logging import print_exception + +logger = logging.getLogger(__name__) + +CHANGE_TYPES: dict[str, str] = { + "modified_nodes": "Modified", + "new_nodes": "New", + "deleted_nodes": "Deleted", +} + +# Beyond this many nodes of one kind, the list is collapsed behind a
element so +# the summary stays readable on a large change set. +COLLAPSE_THRESHOLD = 10 + + +def report(args: Namespace) -> None: + """Render the cached run report and write it to the requested destination.""" + try: + cache = CacheManager(args) + cache_dict = cast(StateChangeSummary | None, cache.get_cache()) + run_report = cast(dict[str, Any] | None, cache.get_cache("report.json")) + + if cache_dict is None and run_report is None: + logger.error( + "No cache found, please run 'dbt-ci init' first to generate the state and " + "run report this command renders." + ) + sys.exit(1) + + output_format = getattr(args, "format", "markdown") + if output_format == "json": + rendered = json.dumps( + {"report": run_report, "changes": cache_dict}, + indent=2, + default=lambda o: list(o) if isinstance(o, set) else str(o), + ) + else: + rendered = render_markdown(cache_dict, run_report, args) + + write_report(rendered, args) + except Exception as e: + print_exception(e, "Error generating report") + sys.exit(1) + + +def write_report(rendered: str, args: Namespace) -> None: + """ + Write the rendered report to --output, GITHUB_STEP_SUMMARY, or stdout. + + Appending to GITHUB_STEP_SUMMARY by default means the report shows up on the job + page in GitHub Actions without the workflow needing to do anything. + """ + destination = getattr(args, "output", None) or os.environ.get("GITHUB_STEP_SUMMARY") + + if not destination: + click.echo(rendered) + return + + # GITHUB_STEP_SUMMARY accumulates across steps, so append rather than truncate. + mode = "a" if destination == os.environ.get("GITHUB_STEP_SUMMARY") else "w" + with open(destination, mode, encoding="utf-8") as f: + f.write(rendered + "\n") + logger.info(f"Report written to {destination}") + + +def render_markdown( + cache_dict: StateChangeSummary | None, + run_report: dict[str, Any] | None, + args: Namespace, +) -> str: + """Render the full report as GitHub-flavoured markdown.""" + lines: list[str] = ["## dbt-ci run report", ""] + lines.extend(render_change_summary(cache_dict)) + lines.extend(render_exposure_impact(cache_dict, args)) + lines.extend(render_command_status(run_report)) + return "\n".join(lines).rstrip() + "\n" + + +def render_change_summary(cache_dict: StateChangeSummary | None) -> list[str]: + """Render the counts and node lists for modified, new and deleted resources.""" + if not cache_dict: + return ["No state comparison found.", ""] + + counts: dict[str, int] = {} + for change_key in CHANGE_TYPES: + structured = cache_dict.get(change_key) or {} + counts[change_key] = sum(len(nodes) for nodes in structured.values()) + + if not any(counts.values()): + return ["No changes detected against the reference state.", ""] + + lines = ["| Change | Count |", "|---|---:|"] + for change_key, label in CHANGE_TYPES.items(): + lines.append(f"| {label} | {counts[change_key]} |") + lines.append("") + + for change_key, label in CHANGE_TYPES.items(): + structured = cache_dict.get(change_key) or {} + if not structured: + continue + entries: list[str] = [] + for resource_type, nodes in sorted(structured.items()): + for node in nodes.values(): + entries.append(f"- `{node.get('name', '?')}` ({resource_type})") + lines.extend(collapsible(f"{label} ({len(entries)})", entries)) + + return lines + + +def render_exposure_impact(cache_dict: StateChangeSummary | None, args: Namespace) -> list[str]: + """ + List exposures downstream of the change set. + + Exposures are what the warehouse is for - dashboards and downstream consumers - so + naming the ones a pull request touches is the most useful single line in the report. + """ + if not cache_dict: + return [] + + entries: set[str] = set() + + # Modified and new nodes are resolved against the target graph; deleted nodes only + # exist in the reference graph, and an exposure downstream of a deleted model is + # precisely the case worth surfacing. + for change_keys, is_reference in ((("modified_nodes", "new_nodes"), False), (("deleted_nodes",), True)): + node_ids: list[str] = [] + for change_key in change_keys: + node_ids.extend(get_node_ids_from_structured_nodes(cache_dict.get(change_key)) or []) + if not node_ids: + continue + + graph = load_graph(args, is_reference=is_reference) + if graph is None: + continue + + exposures = get_downstream_dependencies( + dependency_graph=graph, + node_ids=node_ids, + node_type="exposure", + ) + for exposure_id in exposures or set(): + entries.add(get_display_name(graph, exposure_id)) + + if not entries: + return [] + + listed = sorted(f"- `{name}`" for name in entries) + return [f"### ⚠️ Affects {len(listed)} exposure(s)", "", *listed, ""] + + +def load_graph(args: Namespace, is_reference: bool = False) -> DependencyGraph | None: + """Load a dependency graph from cache, returning None when unavailable.""" + try: + return DbtGraph(args, is_reference=is_reference).to_dict() + except Exception as e: + logger.debug(f"Could not load the dependency graph for exposure impact: {e}") + return None + + +def render_command_status(run_report: dict[str, Any] | None) -> list[str]: + """Render each command's status and how long it took.""" + if not run_report: + return [] + + lines = ["### Commands", "", "| Command | Status | Duration |", "|---|---|---:|"] + for command, entry in run_report.items(): + if not isinstance(entry, dict): + continue + status = entry.get("status", "unknown") + lines.append( + f"| `{command}` | {status_icon(status)} {status} | {format_duration(entry)} |" + ) + lines.append("") + return lines + + +def status_icon(status: str) -> str: + """Return an icon for a command status.""" + return {"completed": "✅", "failed": "❌", "started": "⏳"}.get(status, "•") + + +def format_duration(entry: dict[str, Any]) -> str: + """ + Return how long a command took, based on the timestamps it recorded. + + The end key is named after the terminal status (completed_at, failed_at), so it is + looked up from the status rather than assumed. + """ + status = entry.get("status") + end_key = f"{status}_at" + if not status or end_key == "started_at": + # Still running: the status has no end timestamp of its own, and reusing + # started_at would report a duration of zero. + return "-" + + started_at = entry.get("started_at") + finished_at = entry.get(end_key) + if not started_at or not finished_at: + return "-" + + try: + delta = datetime.fromisoformat(finished_at) - datetime.fromisoformat(started_at) + except (TypeError, ValueError): + return "-" + + seconds = delta.total_seconds() + if seconds < 60: + return f"{seconds:.1f}s" + return f"{int(seconds // 60)}m {int(seconds % 60)}s" + + +def collapsible(summary: str, entries: list[str]) -> list[str]: + """Return entries inline, or folded into a
block when there are many.""" + if not entries: + return [] + if len(entries) <= COLLAPSE_THRESHOLD: + return [f"**{summary}**", "", *entries, ""] + return [ + "
", + f"{summary}", + "", + *entries, + "", + "
", + "", + ] diff --git a/dbt_ci/commands/run/cli.py b/dbt_ci/commands/run/cli.py index b1f6f64..8069233 100644 --- a/dbt_ci/commands/run/cli.py +++ b/dbt_ci/commands/run/cli.py @@ -24,6 +24,17 @@ callback=make_config_callback("DBT_RUN_NODES"), help='Run mode for dbt-ci (default: all)' ) +@click.option( + '--downstream-depth', + envvar=['DBT_DOWNSTREAM_DEPTH'], + type=int, + default=None, + callback=make_config_callback("DBT_RUN_DOWNSTREAM_DEPTH"), + help=( + 'How many levels of downstream dependencies to include (dbt\'s model+N). ' + 'Omit for the full downstream graph; 0 runs only what changed.' + ) +) @click.option( '--filters', '-f', type=click.Choice([ diff --git a/dbt_ci/commands/run/index.py b/dbt_ci/commands/run/index.py index 64df3b0..5727dab 100644 --- a/dbt_ci/commands/run/index.py +++ b/dbt_ci/commands/run/index.py @@ -7,6 +7,7 @@ from dbt_ci.commands.run.run import run_nodes from dbt_ci.graph.dependency_graph import DbtGraph from dbt_ci.utilities.cache import CacheManager +from dbt_ci.dbt.flags import apply_cached_config from dbt_ci.utilities.logging import print_exception, redact_namespace from dbt_ci.graph.graph_utils import get_node_ids_from_structured_nodes @@ -33,6 +34,7 @@ def run(args: Namespace): logger.debug(f"Running with the following arguments: {redact_namespace(args)}") cache = CacheManager(args) cache.start_report("run", args) + apply_cached_config(args) target_graph = DbtGraph(args) # Look for cache diff --git a/dbt_ci/commands/run/run.py b/dbt_ci/commands/run/run.py index 63da68d..f8c4cd3 100644 --- a/dbt_ci/commands/run/run.py +++ b/dbt_ci/commands/run/run.py @@ -87,6 +87,23 @@ def run_nodes( logger.error(f"\n❌ Error running dbt command: {e}") sys.exit(1) +def get_downstream_depth(args: Namespace) -> int | None: + """ + Return the configured downstream selection depth, or None for the full closure. + + A change to a central staging model otherwise selects every descendant, which on a + real project means rebuilding almost everything - the opposite of what a + change-based CI tool is for. Depth 0 runs only what changed. + """ + depth = getattr(args, "downstream_depth", None) + if depth is None: + return None + if depth < 0: + logger.warning(f"Ignoring negative --downstream-depth ({depth}); selecting the full downstream graph.") + return None + return depth + + def seeds( target_graph: DbtGraph, changed_nodes_dict: dict[str, list[str]], @@ -130,7 +147,8 @@ def tests( get_downstream_dependencies( dependency_graph=target_graph.to_dict(), node_ids=changed_nodes, - node_type="test" + node_type="test", + levels=get_downstream_depth(args) ) or [] ) @@ -201,7 +219,8 @@ def snapshots( get_downstream_dependencies( dependency_graph=target_graph.to_dict(), node_ids=changed_nodes, - node_type="snapshot" + node_type="snapshot", + levels=get_downstream_depth(args) ) or [] ) @@ -227,7 +246,8 @@ def models( get_downstream_dependencies( dependency_graph=target_graph.to_dict(), node_ids=changed_nodes, - node_type="model" + node_type="model", + levels=get_downstream_depth(args) ) or [] ) diff --git a/dbt_ci/dbt/flags.py b/dbt_ci/dbt/flags.py index f90ca9b..30436c2 100644 --- a/dbt_ci/dbt/flags.py +++ b/dbt_ci/dbt/flags.py @@ -1,55 +1,77 @@ +"""Resolution of dbt invocation settings that `init` recorded into the cache. + +`init` writes the target, vars, runner and comparison strategy it ran with into +cache.json, and the README promises you "specify state once in init, reuse everywhere". +That was only ever true for state paths: every later command re-read target and vars +from the CLI, so they had to be repeated. These helpers close that gap - an explicitly +provided value still wins, the cache only fills in what was left unset. +""" +import logging from argparse import Namespace -from typing import Literal, cast +from typing import Any, Literal, cast +from dbt_ci.schema import DbtCiManifest from dbt_ci.utilities.cache import CacheManager -from dbt_ci.schema import DbtCiManifest, DependencyGraphNodeType - -dependency_graph_node_type: list[DependencyGraphNodeType] = [ - "model", - "macro", - "source", - "seed", - "snapshot", - "test", - "exposure" -] - -def exclude_flag(include: list[DependencyGraphNodeType] | None) -> list[str]: - """Generate a list of node types to exclude based on the include list.""" - default_list = { - "model": "--exclude resource_type:model", - "macro": "--exclude resource_type:macro", - "source": "--exclude resource_type:source", - "seed": "--exclude resource_type:seed", - "snapshot": "--exclude resource_type:snapshot", - "test": "--exclude resource_type:test", - "exposure": "--exclude resource_type:exposure" - } - - - exclude_flags: set[str] = set() - - if include is None: - exclude_flags = set(default_list.values()) - else: - for node_type in dependency_graph_node_type: - if node_type not in include: - exclude_flags.add(default_list[node_type]) - - return list(exclude_flags) - -def get_target(args: Namespace, target: Literal["target", "reference_target"]) -> str | None: - """Get the target from the Namespace, checking both 'target' and 'reference_target' keys.""" - cache = CacheManager(args) - cache_manifest = cast(DbtCiManifest, cache.get_cache()) - - cache_target = cache_manifest.get("config", {}).get(target, {}).get("target", None) - return getattr(args, target, cache_target) - -def get_vars(args: Namespace, target: Literal["vars", "reference_vars"]) -> dict | None: - """Get the vars from the Namespace, checking both 'vars' and 'reference_vars' keys.""" - cache = CacheManager(args) - cache_manifest = cast(DbtCiManifest, cache.get_cache()) - - cache_vars = cache_manifest.get("config", {}).get(target, {}).get("vars", None) - return getattr(args, target, cache_vars) \ No newline at end of file + +logger = logging.getLogger(__name__) + +# Settings recorded by init that later commands can inherit, mapped to where they live +# in the cached config block. +type CachedTargetKey = Literal["target", "reference_target"] +type CachedVarsKey = Literal["vars", "reference_vars"] + +CONFIG_SECTION: dict[str, str] = { + "target": "target", + "vars": "target", + "reference_target": "reference", + "reference_vars": "reference", +} + + +def get_cached_config(args: Namespace) -> dict[str, Any]: + """Return the config block init recorded, or an empty mapping when unavailable.""" + cache_manifest = cast(DbtCiManifest | None, CacheManager(args).get_cache()) + if not cache_manifest: + return {} + return cache_manifest.get("config") or {} + + +def get_cached_value(args: Namespace, key: str) -> Any: + """Look a single recorded setting up out of the cached config block.""" + section_name = CONFIG_SECTION.get(key) + if section_name is None: + return None + + section = get_cached_config(args).get(section_name) or {} + # Within a section the keys are unprefixed: reference_target -> reference.target + field = key.replace("reference_", "") + return section.get(field) + + +def get_target(args: Namespace, target: CachedTargetKey = "target") -> str | None: + """Return the dbt target, preferring an explicit value over the cached one.""" + return getattr(args, target, None) or get_cached_value(args, target) + + +def get_vars(args: Namespace, key: CachedVarsKey = "vars") -> str | None: + """Return the dbt vars, preferring an explicit value over the cached one.""" + return getattr(args, key, None) or get_cached_value(args, key) + + +def apply_cached_config(args: Namespace) -> Namespace: + """ + Fill unset target/vars on the args namespace from what init recorded. + + Mutates and returns the namespace so it can be applied at the top of a command. + Only unset values are filled, so a flag passed on the command line still wins. + """ + for key in ("target", "vars"): + current = getattr(args, key, None) + if current: + continue + cached = get_cached_value(args, key) + if cached: + logger.debug(f"Using {key} recorded by init: {cached}") + setattr(args, key, cached) + + return args diff --git a/dbt_ci/graph/graph_utils.py b/dbt_ci/graph/graph_utils.py index e46f879..44be95e 100644 --- a/dbt_ci/graph/graph_utils.py +++ b/dbt_ci/graph/graph_utils.py @@ -1,6 +1,6 @@ """Getters for dependency graph nodes""" import logging -from typing import cast +from typing import Literal, cast from dbt_ci.schema import DependencyGraph, DependencyGraphNode, DependencyGraphNodeType logger = logging.getLogger(__name__) @@ -120,31 +120,73 @@ def get_node_ids_from_structured_nodes(structured_nodes: dict[str, DependencyGra return list(node_ids) if len(node_ids) > 0 else None +def collect_dependencies_to_depth( + dependency_graph: DependencyGraph, + node_ids: list[str], + direction: Literal["upstream", "downstream"] = "downstream", + depth: int = 1, +) -> set[str]: + """ + Walk `depth` hops out from the given nodes, following direct edges only. + + Breadth-first rather than reusing the precomputed closure, because the closure has + no notion of distance. A depth of 0 selects nothing beyond the starting nodes, which + is how a caller asks for "just what changed". + """ + key = f"{direction}_dependencies" + reached: set[str] = set() + frontier: list[str] = list(node_ids) + + for _ in range(max(depth, 0)): + next_frontier: list[str] = [] + for node_id in frontier: + node = get_node(dependency_graph, node_id) + if node is None: + continue + for dep_id in node[key]["node_dependencies"]: + if dep_id in reached: + continue + reached.add(dep_id) + next_frontier.append(dep_id) + if not next_frontier: + break + frontier = next_frontier + + return reached + def get_downstream_dependencies( dependency_graph: DependencyGraph, node_ids: list[str] | None, node_type: DependencyGraphNodeType | None = None, - levels: int | None = None # To be implemented in the future + levels: int | None = None ) -> set[str] | None: """ - Get downstream dependencies for a list of node IDs, - optionally up to a certain number of levels. Defaults to indirect downstream dependencies - if levels is not specified. + Get downstream dependencies for a list of node IDs, + optionally up to a certain number of levels. Defaults to the full downstream + closure when levels is not specified. """ - key = "indirect_downstream_dependencies" if node_ids is None or len(node_ids) == 0: return None - if levels is not None: - key = f"downstream_dependencies_level_{levels}" - downstream_dependencies: set[str] = set() + if levels is not None: + downstream_dependencies = collect_dependencies_to_depth( + dependency_graph, node_ids, "downstream", levels + ) + if node_type: + downstream_dependencies = { + dep_id for dep_id in downstream_dependencies + if (node := get_node(dependency_graph, dep_id)) and node.get("resource_type") == node_type + } + return downstream_dependencies or None + + downstream_dependencies = set() for node_id in node_ids: node = get_node(dependency_graph, node_id) if node is None: logger.warning(f"Node ID '{node_id}' not found in dependency graph when attempting to get downstream dependencies.") continue - dependency_by_type: dict[DependencyGraphNodeType, list[str]] = node[key]["dependencies_by_type"] + dependency_by_type: dict[DependencyGraphNodeType, list[str]] = node["indirect_downstream_dependencies"]["dependencies_by_type"] for dep_type, dep_names in dependency_by_type.items(): if node_type and dep_type != node_type: continue @@ -160,13 +202,25 @@ def get_upstream_dependencies( node_ids: list[str] | None, node_type: list[DependencyGraphNodeType] | None = None, filters: list[DependencyGraphNodeType] | None = None, - levels: int | None = None # To be implemented in the future + levels: int | None = None ) -> set[str] | None: """Get upstream dependencies for a list of node IDs, optionally up to a certain number of levels.""" if node_ids is None or len(node_ids) == 0: return None - upstream_dependencies: set[str] = set() + if levels is not None: + upstream_dependencies = collect_dependencies_to_depth( + dependency_graph, node_ids, "upstream", levels + ) + allowed = set(node_type or []) | set(filters or []) + if allowed: + upstream_dependencies = { + dep_id for dep_id in upstream_dependencies + if (node := get_node(dependency_graph, dep_id)) and node.get("resource_type") in allowed + } + return upstream_dependencies or None + + upstream_dependencies = set() for node_id in node_ids: node = get_node(dependency_graph, node_id) if node is None: diff --git a/dbt_ci/main.py b/dbt_ci/main.py index 2cd6f3e..bdb15bb 100644 --- a/dbt_ci/main.py +++ b/dbt_ci/main.py @@ -20,6 +20,7 @@ def cli(): import dbt_ci.commands.migration.cli # noqa: E402, F401 import dbt_ci.commands.finalize.cli # noqa: E402, F401 import dbt_ci.commands.config.cli # noqa: E402, F401 +import dbt_ci.commands.report.cli # noqa: E402, F401 if __name__ == "__main__": cli() \ No newline at end of file diff --git a/tests/unit/commands/test_report.py b/tests/unit/commands/test_report.py new file mode 100644 index 0000000..e6fe99f --- /dev/null +++ b/tests/unit/commands/test_report.py @@ -0,0 +1,157 @@ +"""Unit tests for the report command.""" +from argparse import Namespace +from unittest.mock import patch + +from dbt_ci.commands.report.index import ( + format_duration, + render_change_summary, + render_command_status, + render_markdown, + write_report, +) + + +def _structured(resource_type: str, *names: str) -> dict: + """Build a structured-nodes mapping as init writes it into the cache.""" + return { + resource_type: { + f"{resource_type}.p.{name}": { + "id": f"{resource_type}.p.{name}", + "name": name, + "resource_type": resource_type, + } + for name in names + } + } + + +CACHE = { + "modified_nodes": _structured("model", "customers"), + "new_nodes": _structured("model", "orders"), + "deleted_nodes": _structured("model", "legacy"), +} + +RUN_REPORT = { + "init": { + "status": "completed", + "started_at": "2026-01-01T00:00:00", + "completed_at": "2026-01-01T00:00:07.500000", + }, + "run": { + "status": "failed", + "started_at": "2026-01-01T00:01:00", + "failed_at": "2026-01-01T00:03:30", + }, +} + + +class TestRenderChangeSummary: + """Test the change counts and node listings.""" + + def test_counts_each_change_type(self): + """The summary table reports one row per change type.""" + rendered = "\n".join(render_change_summary(CACHE)) + assert "| Modified | 1 |" in rendered + assert "| New | 1 |" in rendered + assert "| Deleted | 1 |" in rendered + + def test_lists_node_names(self): + """Each changed node is named with its resource type.""" + rendered = "\n".join(render_change_summary(CACHE)) + assert "`customers` (model)" in rendered + assert "`legacy` (model)" in rendered + + def test_reports_no_changes(self): + """An empty change set says so rather than rendering empty tables.""" + empty = {"modified_nodes": None, "new_nodes": None, "deleted_nodes": None} + assert "No changes detected" in "\n".join(render_change_summary(empty)) + + def test_missing_cache(self): + """A missing state comparison is reported, not crashed on.""" + assert "No state comparison found." in "\n".join(render_change_summary(None)) + + def test_long_lists_are_collapsed(self): + """Large change sets fold into
so the summary stays readable.""" + many = {"modified_nodes": _structured("model", *[f"m{i}" for i in range(25)])} + rendered = "\n".join(render_change_summary(many)) + assert "
" in rendered + assert "Modified (25)" in rendered + + +class TestRenderCommandStatus: + """Test the per-command status table.""" + + def test_lists_commands_with_icons(self): + """Each command's terminal status is shown.""" + rendered = "\n".join(render_command_status(RUN_REPORT)) + assert "| `init` | ✅ completed |" in rendered + assert "| `run` | ❌ failed |" in rendered + + def test_no_report_renders_nothing(self): + """Without a report there is no command section.""" + assert render_command_status(None) == [] + + +class TestFormatDuration: + """Test duration formatting from the recorded timestamps.""" + + def test_seconds(self): + """Sub-minute durations render in seconds.""" + assert format_duration(RUN_REPORT["init"]) == "7.5s" + + def test_minutes(self): + """Longer durations render as minutes and seconds.""" + assert format_duration(RUN_REPORT["run"]) == "2m 30s" + + def test_end_key_follows_the_status(self): + """The end timestamp key is named after the status, not assumed to be completed_at.""" + entry = {"status": "failed", "started_at": "2026-01-01T00:00:00", "failed_at": "2026-01-01T00:00:05"} + assert format_duration(entry) == "5.0s" + + def test_missing_timestamps(self): + """A command still running has no duration to report.""" + assert format_duration({"status": "started", "started_at": "2026-01-01T00:00:00"}) == "-" + + def test_unparseable_timestamps(self): + """Malformed timestamps degrade rather than raising.""" + assert format_duration({"status": "completed", "started_at": "nope", "completed_at": "nope"}) == "-" + + +class TestWriteReport: + """Test where the rendered report is written.""" + + def test_writes_to_explicit_output(self, tmp_path): + """--output takes precedence and truncates the file.""" + target = tmp_path / "report.md" + write_report("# hello", Namespace(output=str(target))) + assert target.read_text(encoding="utf-8").strip() == "# hello" + + def test_appends_to_github_step_summary(self, tmp_path, monkeypatch): + """The job summary accumulates across steps, so it must be appended to.""" + summary = tmp_path / "summary.md" + summary.write_text("existing\n", encoding="utf-8") + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary)) + + write_report("# added", Namespace(output=None)) + + contents = summary.read_text(encoding="utf-8") + assert "existing" in contents and "# added" in contents + + def test_falls_back_to_stdout(self, capsys, monkeypatch): + """Outside CI the report goes to stdout.""" + monkeypatch.delenv("GITHUB_STEP_SUMMARY", raising=False) + write_report("# stdout", Namespace(output=None)) + assert "# stdout" in capsys.readouterr().out + + +class TestRenderMarkdown: + """Test the assembled document.""" + + def test_includes_each_section(self): + """The report carries a heading, the change summary and the command table.""" + with patch("dbt_ci.commands.report.index.render_exposure_impact", return_value=[]): + rendered = render_markdown(CACHE, RUN_REPORT, Namespace()) + + assert rendered.startswith("## dbt-ci run report") + assert "| Modified | 1 |" in rendered + assert "### Commands" in rendered diff --git a/tests/unit/graph/test_depth.py b/tests/unit/graph/test_depth.py new file mode 100644 index 0000000..b0e41be --- /dev/null +++ b/tests/unit/graph/test_depth.py @@ -0,0 +1,121 @@ +"""Unit tests for depth-limited dependency selection.""" +from dbt_ci.graph.graph_utils import ( + collect_dependencies_to_depth, + get_downstream_dependencies, + get_upstream_dependencies, +) + + +def _node(unique_id: str, downstream: list[str], upstream: list[str]) -> dict: + """Build a graph node carrying only the edges the traversal reads.""" + return { + "id": unique_id, + "name": unique_id.split(".")[-1], + "resource_type": unique_id.split(".")[0], + "downstream_dependencies": {"node_dependencies": set(downstream), "dependencies_by_type": {}}, + "upstream_dependencies": {"node_dependencies": set(upstream), "dependencies_by_type": {}}, + } + + +def _chain_graph() -> dict: + """a -> b -> c -> d, with a second branch a -> e, and a test hanging off b.""" + edges = { + "model.p.a": (["model.p.b", "model.p.e"], []), + "model.p.b": (["model.p.c", "test.p.b_is_valid"], ["model.p.a"]), + "model.p.c": (["model.p.d"], ["model.p.b"]), + "model.p.d": ([], ["model.p.c"]), + "model.p.e": ([], ["model.p.a"]), + "test.p.b_is_valid": ([], ["model.p.b"]), + } + graph: dict = {"metadata": {}, "model": {}, "test": {}} + for unique_id, (down, up) in edges.items(): + graph[unique_id.split(".")[0]][unique_id] = _node(unique_id, down, up) + return graph + + +class TestCollectDependenciesToDepth: + """Test the bounded breadth-first traversal.""" + + def test_depth_zero_selects_nothing(self): + """Depth 0 means "only what changed" - no dependencies at all.""" + assert collect_dependencies_to_depth(_chain_graph(), ["model.p.a"], "downstream", 0) == set() + + def test_depth_one_selects_direct_children_only(self): + """One hop reaches both branches but goes no further.""" + result = collect_dependencies_to_depth(_chain_graph(), ["model.p.a"], "downstream", 1) + assert result == {"model.p.b", "model.p.e"} + + def test_depth_accumulates_each_level(self): + """Each additional level adds the next ring, keeping the earlier ones.""" + graph = _chain_graph() + two = collect_dependencies_to_depth(graph, ["model.p.a"], "downstream", 2) + assert two == {"model.p.b", "model.p.e", "model.p.c", "test.p.b_is_valid"} + + def test_depth_beyond_the_graph_is_the_full_closure(self): + """Asking for more levels than exist simply stops early.""" + graph = _chain_graph() + deep = collect_dependencies_to_depth(graph, ["model.p.a"], "downstream", 99) + assert deep == { + "model.p.b", "model.p.c", "model.p.d", "model.p.e", "test.p.b_is_valid", + } + + def test_upstream_direction(self): + """The same traversal walks ancestors when asked.""" + result = collect_dependencies_to_depth(_chain_graph(), ["model.p.d"], "upstream", 2) + assert result == {"model.p.c", "model.p.b"} + + def test_unknown_start_node_is_skipped(self): + """A node id absent from the graph contributes nothing and does not raise.""" + assert collect_dependencies_to_depth(_chain_graph(), ["model.p.ghost"], "downstream", 3) == set() + + def test_cycle_terminates(self): + """A cyclic graph must not loop forever.""" + graph = { + "metadata": {}, + "model": { + "model.p.a": _node("model.p.a", ["model.p.b"], ["model.p.b"]), + "model.p.b": _node("model.p.b", ["model.p.a"], ["model.p.a"]), + }, + } + assert collect_dependencies_to_depth(graph, ["model.p.a"], "downstream", 99) == { + "model.p.a", "model.p.b", + } + + +class TestDepthThroughPublicHelpers: + """Test that levels= is honoured by the selection helpers the commands call.""" + + def test_downstream_levels_limits_selection(self): + """levels caps how far a change propagates.""" + graph = _chain_graph() + assert get_downstream_dependencies(graph, ["model.p.a"], levels=1) == { + "model.p.b", "model.p.e", + } + + def test_downstream_levels_zero_returns_none(self): + """Depth 0 yields no dependencies, reported as None like an empty result.""" + assert get_downstream_dependencies(_chain_graph(), ["model.p.a"], levels=0) is None + + def test_downstream_levels_respects_node_type(self): + """Type filtering still applies within the depth limit.""" + graph = _chain_graph() + result = get_downstream_dependencies(graph, ["model.p.a"], node_type="test", levels=2) + assert result == {"test.p.b_is_valid"} + + def test_omitting_levels_uses_the_full_closure(self): + """Without levels the precomputed closure is used, preserving prior behaviour.""" + graph = _chain_graph() + # The closure lives in indirect_*; provide it for the node under test. + graph["model"]["model.p.a"]["indirect_downstream_dependencies"] = { + "node_dependencies": {"model.p.b", "model.p.c", "model.p.d", "model.p.e"}, + "dependencies_by_type": {"model": {"model.p.b", "model.p.c", "model.p.d", "model.p.e"}}, + } + assert get_downstream_dependencies(graph, ["model.p.a"]) == { + "model.p.b", "model.p.c", "model.p.d", "model.p.e", + } + + def test_upstream_levels_filters_by_type(self): + """Upstream depth honours the node_type filter too.""" + graph = _chain_graph() + result = get_upstream_dependencies(graph, ["model.p.d"], node_type=["model"], levels=1) + assert result == {"model.p.c"} From c43dee78972fe2b9fd6b906bd9e5117e6ddc7400 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 9 Aug 2026 10:40:45 +0000 Subject: [PATCH 3/3] fix: stop .gitignore silently excluding directories named dbt The ignore rule was an unanchored `dbt`, which matches a directory of that name at any depth rather than the dbt project at the repository root it was meant to exclude. Two `!dbt_ci/**/dbt` negations (the same line, twice) rescued the package directory, but nothing else: tests/unit/dbt/ was silently dropped by `git add -A`, so a new test module was excluded from a commit without warning and would never have run in CI. Anchoring the pattern to `/dbt/` restricts it to the repository root and makes the negations unnecessary. Verified both directions: a root dbt/ project is still ignored, while tests/unit/dbt/ and new files under dbt_ci/dbt/ are not. Adds the previously-swallowed tests for the cached run configuration. --- .gitignore | 8 +-- tests/unit/dbt/__init__.py | 0 tests/unit/dbt/test_flags.py | 95 ++++++++++++++++++++++++++++++++++++ 3 files changed, 100 insertions(+), 3 deletions(-) create mode 100644 tests/unit/dbt/__init__.py create mode 100644 tests/unit/dbt/test_flags.py diff --git a/.gitignore b/.gitignore index f10b1ad..1dec406 100644 --- a/.gitignore +++ b/.gitignore @@ -206,9 +206,11 @@ marimo/_static/ marimo/_lsp/ __marimo__/ -dbt -!dbt_ci/**/dbt -!dbt_ci/**/dbt +# A dbt project kept at the repository root for local testing. Anchored with a +# leading slash: an unanchored "dbt" matches every directory of that name at any +# depth, which silently excluded dbt_ci/dbt/ and tests/unit/dbt/ and needed +# negation rules to undo. +/dbt/ dependency_graph.json logs dbt.duckdb diff --git a/tests/unit/dbt/__init__.py b/tests/unit/dbt/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/dbt/test_flags.py b/tests/unit/dbt/test_flags.py new file mode 100644 index 0000000..57a1183 --- /dev/null +++ b/tests/unit/dbt/test_flags.py @@ -0,0 +1,95 @@ +"""Unit tests for reusing the run configuration that init recorded.""" +from argparse import Namespace +from unittest.mock import MagicMock, patch + +from dbt_ci.dbt.flags import apply_cached_config, get_cached_value, get_target, get_vars + +CACHED = { + "config": { + "target": {"target": "dev", "vars": "{'k': 1}"}, + "reference": {"target": "production", "vars": "{'k': 2}"}, + } +} + + +def _with_cache(cache_contents): + """Patch CacheManager so the helpers read a known cache payload.""" + manager = MagicMock() + manager.get_cache.return_value = cache_contents + return patch("dbt_ci.dbt.flags.CacheManager", return_value=manager) + + +class TestGetCachedValue: + """Test lookups into the cached config block.""" + + def test_reads_target_section(self): + """target/vars come from the target section.""" + with _with_cache(CACHED): + assert get_cached_value(Namespace(), "target") == "dev" + assert get_cached_value(Namespace(), "vars") == "{'k': 1}" + + def test_reads_reference_section(self): + """reference_target/reference_vars come from the reference section.""" + with _with_cache(CACHED): + assert get_cached_value(Namespace(), "reference_target") == "production" + assert get_cached_value(Namespace(), "reference_vars") == "{'k': 2}" + + def test_unknown_key(self): + """A key with no recorded home resolves to None.""" + with _with_cache(CACHED): + assert get_cached_value(Namespace(), "runner") is None + + def test_missing_cache(self): + """Without a cache nothing is inherited.""" + with _with_cache(None): + assert get_cached_value(Namespace(), "target") is None + + +class TestGetters: + """Test that an explicit value always beats the cached one.""" + + def test_explicit_target_wins(self): + """A target passed on the command line is not overridden by the cache.""" + with _with_cache(CACHED): + assert get_target(Namespace(target="staging")) == "staging" + + def test_falls_back_to_cache(self): + """An unset target is inherited from what init recorded.""" + with _with_cache(CACHED): + assert get_target(Namespace(target=None)) == "dev" + + def test_vars_fall_back_to_cache(self): + """Same for vars.""" + with _with_cache(CACHED): + assert get_vars(Namespace(vars="")) == "{'k': 1}" + + +class TestApplyCachedConfig: + """Test the namespace-filling entry point the commands call.""" + + def test_fills_unset_values(self): + """init's target and vars are applied when the command didn't specify them.""" + args = Namespace(target=None, vars="") + with _with_cache(CACHED): + apply_cached_config(args) + + assert args.target == "dev" + assert args.vars == "{'k': 1}" + + def test_preserves_explicit_values(self): + """Explicitly provided values survive untouched.""" + args = Namespace(target="staging", vars="{'own': true}") + with _with_cache(CACHED): + apply_cached_config(args) + + assert args.target == "staging" + assert args.vars == "{'own': true}" + + def test_no_cache_is_a_noop(self): + """With no cache the namespace is left exactly as it was.""" + args = Namespace(target=None, vars="") + with _with_cache(None): + apply_cached_config(args) + + assert args.target is None + assert args.vars == ""