From de296fdc554bcd70e43bc919a4b9e53c05eab2ec Mon Sep 17 00:00:00 2001 From: jakeross Date: Wed, 19 Aug 2026 12:17:27 -0700 Subject: [PATCH] test(ingestion): cover the Dagster assets ingest.py was at 16%. Every part of san_acacia_observations was covered on its own -- matching, resolving, watermarks, the adapter, the loader -- but the orchestration between them was not, which is where the decisions live: which wells get skipped, what the metadata reports, and whether one unresolvable well costs the others. Now 87%, and automated_ingestion overall 83% to 88%. Fakes stand in at the process boundaries only -- the vendor client, the database session, the dlt pipeline. The reconciler, resolver and adapter run for real, so a change in their behaviour surfaces here rather than being absorbed by a mock. Covered: a well with no match is skipped rather than invented; two wells sharing a name are skipped rather than picked between; a well with no open transducer, and one whose transducer was removed, are skipped; a refused reading is counted rather than vanishing; one bad well does not cost the others; both raw assets write parquet; a point the vendor refuses is counted and reported. What remains uncovered is _client and the two SQLAlchemy query helpers. They are thin wrappers around a live database and a real HTTP session, and testing them would mean asserting that mocks were called. Co-Authored-By: Claude Opus 5 --- .../tests/test_ingest_assets.py | 328 ++++++++++++++++++ 1 file changed, 328 insertions(+) create mode 100644 automated_ingestion/tests/test_ingest_assets.py diff --git a/automated_ingestion/tests/test_ingest_assets.py b/automated_ingestion/tests/test_ingest_assets.py new file mode 100644 index 000000000..7170fd9b3 --- /dev/null +++ b/automated_ingestion/tests/test_ingest_assets.py @@ -0,0 +1,328 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +""" +The Dagster assets, which is where the pieces meet. + +Every part of `san_acacia_observations` is covered on its own -- matching, +resolving, watermarks, the adapter, the loader. What was not covered is the +orchestration between them: which wells get skipped, what the metadata says, and +whether one unresolvable well costs the others. + +Fakes stand in at the process boundaries -- the vendor client, the database +session, the dlt pipeline -- and nowhere else. The reconciler, resolver and +adapter run for real, so a change in their behaviour shows up here. +""" + +from contextlib import contextmanager +from datetime import date + +import pytest +from dagster import build_asset_context + +from automated_ingestion.ocotillo import loader as loader_module +from automated_ingestion.ocotillo.loader import LoadResult +from automated_ingestion.shared import watermark as watermark_module +from automated_ingestion.sources.san_acacia import ingest +from automated_ingestion.sources.san_acacia.reconcile import ThingCandidate +from automated_ingestion.sources.san_acacia.resolve import DeploymentCandidate + +READING = {"dateAndTime": "2026-04-15T22:45:00", "level": 471.518} + + +class FakeClient: + """The vendor, reduced to what the assets ask of it.""" + + def __init__(self, points, readings=None, approved=()): + self._points = points + self._readings = READING if readings is None else readings + self._approved = approved + self.water_level_calls = [] + + def monitoring_points(self, project_id): + return self._points + + def water_levels(self, point_id, start, end, reference, approved=None, span=None): + self.water_level_calls.append((point_id, start, end, approved)) + if approved: + return iter(self._approved) + return iter( + self._readings if isinstance(self._readings, list) else [self._readings] + ) + + +class FakeDatabase: + """Stands in for OcotilloDatabase. The session is never really used -- + every function that would touch it is replaced.""" + + @contextmanager + def session(self): + yield object() + + +class NoWatermark: + def __init__(self, session): + pass + + def get(self, thing_id, parameter_id): + return None + + +@pytest.fixture() +def wired(monkeypatch): + """Wire the asset to fakes, returning the recorded loads.""" + loaded = [] + + def fake_load(session, records, deployment_id, parameter_id, release_status, **kw): + records = list(records) + loaded.append( + { + "deployment_id": deployment_id, + "parameter_id": parameter_id, + "release_status": release_status, + "rows": len(records), + } + ) + return LoadResult(rows_seen=len(records), rows_written=len(records), batches=1) + + monkeypatch.setattr(loader_module, "load_observations", fake_load) + monkeypatch.setattr(loader_module, "ensure_block", lambda *a, **k: 1) + monkeypatch.setattr(watermark_module, "PostgresWatermarkStore", NoWatermark) + monkeypatch.setattr(ingest, "_parameter_id", lambda session, name: 1) + return loaded + + +def _run(monkeypatch, client, candidates, deployments): + monkeypatch.setattr(ingest, "_client", lambda: client) + monkeypatch.setattr(ingest, "_well_candidates", lambda session: candidates) + monkeypatch.setattr(ingest, "_deployments", lambda session, thing_id: deployments) + return ingest.san_acacia_observations(build_asset_context(), FakeDatabase()) + + +TRANSDUCER = DeploymentCandidate(437, "Pressure Transducer") + + +class TestObservationsAsset: + def test_a_resolvable_well_is_loaded(self, monkeypatch, wired): + output = _run( + monkeypatch, + FakeClient([{"id": 39, "name": "SO-0125"}]), + [ThingCandidate(2343, "SO-0125")], + [TRANSDUCER], + ) + assert output.value == 1 + assert wired[0]["deployment_id"] == 437 + assert wired[0]["release_status"] == "public" + assert output.metadata["wells_skipped"].value == 0 + + def test_an_unmatched_well_is_skipped_not_invented(self, monkeypatch, wired): + # No Ocotillo well by that name. Ingestion does not create wells. + output = _run( + monkeypatch, + FakeClient([{"id": 39, "name": "SO-9999"}]), + [ThingCandidate(2343, "SO-0125")], + [TRANSDUCER], + ) + assert output.value == 0 + assert wired == [] + assert output.metadata["wells_skipped"].value == 1 + assert "unmatched" in str(output.metadata["skipped"].data) + + def test_an_ambiguous_well_is_skipped(self, monkeypatch, wired): + # Two wells share the name -- picking one would be a silent guess. + output = _run( + monkeypatch, + FakeClient([{"id": 39, "name": "SO-0125"}]), + [ThingCandidate(1, "SO-0125"), ThingCandidate(2, "SO-0125")], + [TRANSDUCER], + ) + assert wired == [] + assert "ambiguous" in str(output.metadata["skipped"].data) + + def test_a_well_without_a_transducer_is_skipped(self, monkeypatch, wired): + # SO-0246 is in this state in production. + output = _run( + monkeypatch, + FakeClient([{"id": 39, "name": "SO-0125"}]), + [ThingCandidate(2343, "SO-0125")], + [DeploymentCandidate(436, "DiverLink")], + ) + assert wired == [] + assert "missing" in str(output.metadata["skipped"].data) + + def test_a_removed_transducer_does_not_qualify(self, monkeypatch, wired): + output = _run( + monkeypatch, + FakeClient([{"id": 39, "name": "SO-0125"}]), + [ThingCandidate(2343, "SO-0125")], + [ + DeploymentCandidate( + 437, "Pressure Transducer", removal_date=date(2024, 1, 1) + ) + ], + ) + assert wired == [] + + def test_one_bad_well_does_not_cost_the_others(self, monkeypatch, wired): + # The point of skipping rather than raising. + output = _run( + monkeypatch, + FakeClient( + [ + {"id": 39, "name": "SO-9999"}, + {"id": 40, "name": "SO-0125"}, + ] + ), + [ThingCandidate(2343, "SO-0125")], + [TRANSDUCER], + ) + assert output.value == 1 + assert output.metadata["wells_attempted"].value == 2 + assert output.metadata["wells_skipped"].value == 1 + + def test_a_reading_the_adapter_refuses_is_counted(self, monkeypatch, wired): + # A null level has nothing to store; it should surface, not vanish. + client = FakeClient( + [{"id": 39, "name": "SO-0125"}], + readings=[{"dateAndTime": "2026-04-15T22:45:00", "level": None}], + ) + output = _run( + monkeypatch, client, [ThingCandidate(2343, "SO-0125")], [TRANSDUCER] + ) + assert output.value == 0 + assert output.metadata["adapter_failures"].value == 1 + + def test_no_wells_at_all(self, monkeypatch, wired): + output = _run(monkeypatch, FakeClient([]), [], [TRANSDUCER]) + assert output.value == 0 + assert output.metadata["wells_attempted"].value == 0 + + +class TestParameterLookup: + def test_a_missing_parameter_is_a_clear_error(self): + # Ingestion does not create parameters, so the message has to say what + # to do instead of surfacing an integrity error later. + class Empty: + def scalar(self, *_): + return None + + with pytest.raises(RuntimeError, match="does not create parameters"): + ingest._parameter_id(Empty(), "groundwater level") + + def test_a_found_parameter_is_returned(self): + class Found: + def scalar(self, *_): + return 7 + + assert ingest._parameter_id(Found(), "groundwater level") == 7 + + +class TestRowCount: + def test_malformed_load_info_reports_zero(self): + # Metadata must never fail a load that worked. + assert ingest._row_count(object()) == 0 + + def test_none_reports_zero(self): + assert ingest._row_count(None) == 0 + + +class FakePipeline: + """Stands in for the dlt pipeline. Records what it was asked to run.""" + + def __init__(self): + self.runs = [] + + def run(self, resource, loader_file_format=None): + # Consume the resource so the generator body actually executes. + rows = list(resource) if hasattr(resource, "__iter__") else [] + self.runs.append({"format": loader_file_format, "rows": len(rows)}) + return object() + + +class TestRawAssets: + def test_locations_reports_what_it_landed(self, monkeypatch): + from automated_ingestion.sources.san_acacia import dlt_pipeline + + pipeline = FakePipeline() + client = FakeClient( + [{"id": 39, "name": "SO-0125"}, {"id": 40, "name": "SO-0131"}] + ) + monkeypatch.setattr(ingest, "_client", lambda: client) + monkeypatch.setattr(dlt_pipeline, "build_pipeline", lambda: pipeline) + + output = ingest.raw_san_acacia_locations(build_asset_context()) + + assert output.value == 2 + assert output.metadata["monitoring_points"].value == 2 + assert "SO-0125" in output.metadata["names"].value + + def test_locations_are_written_as_parquet(self, monkeypatch): + # dlt writes gzipped JSONL unless told otherwise, and Mode B replay + # assumes parquet. + from automated_ingestion.sources.san_acacia import dlt_pipeline + + pipeline = FakePipeline() + monkeypatch.setattr(ingest, "_client", lambda: FakeClient([])) + monkeypatch.setattr(dlt_pipeline, "build_pipeline", lambda: pipeline) + + ingest.raw_san_acacia_locations(build_asset_context()) + assert pipeline.runs[0]["format"] == "parquet" + + def test_readings_report_per_point_failures(self, monkeypatch): + from automated_ingestion.sources.san_acacia import dlt_pipeline + + pipeline = FakePipeline() + client = FakeClient([{"id": 39, "name": "SO-0125"}]) + monkeypatch.setattr(ingest, "_client", lambda: client) + monkeypatch.setattr(dlt_pipeline, "build_pipeline", lambda: pipeline) + + output = ingest.raw_san_acacia_readings(build_asset_context()) + + assert output.metadata["points_attempted"].value == 1 + assert output.metadata["points_failed"].value == 0 + + def test_readings_are_written_as_parquet(self, monkeypatch): + from automated_ingestion.sources.san_acacia import dlt_pipeline + + pipeline = FakePipeline() + monkeypatch.setattr(ingest, "_client", lambda: FakeClient([])) + monkeypatch.setattr(dlt_pipeline, "build_pipeline", lambda: pipeline) + + ingest.raw_san_acacia_readings(build_asset_context()) + assert pipeline.runs[0]["format"] == "parquet" + + def test_readings_count_a_point_the_vendor_refuses(self, monkeypatch): + # One diver failing must cost that diver, not the run. The count is how + # anyone finds out it happened. + from automated_ingestion.sources.san_acacia import dlt_pipeline + from automated_ingestion.sources.san_acacia.client import DiverHubError + + class Refusing(FakeClient): + def water_levels(self, *a, **kw): + raise DiverHubError("500 at the minimum window") + + pipeline = FakePipeline() + client = Refusing([{"id": 39, "name": "SO-0125"}]) + monkeypatch.setattr(ingest, "_client", lambda: client) + monkeypatch.setattr(dlt_pipeline, "build_pipeline", lambda: pipeline) + + output = ingest.raw_san_acacia_readings(build_asset_context()) + + assert output.metadata["points_failed"].value == 1 + assert output.metadata["points_attempted"].value == 1 + assert "500" in str(output.metadata["failures"].data) + + +# ============= EOF =============================================