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/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/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/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_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() 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/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 == "" 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"}