From 6d665b1680a56623a8affa4677bfaae87f79d567 Mon Sep 17 00:00:00 2001 From: Shreyas Telkar Date: Thu, 20 Aug 2026 15:51:55 -0700 Subject: [PATCH] fix: [AI-8671] accept dbt Fusion's capitalized freshness statuses in sources v1-v3 The dbt Fusion engine serializes source-freshness `status` with its Rust variant names -- `"Pass"` / `"Warn"` / `"Error"` -- while dbt-core and the published `sources/v3.json` schema Fusion itself stamps into the artifact both use the lowercase forms. Every result row therefore failed both members of the `results` union, the whole `sources.json` raised a `ValidationError`, and the ingestion worker silently dropped it. Nine of harvestgroup's ten production environments have zero source-freshness rows as a result. This is the residual of the AI-7675 work: PR #106/#108 added the `_missing_` forward-compat shim to the `run_results` status enums but explicitly left the freshness enums, and the `sources` parsers entirely, untouched. - Make `Status1` a `str, Enum` whose `_missing_` case-folds to the canonical lowercase member first, then falls back to the same forward-compat pseudo-member used by the `run_results` shim for unknown statuses. - Case-folding rather than adding PascalCase members is deliberate: the extractor persists `status.value`, and every dbt-core-backed tenant already writes lowercase into the same table. - Leave the runtime-error-only `Status` enum strict. Fusion has no runtime-error variant, and loosening it would let a row missing a required field fall silently into the field-less branch instead of erroring. - Apply to v1-v3 rather than v3 alone, mirroring how #108 had to follow #106 across the older schemas. Verified against 432 real production artifacts spanning all nine Fusion environments and 13 Fusion builds (preview.190 -> .210), run through parse AND the worker's own `extract_sources`: 432/432 parsed, 10,893 freshness rows extracted, every capitalized input landing as its lowercase counterpart. Co-Authored-By: Claude Opus 5 (1M context) --- .../parsers/sources/sources_v1.py | 23 +++- .../parsers/sources/sources_v2.py | 23 +++- .../parsers/sources/sources_v3.py | 23 +++- .../test_sources_freshness_status.py | 127 ++++++++++++++++++ 4 files changed, 193 insertions(+), 3 deletions(-) create mode 100644 tests/test_vendor/test_sources_freshness_status.py diff --git a/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v1.py b/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v1.py index 6f9a036..44c65e6 100644 --- a/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v1.py +++ b/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v1.py @@ -38,12 +38,33 @@ class SourceFreshnessRuntimeError(BaseParserModel): status: Status -class Status1(Enum): +class Status1(str, Enum): pass_ = "pass" warn = "warn" error = "error" runtime_error = "runtime error" + @classmethod + def _missing_(cls, value): + # The dbt Fusion engine serializes freshness statuses with its Rust + # variant names -- "Pass" / "Warn" / "Error" -- while the published + # sources schema it stamps into the artifact, and every dbt-core + # release, use the lowercase forms. Fold case first so a Fusion + # artifact resolves to the canonical lowercase member and `.value` + # stays stable for downstream storage. + if isinstance(value, str): + folded = value.casefold() + for member in cls: + if member.value.casefold() == folded: + return member + # Forward-compatibility: surface any other unknown status as a real + # member so downstream `.value` access keeps working instead of failing + # validation and silently dropping the entire sources.json. + member = str.__new__(cls, value) + member._name_ = str(value) + member._value_ = value + return member + class Period(Enum): minute = "minute" diff --git a/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v2.py b/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v2.py index a83c3c4..4dcb64f 100644 --- a/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v2.py +++ b/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v2.py @@ -38,12 +38,33 @@ class SourceFreshnessRuntimeError(BaseParserModel): status: Status -class Status1(Enum): +class Status1(str, Enum): pass_ = "pass" warn = "warn" error = "error" runtime_error = "runtime error" + @classmethod + def _missing_(cls, value): + # The dbt Fusion engine serializes freshness statuses with its Rust + # variant names -- "Pass" / "Warn" / "Error" -- while the published + # sources schema it stamps into the artifact, and every dbt-core + # release, use the lowercase forms. Fold case first so a Fusion + # artifact resolves to the canonical lowercase member and `.value` + # stays stable for downstream storage. + if isinstance(value, str): + folded = value.casefold() + for member in cls: + if member.value.casefold() == folded: + return member + # Forward-compatibility: surface any other unknown status as a real + # member so downstream `.value` access keeps working instead of failing + # validation and silently dropping the entire sources.json. + member = str.__new__(cls, value) + member._name_ = str(value) + member._value_ = value + return member + class Period(Enum): minute = "minute" diff --git a/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v3.py b/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v3.py index b4465c0..4bca714 100644 --- a/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v3.py +++ b/src/vendor/dbt_artifacts_parser/parsers/sources/sources_v3.py @@ -38,12 +38,33 @@ class Results(BaseParserModel): status: Status -class Status1(Enum): +class Status1(str, Enum): pass_ = "pass" warn = "warn" error = "error" runtime_error = "runtime error" + @classmethod + def _missing_(cls, value): + # The dbt Fusion engine serializes freshness statuses with its Rust + # variant names -- "Pass" / "Warn" / "Error" -- while the published + # sources schema it stamps into the artifact, and every dbt-core + # release, use the lowercase forms. Fold case first so a Fusion + # artifact resolves to the canonical lowercase member and `.value` + # stays stable for downstream storage. + if isinstance(value, str): + folded = value.casefold() + for member in cls: + if member.value.casefold() == folded: + return member + # Forward-compatibility: surface any other unknown status as a real + # member so downstream `.value` access keeps working instead of failing + # validation and silently dropping the entire sources.json. + member = str.__new__(cls, value) + member._name_ = str(value) + member._value_ = value + return member + class Period(Enum): minute = "minute" diff --git a/tests/test_vendor/test_sources_freshness_status.py b/tests/test_vendor/test_sources_freshness_status.py new file mode 100644 index 0000000..2345d81 --- /dev/null +++ b/tests/test_vendor/test_sources_freshness_status.py @@ -0,0 +1,127 @@ +"""Tests for the sources v1-v3 parsers, specifically the resilient freshness `Status1` enum. + +Regression coverage for the dbt Fusion engine emitting capitalized freshness +statuses (``"Pass"`` / ``"Warn"`` / ``"Error"``) in sources.json. Every result row +failed both members of the ``results`` union, so the ENTIRE sources.json raised a +``ValidationError`` and was silently dropped during ingestion -- no source freshness +ever reached Postgres for a Fusion-backed environment. + +The statuses must fold to their canonical lowercase members, because the extractor +persists ``result.status.value`` and the rest of the platform (including every +dbt-core-backed tenant already in the same table) uses the lowercase vocabulary. +""" +import pytest + +from vendor.dbt_artifacts_parser.parser import parse_sources +from vendor.dbt_artifacts_parser.parsers.sources.sources_v1 import SourceFreshnessOutput as OutputV1 +from vendor.dbt_artifacts_parser.parsers.sources.sources_v1 import Status1 as StatusV1 +from vendor.dbt_artifacts_parser.parsers.sources.sources_v2 import SourceFreshnessOutput as OutputV2 +from vendor.dbt_artifacts_parser.parsers.sources.sources_v2 import Status1 as StatusV2 +from vendor.dbt_artifacts_parser.parsers.sources.sources_v3 import Results as RuntimeErrorV3 +from vendor.dbt_artifacts_parser.parsers.sources.sources_v3 import Results1 as OutputV3 +from vendor.dbt_artifacts_parser.parsers.sources.sources_v3 import Status1 as StatusV3 + +V3_SCHEMA = "https://schemas.getdbt.com/dbt/sources/v3.json" + +# (output model, status enum) per schema version -- the enum is identical in all three. +VERSIONS = [ + pytest.param(OutputV1, StatusV1, id="v1"), + pytest.param(OutputV2, StatusV2, id="v2"), + pytest.param(OutputV3, StatusV3, id="v3"), +] + +# Fusion's Rust variant name -> the canonical lowercase status dbt-core emits. +FUSION_CASINGS = [("Pass", "pass"), ("Warn", "warn"), ("Error", "error")] + + +def _output(status: str, unique_id: str = "source.proj.schema.tbl") -> dict: + """A complete freshness result -- the shape Fusion always emits.""" + return { + "unique_id": unique_id, + "max_loaded_at": "2026-08-19T08:33:50.855920Z", + "snapshotted_at": "2026-08-19T17:00:33.833000Z", + "max_loaded_at_time_ago_in_s": 30402.0, + "status": status, + "criteria": { + "warn_after": {"count": 24, "period": "hour"}, + "error_after": {"count": 48, "period": "hour"}, + }, + "adapter_response": {}, + "timing": [], + "thread_id": "Thread-20", + "execution_time": 0.0, + } + + +def _sources_v3(*statuses: str) -> dict: + return { + "metadata": { + "dbt_schema_version": V3_SCHEMA, + "dbt_version": "2.0.0-preview.210", + "invocation_id": "test-invocation-123", + }, + "elapsed_time": 1.5, + "results": [_output(s, f"source.proj.sch.t{i}") for i, s in enumerate(statuses)], + } + + +class TestFusionStatusCasing: + """Fusion's capitalized statuses must parse AND normalize to lowercase.""" + + @pytest.mark.parametrize(("model", "status_enum"), VERSIONS) + @pytest.mark.parametrize(("fusion", "canonical"), FUSION_CASINGS) + def test_capitalized_status_folds_to_canonical_member(self, model, status_enum, fusion, canonical): + result = model(**_output(fusion)) + assert result.status is status_enum(canonical) + assert result.status.value == canonical + + @pytest.mark.parametrize(("model", "status_enum"), VERSIONS) + def test_lowercase_statuses_unchanged(self, model, status_enum): + """dbt-core's existing lowercase vocabulary must keep resolving as before.""" + for status in ("pass", "warn", "error", "runtime error"): + assert model(**_output(status)).status.value == status + + @pytest.mark.parametrize(("model", "status_enum"), VERSIONS) + def test_unknown_future_status_parses(self, model, status_enum): + """Forward-compat: a status dbt has not shipped yet must not drop the file.""" + assert model(**_output("some_future_status")).status.value == "some_future_status" + + +class TestUnionResolutionIsNotLossy: + """A complete freshness row must NEVER resolve to the runtime-error branch. + + ``SourcesV3.results`` is ``list[Union[Results, Results1]]`` and ``Results`` (the + runtime-error shape) requires only ``unique_id`` + ``status`` with ``extra="allow"``. + If a full row resolved there, every freshness field would be dropped -- turning a + loud parse failure into silent data loss, which is strictly worse. + """ + + @pytest.mark.parametrize(("fusion", "canonical"), FUSION_CASINGS) + def test_full_row_resolves_to_output_branch(self, fusion, canonical): + parsed = parse_sources(_sources_v3(fusion)) + (result,) = parsed.results + assert isinstance(result, OutputV3) + assert result.status.value == canonical + assert result.max_loaded_at == "2026-08-19T08:33:50.855920Z" + assert result.criteria.error_after.count == 48 + + def test_runtime_error_row_still_resolves_to_runtime_error_branch(self): + """dbt-core emits a distinct, field-less shape for a failed freshness check.""" + artifact = _sources_v3() + artifact["results"] = [ + { + "unique_id": "source.proj.sch.broken", + "error": "Database Error: permission denied", + "status": "runtime error", + } + ] + (result,) = parse_sources(artifact).results + assert isinstance(result, RuntimeErrorV3) + assert result.status.value == "runtime error" + + +class TestFullArtifactParse: + def test_fusion_artifact_parses_end_to_end(self): + """The whole-file failure this regression is about: mixed Fusion statuses.""" + parsed = parse_sources(_sources_v3("Pass", "Error", "Warn", "Pass")) + assert [r.status.value for r in parsed.results] == ["pass", "error", "warn", "pass"]