diff --git a/docs/integrations.md b/docs/integrations.md index 1049a28..54fa401 100644 --- a/docs/integrations.md +++ b/docs/integrations.md @@ -96,7 +96,11 @@ dbt-gate --manifest target/manifest.json --threshold 80 --fail ``` It parses dbt's `manifest.json`, reads each model's materialized table via SQLAlchemy, -gates it, and exits non-zero (with `--fail`) if any model is below the threshold. For +gates it, and exits non-zero (with `--fail`) if any model is below the threshold. +Ephemeral and disabled models are not read; they are listed under `"skipped"`. If +no model is gated at all (an empty manifest, or only ephemeral models), `all_passed` +is `false` and `--fail` exits 1. A file that is not a dbt manifest (for example +`run_results.json`) is reported as a one-line error with exit 1. For a single model — or to write per-model `_audit.json` files — use `FreshDataDbtTransform`: diff --git a/src/freshdata/enterprise/cli.py b/src/freshdata/enterprise/cli.py index c285c8c..99774d4 100644 --- a/src/freshdata/enterprise/cli.py +++ b/src/freshdata/enterprise/cli.py @@ -56,6 +56,20 @@ def _add_display_flags(parser: argparse.ArgumentParser) -> None: ) +def _safe_print(text: str) -> None: + """Print *text*, replacing characters stdout's encoding cannot represent. + + Summaries contain non-ASCII (``→``, ``—``). On a cp1252/ascii stdout a plain + ``print`` raises :class:`UnicodeEncodeError` (a ``ValueError``), which ``main`` + would turn into exit 1 *after* outputs were written and the gate decided. + """ + try: + print(text) + except UnicodeEncodeError: + encoding = getattr(sys.stdout, "encoding", None) or "ascii" + print(text.encode(encoding, errors="replace").decode(encoding, errors="replace")) + + def _emit_report(report: Any, args: argparse.Namespace, legacy_text: str) -> None: """Print a clean report honoring the display flags. @@ -68,10 +82,10 @@ def _emit_report(report: Any, args: argparse.Namespace, legacy_text: str) -> Non display = getattr(args, "display", "legacy") if fmt == "json": - print(json.dumps(report.to_dict(), default=str, indent=2)) + _safe_print(json.dumps(report.to_dict(), default=str, indent=2)) return if verbose == 0 and display == "legacy": - print(legacy_text) + _safe_print(legacy_text) return from ..render.normalize import normalize @@ -82,9 +96,9 @@ def _emit_report(report: Any, args: argparse.Namespace, legacy_text: str) -> Non color = "never" if getattr(args, "no_color", False) else "auto" try: options = get_display(mode=mode, color=color) - print(render_terminal_text(normalize(report), options)) + _safe_print(render_terminal_text(normalize(report), options)) except Exception: - print(legacy_text) # display must never break the command + _safe_print(legacy_text) # display must never break the command def _infer_format(path: str) -> str: @@ -120,13 +134,37 @@ def _write_frame( def _load_config_file(path: str) -> dict[str, Any]: + """Load a ``--config`` file; malformed content raises ``ValueError`` naming *path*.""" if path.lower().endswith((".yaml", ".yml")): import yaml with open(path, encoding="utf-8") as fh: - return yaml.safe_load(fh) or {} - with open(path, encoding="utf-8") as fh: - return json.load(fh) + try: + data = yaml.safe_load(fh) or {} + except yaml.YAMLError as exc: + # PyYAML messages span several lines; keep the CLI error to one. + detail = "; ".join(ln.strip() for ln in str(exc).splitlines() if ln.strip()) + raise ValueError(f"invalid YAML in config file {path}: {detail}") from exc + else: + with open(path, encoding="utf-8") as fh: + try: + data = json.load(fh) + except json.JSONDecodeError as exc: + raise ValueError(f"invalid JSON in config file {path}: {exc}") from exc + if not isinstance(data, dict): + raise ValueError( + f"config file {path} must contain a JSON/YAML object, got {type(data).__name__}" + ) + return data + + +def _config_section(data: dict[str, Any], key: str, path: str) -> dict[str, Any]: + section = data.get(key) or {} + if not isinstance(section, dict): + raise ValueError( + f"'{key}' in config file {path} must be an object, got {type(section).__name__}" + ) + return section def _build_enterprise(spec: dict[str, Any]) -> EnterpriseConfig: @@ -184,8 +222,14 @@ def cmd_clean(args: argparse.Namespace) -> int: ec = EnterpriseConfig() if args.config: data = _load_config_file(args.config) - file_clean = data.get("clean", {}) - ec = _build_enterprise(data.get("enterprise", {})) + file_clean = _config_section(data, "clean", args.config) + try: + ec = _build_enterprise(_config_section(data, "enterprise", args.config)) + except TypeError as exc: + # Unknown/misspelled keys (MaskingRule(**rule)) or a non-object entry. + raise ValueError( + f"invalid 'enterprise' section in config file {args.config}: {exc}" + ) from exc overrides: dict[str, Any] = {"strategy": args.strategy} if args.strategy else {} if getattr(args, "drop_duplicates", None): @@ -199,7 +243,11 @@ def cmd_clean(args: argparse.Namespace) -> int: if getattr(args, "strict", False): overrides["strict"] = True merged_clean = {**file_clean, **overrides} - clean_config = merge_options(None, **merged_clean) if merged_clean else None + try: + clean_config = merge_options(None, **merged_clean) if merged_clean else None + except TypeError as exc: # unknown option names, e.g. a typo in the config file + source = f" in config file {args.config}" if args.config else "" + raise ValueError(f"invalid 'clean' options{source}: {exc}") from exc extra_masks = [] for spec in args.mask or []: @@ -254,14 +302,14 @@ def cmd_clean(args: argparse.Namespace) -> int: _emit_report(result.clean_report, args, result.summary()) for event in result.clean_report.fallback_events: if event.get("fallback_step") == "semantic": - print( + _safe_print( f"note: semantic backend '{event.get('backend')}' skipped: " f"{event.get('fallback_reason')}" ) replay = getattr(result.clean_report, "profile_replay", None) if replay is not None and not replay.get("ok"): reasons = replay.get("reasons") or ["severe schema drift"] - print(f"note: learned profile not replayed: {reasons[0]}") + _safe_print(f"note: learned profile not replayed: {reasons[0]}") elif replay is not None and replay.get("severity") == "mild": print("note: learned profile partially replayed (mild schema drift)") return 0 if result.passed_gate else 1 @@ -472,7 +520,12 @@ def cmd_validate(args: argparse.Namespace) -> int: from .contracts import DataContract with open(args.contract, encoding="utf-8") as fh: - suite = ValidationSuite.from_contract(DataContract.from_dict(json.load(fh))) + raw = json.load(fh) + if not isinstance(raw, dict): + raise ValueError( + f"a data contract must be a JSON object, got {type(raw).__name__}" + ) + suite = ValidationSuite.from_contract(DataContract.from_dict(raw)) except FileNotFoundError: raise except (ValueError, KeyError, TypeError, json.JSONDecodeError) as exc: @@ -486,13 +539,13 @@ def cmd_validate(args: argparse.Namespace) -> int: fh.write(result.to_json()) if not args.quiet: verdict = "PASS" if result.passed else "FAIL" - print( + _safe_print( f"freshdata validate: {verdict} — {result.n_errors} error(s), " f"{result.n_warnings} warning(s) against suite {suite.name!r}" ) for f in result.report.findings: if f.status != "passed": - print(f" [{f.status}] {f.check_id}: {f.message}") + _safe_print(f" [{f.status}] {f.check_id}: {f.message}") return 0 if result.passed else 1 diff --git a/src/freshdata/integrations/dbt/__init__.py b/src/freshdata/integrations/dbt/__init__.py index f09c2b8..02700ca 100644 --- a/src/freshdata/integrations/dbt/__init__.py +++ b/src/freshdata/integrations/dbt/__init__.py @@ -158,15 +158,39 @@ def gate_manifest( ) -> dict[str, Any]: """Gate every model in a dbt ``manifest.json`` and return a summary dict. - The summary has shape ``{"models": [...], "models_processed": int, - "failed_models": int, "all_passed": bool}``. A model that raises (e.g. its table - is missing) is recorded with an ``"error"`` and counted as failed, so one bad - model never aborts the whole run. + The summary has shape ``{"models": [...], "skipped": [...], + "models_processed": int, "failed_models": int, "all_passed": bool}``. A model + that raises (e.g. its table is missing) is recorded with an ``"error"`` and + counted as failed, so one bad model never aborts the whole run. + + Ephemeral models (never materialized by dbt) and disabled models are not read; + they are listed under ``"skipped"`` and not counted in ``models_processed``. + ``all_passed`` is ``False`` when no model was gated, so a manifest with nothing + to gate cannot pass as a clean run. + + Raises: + ValueError: the file is not valid JSON or has no ``nodes`` mapping (i.e. it + is not a dbt manifest, e.g. ``run_results.json``). """ on_low_score = validate_on_low_score(on_low_score) - manifest = json.loads(Path(manifest_path).read_text()) - nodes = manifest.get("nodes", {}) - models = [n for n in nodes.values() if n.get("resource_type") == "model"] + manifest = json.loads(Path(manifest_path).read_text(encoding="utf-8")) + nodes = manifest.get("nodes") if isinstance(manifest, dict) else None + if not isinstance(nodes, dict): + raise ValueError(f"{manifest_path} is not a dbt manifest: no 'nodes' mapping") + + models: list[Any] = [] # raw manifest nodes (untyped JSON) + skipped: list[dict[str, Any]] = [] + for node in nodes.values(): + if not isinstance(node, dict) or node.get("resource_type") != "model": + continue + config = node.get("config") + config = config if isinstance(config, dict) else {} + if config.get("materialized") == "ephemeral": + skipped.append({"model": node.get("name"), "reason": "ephemeral"}) + elif config.get("enabled") is False: + skipped.append({"model": node.get("name"), "reason": "disabled"}) + else: + models.append(node) summaries: list[dict[str, Any]] = [] failed = 0 @@ -205,7 +229,8 @@ def gate_manifest( return { "models": summaries, + "skipped": skipped, "models_processed": len(models), "failed_models": failed, - "all_passed": failed == 0, + "all_passed": failed == 0 and len(models) > 0, } diff --git a/src/freshdata/integrations/dbt/cli.py b/src/freshdata/integrations/dbt/cli.py index 5b1f42f..4e40187 100644 --- a/src/freshdata/integrations/dbt/cli.py +++ b/src/freshdata/integrations/dbt/cli.py @@ -52,7 +52,7 @@ def _build_parser() -> argparse.ArgumentParser: parser.add_argument( "--fail", action="store_true", - help="Exit non-zero when any model is below the threshold.", + help="Exit non-zero when any model is below the threshold or no model was gated.", ) return parser @@ -68,13 +68,21 @@ def main(argv: list[str] | None = None) -> int: on_low_score=args.on_low_score, output_dir=args.output_dir, ) - except FileNotFoundError as exc: - # A wrong manifest path is routine CLI misuse, not a crash: report it - # in one line instead of a traceback. Everything else propagates intact. + except (OSError, ValueError) as exc: + # A wrong, unreadable or malformed manifest (missing file, a directory, + # invalid JSON, not a dbt manifest) is routine CLI misuse, not a crash: + # report it in one line instead of a traceback. Exit 1, as `freshdata` + # does for bad input files. Everything else propagates intact. print(f"dbt-gate: error: {exc}", file=sys.stderr) return 1 json.dump(summary, sys.stdout, indent=2, default=str) sys.stdout.write("\n") + if summary["models_processed"] == 0: + print( + f"dbt-gate: no models were gated: {args.manifest} has no materialized " + "models (all_passed is false)", + file=sys.stderr, + ) if args.fail and not summary["all_passed"]: return 1 return 0 diff --git a/src/freshdata/validation_suite.py b/src/freshdata/validation_suite.py index 6ecf2ff..47d920f 100644 --- a/src/freshdata/validation_suite.py +++ b/src/freshdata/validation_suite.py @@ -225,6 +225,10 @@ def to_dict(self) -> dict[str, Any]: @classmethod def from_dict(cls, d: dict[str, Any]) -> ValidationSuite: + if not isinstance(d, dict): + raise ValueError( + f"a validation suite must be a JSON object, got {type(d).__name__}" + ) schema = d.get("schema_version", SUITE_SCHEMA_VERSION) if schema != SUITE_SCHEMA_VERSION: raise ValueError( diff --git a/tests/test_cli_malformed_inputs.py b/tests/test_cli_malformed_inputs.py new file mode 100644 index 0000000..af8b1cd --- /dev/null +++ b/tests/test_cli_malformed_inputs.py @@ -0,0 +1,191 @@ +"""Malformed CLI input files and non-UTF-8 stdout (#289, #295). + +Exit-code convention: ``freshdata clean`` reports a bad input/config file as a +one-line ``freshdata: error: ...`` with exit 1 (the same as a missing input file); +``freshdata validate`` reports rules it cannot load with exit 2 ("usage/load error") +so it is never confused with exit 1 ("validation failed"). +""" + +from __future__ import annotations + +import io +import json +import sys + +import pandas as pd +import pytest + +from freshdata.enterprise import cli +from freshdata.validation_suite import ValidationSuite + + +@pytest.fixture +def src(tmp_path): + path = tmp_path / "in.csv" + pd.DataFrame({"id": [1, 2], "email": ["a@b.com", "c@d.com"]}).to_csv(path, index=False) + return path + + +def _clean_with_config(src, cfg, tmp_path): + return cli.main( + ["clean", str(src), "-o", str(tmp_path / "out.csv"), "--config", str(cfg), "--quiet"] + ) + + +def _assert_one_line_error(capsys, *needles): + err = capsys.readouterr().err + assert "Traceback" not in err + assert "freshdata: error:" in err + for needle in needles: + assert needle in err + return err + + +# --------------------------------------------------------------------------- # +# #289: freshdata clean --config # +# --------------------------------------------------------------------------- # +def test_clean_invalid_yaml_config_is_one_line_error(src, tmp_path, capsys): + pytest.importorskip("yaml") + cfg = tmp_path / "bad.yaml" + cfg.write_text("enterprise:\n masking: [\n") + assert _clean_with_config(src, cfg, tmp_path) == 1 + err = _assert_one_line_error(capsys, "invalid YAML", "bad.yaml") + assert len(err.strip().splitlines()) == 1 + + +def test_clean_unknown_masking_rule_key_is_one_line_error(src, tmp_path, capsys): + cfg = tmp_path / "typo.json" + cfg.write_text( + json.dumps( + {"enterprise": {"masking": [{"name": "m", "columns": ["email"], "colums": 1}]}} + ) + ) + assert _clean_with_config(src, cfg, tmp_path) == 1 + _assert_one_line_error(capsys, "typo.json", "colums") + + +@pytest.mark.parametrize("payload", ["[1, 2]", '"text"', "3"]) +def test_clean_non_object_json_config_is_one_line_error(src, tmp_path, capsys, payload): + cfg = tmp_path / "list.json" + cfg.write_text(payload) + assert _clean_with_config(src, cfg, tmp_path) == 1 + _assert_one_line_error(capsys, "list.json", "must contain a JSON/YAML object") + + +def test_clean_non_object_yaml_config_is_one_line_error(src, tmp_path, capsys): + pytest.importorskip("yaml") + cfg = tmp_path / "list.yaml" + cfg.write_text("- 1\n- 2\n") + assert _clean_with_config(src, cfg, tmp_path) == 1 + _assert_one_line_error(capsys, "list.yaml", "must contain a JSON/YAML object") + + +@pytest.mark.parametrize("key", ["clean", "enterprise"]) +def test_clean_non_object_config_section_is_one_line_error(src, tmp_path, capsys, key): + cfg = tmp_path / "cfg.json" + cfg.write_text(json.dumps({key: [1, 2]})) + assert _clean_with_config(src, cfg, tmp_path) == 1 + _assert_one_line_error(capsys, f"'{key}'", "must be an object") + + +def test_clean_non_object_masking_entry_is_one_line_error(src, tmp_path, capsys): + cfg = tmp_path / "cfg.json" + cfg.write_text(json.dumps({"enterprise": {"masking": ["email"]}})) + assert _clean_with_config(src, cfg, tmp_path) == 1 + _assert_one_line_error(capsys, "'enterprise'", "cfg.json") + + +def test_clean_unknown_clean_option_is_one_line_error(src, tmp_path, capsys): + cfg = tmp_path / "cfg.json" + cfg.write_text(json.dumps({"clean": {"stratgy": "conservative"}})) + assert _clean_with_config(src, cfg, tmp_path) == 1 + _assert_one_line_error(capsys, "cfg.json", "stratgy") + + +def test_clean_empty_yaml_config_still_works(src, tmp_path): + pytest.importorskip("yaml") + cfg = tmp_path / "empty.yaml" + cfg.write_text("") + assert _clean_with_config(src, cfg, tmp_path) == 0 + + +def test_clean_valid_config_sections_still_work(src, tmp_path): + cfg = tmp_path / "ok.json" + cfg.write_text( + json.dumps( + { + "clean": {"strategy": "conservative"}, + "enterprise": {"masking": [{"name": "m", "columns": ["email"]}]}, + } + ) + ) + assert _clean_with_config(src, cfg, tmp_path) == 0 + assert "a@b.com" not in (tmp_path / "out.csv").read_text() + + +# --------------------------------------------------------------------------- # +# #289: freshdata validate --suite / --contract exit 2 on unloadable rules # +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("flag", ["--suite", "--contract"]) +@pytest.mark.parametrize("payload", ["[]", "[1, 2]", '"text"']) +def test_validate_non_object_rules_exit_2(src, tmp_path, capsys, flag, payload): + rules = tmp_path / "rules.json" + rules.write_text(payload) + assert cli.main(["validate", str(src), flag, str(rules)]) == 2 + err = capsys.readouterr().err + assert "could not load rules" in err + assert "must be a JSON object" in err + assert "Traceback" not in err + + +def test_suite_from_dict_rejects_non_mapping(): + with pytest.raises(ValueError, match="must be a JSON object"): + ValidationSuite.from_dict([]) # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- # +# #295: a non-UTF-8 stdout must not turn a passed gate into exit 1 # +# --------------------------------------------------------------------------- # +def _ascii_stdout(monkeypatch): + raw = io.BytesIO() + stream = io.TextIOWrapper(raw, encoding="ascii", errors="strict") + monkeypatch.setattr(sys, "stdout", stream) + return stream, raw + + +def test_clean_summary_on_ascii_stdout_exits_0(src, tmp_path, monkeypatch): + stream, raw = _ascii_stdout(monkeypatch) + out = tmp_path / "out.csv" + code = cli.main(["clean", str(src), "-o", str(out), "--fail-under-trust", "50"]) + stream.flush() + assert code == 0 + assert out.exists() + printed = raw.getvalue().decode("ascii") + assert "freshdata enterprise" in printed + assert "?" in printed # the arrow was replaced, not raised + + +def test_clean_failed_gate_on_ascii_stdout_still_exits_1(src, tmp_path, monkeypatch): + stream, _ = _ascii_stdout(monkeypatch) + code = cli.main( + ["clean", str(src), "-o", str(tmp_path / "out.csv"), "--fail-under-trust", "101"] + ) + stream.flush() + assert code == 1 + + +def test_validate_verdict_on_ascii_stdout_exits_0(src, tmp_path, monkeypatch): + suite = tmp_path / "suite.json" + suite.write_text(json.dumps({"name": "s"})) + stream, raw = _ascii_stdout(monkeypatch) + code = cli.main(["validate", str(src), "--suite", str(suite)]) + stream.flush() + assert code == 0 + assert "PASS" in raw.getvalue().decode("ascii") + + +def test_safe_print_replaces_unencodable_characters(monkeypatch): + stream, raw = _ascii_stdout(monkeypatch) + cli._safe_print("trust 1.0 → 2.0 — ok") + stream.flush() + assert raw.getvalue() == b"trust 1.0 ? 2.0 ? ok\n" diff --git a/tests/test_integrations/test_dbt.py b/tests/test_integrations/test_dbt.py index f9d0fb7..4fd8150 100644 --- a/tests/test_integrations/test_dbt.py +++ b/tests/test_integrations/test_dbt.py @@ -128,3 +128,132 @@ def test_cli_missing_manifest_prints_one_line_error(capsys): assert "dbt-gate: error:" in err assert "definitely_not_here.json" in err assert "Traceback" not in err + + +# --------------------------------------------------------------------------- # +# #289: malformed manifests are one-line errors, not tracebacks # +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize( + "content", ["{not json", "[1, 2]", '{"metadata": {}, "results": []}', '{"nodes": []}'] +) +def test_cli_malformed_manifest_prints_one_line_error(tmp_path, capsys, content): + path = tmp_path / "manifest.json" + path.write_text(content) + assert main(["--manifest", str(path), "--fail"]) == 1 + captured = capsys.readouterr() + assert "dbt-gate: error:" in captured.err + assert "Traceback" not in captured.err + assert captured.out == "" + + +def test_cli_manifest_directory_prints_one_line_error(tmp_path, capsys): + assert main(["--manifest", str(tmp_path), "--fail"]) == 1 + err = capsys.readouterr().err + assert "dbt-gate: error:" in err + assert "Traceback" not in err + + +# --------------------------------------------------------------------------- # +# #296: nothing gated must not look like a passing gate # +# --------------------------------------------------------------------------- # +def test_gate_manifest_rejects_non_manifest(tmp_path): + path = tmp_path / "run_results.json" + path.write_text(json.dumps({"metadata": {}, "results": []})) + with pytest.raises(ValueError, match="not a dbt manifest"): + gate_manifest(str(path)) + + +def test_manifest_with_no_models_does_not_pass(tmp_path, capsys): + path = tmp_path / "manifest.json" + path.write_text(json.dumps({"nodes": {"test.proj.t": {"resource_type": "test"}}})) + summary = gate_manifest(str(path)) + assert summary["models_processed"] == 0 + assert summary["failed_models"] == 0 + assert summary["all_passed"] is False + + assert main(["--manifest", str(path), "--fail"]) == 1 + err = capsys.readouterr().err + assert "no models were gated" in err + # Without --fail the run is still reported (exit 0), but not as a pass. + assert main(["--manifest", str(path)]) == 0 + captured = capsys.readouterr() + assert json.loads(captured.out)["all_passed"] is False + assert "no models were gated" in captured.err + + +# --------------------------------------------------------------------------- # +# #249: ephemeral / disabled models are skipped, not counted as failures # +# --------------------------------------------------------------------------- # +def _manifest_with_unmaterialized(tmp_path): + path = tmp_path / "manifest.json" + path.write_text( + json.dumps( + { + "nodes": { + "model.proj.orders": { + "resource_type": "model", + "name": "orders", + "schema": None, + "config": {"materialized": "table"}, + }, + "model.proj.stg_x": { + "resource_type": "model", + "name": "stg_x", + "schema": None, + "config": {"materialized": "ephemeral"}, + }, + "model.proj.old": { + "resource_type": "model", + "name": "old", + "schema": None, + "config": {"materialized": "table", "enabled": False}, + }, + } + } + ) + ) + return path + + +def test_manifest_skips_ephemeral_and_disabled_models(warehouse, tmp_path): + summary = gate_manifest( + str(_manifest_with_unmaterialized(tmp_path)), + conn_str=warehouse, + trust_score_threshold=0.0, + ) + assert [m["model"] for m in summary["models"]] == ["orders"] + assert summary["skipped"] == [ + {"model": "stg_x", "reason": "ephemeral"}, + {"model": "old", "reason": "disabled"}, + ] + assert summary["models_processed"] == 1 + assert summary["failed_models"] == 0 + assert summary["all_passed"] is True + + +def test_cli_fail_passes_with_ephemeral_model(warehouse, tmp_path, capsys): + manifest = str(_manifest_with_unmaterialized(tmp_path)) + rc = main(["--manifest", manifest, "--conn", warehouse, "--threshold", "0", "--fail"]) + assert rc == 0 + assert "stg_x" not in capsys.readouterr().err + + +def test_manifest_only_ephemeral_models_does_not_pass(tmp_path): + path = tmp_path / "manifest.json" + path.write_text( + json.dumps( + { + "nodes": { + "model.proj.stg_x": { + "resource_type": "model", + "name": "stg_x", + "config": {"materialized": "ephemeral"}, + } + } + } + ) + ) + summary = gate_manifest(str(path)) + assert summary["models_processed"] == 0 + assert summary["skipped"] == [{"model": "stg_x", "reason": "ephemeral"}] + assert summary["all_passed"] is False