diff --git a/CLAUDE.md b/CLAUDE.md index 30549235f..1bb5bdcaa 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -164,6 +164,12 @@ Authentik groups granted. **Role families are orthogonal**: general `Admin` confers nothing in the AMP or Lexicon families. Only tiers *within* a family nest. +**`AMP.Staging`** is a standalone group, not a fourth AMP tier — `AMPAdmin` +does not satisfy it. It gates the hydrograph corrector's publish and range-delete +routes while the workbench is being validated against real logger files, so they +ship dark. Read **`docs/hydrograph-correction-publish.md`** before changing +them. + **Authorization is opt-in per endpoint** — a `user: _dependency` parameter in the signature, not a router-level `dependencies=[...]`. Omitting it produces a fully public endpoint with no error. `tests/test_authorization.py` holds the diff --git a/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py b/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py new file mode 100644 index 000000000..48e6bebdf --- /dev/null +++ b/alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py @@ -0,0 +1,130 @@ +"""publish provenance for corrected transducer blocks + +Revision ID: c3d4e5f6a7b8 +Revises: b2c3d4e5f6a7 +Create Date: 2026-08-19 + +The hydrograph corrector publishes a *derived* series: water head converted to +depth below ground surface against manual anchors, then shifted, snapped, and +drift-corrected. None of those numbers are what the instrument recorded, so the +database has to carry enough to tell a reviewer what happened to them. + +Three columns on the block cover the batch: the file it came from, whether that +file held water head or depth to water, and the ordered list of corrections +applied. `comment` already exists and takes the publisher's free-text note. + +One column on the observation covers the row: `note`, set only on readings a +correction actually moved. NULL therefore means "as measured", which is the +distinction review needs. The legacy `nma_waterlevelscontinuous_*_notes` +columns cannot serve -- each is scoped to one legacy source table. + +The block time-order check is relaxed from `>` to `>=`. A block spanning a +single instant is legitimate: a published file with one reading, or a block +narrowed by a range delete until one observation survives. The block reader +matches observations inclusively on both bounds, so a zero-width block still +covers its reading. Loosening a check constraint cannot invalidate existing +rows. + +That check also gets its name spelled right on the way through. It was created +as `check_transuder_block_time_order` -- no `c` -- and since Postgres cannot +alter a check in place, the drop-and-recreate this migration already performs +is the free moment to fix it. The old name is dropped and the new one created; +no separate RENAME is needed. +""" + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision = "c3d4e5f6a7b8" +down_revision = "b2c3d4e5f6a7" +branch_labels = None +depends_on = None + +# The name as created by the initial migration, misspelled. Drops and +# downgrades have to use it verbatim: `op.drop_constraint` matches on the name +# in the live database, so correcting the spelling anywhere it is used to +# *find* the constraint would make the statement a no-op target and fail. +LEGACY_TIME_ORDER_CONSTRAINT = "check_transuder_block_time_order" + +# What it is called from this migration forward. +TIME_ORDER_CONSTRAINT = "check_transducer_block_time_order" + + +def upgrade() -> None: + op.add_column( + "transducer_observation_block", + sa.Column( + "source_file", + sa.String(length=255), + nullable=True, + comment="Name of the logger file the corrected series was derived from", + ), + ) + op.add_column( + "transducer_observation_block", + sa.Column( + "source_kind", + sa.String(length=50), + nullable=True, + comment="What the source file measured: water_head or depth_to_water", + ), + ) + op.add_column( + "transducer_observation_block", + sa.Column( + "corrections", + postgresql.JSONB(astext_type=sa.Text()), + nullable=True, + comment="Corrections applied to the source series, in applied order", + ), + ) + op.add_column( + "transducer_observation", + sa.Column( + "note", + sa.Text(), + nullable=True, + comment=( + "Per-reading correction annotation; NULL means the value is as " + "measured" + ), + ), + ) + + # Dropped under the old name, recreated under the new one: the rename and + # the relaxation are the same statement pair, so there is no window where + # the table is unconstrained beyond the one this already needs. + op.drop_constraint( + LEGACY_TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check" + ) + op.create_check_constraint( + TIME_ORDER_CONSTRAINT, + "transducer_observation_block", + "end_datetime >= start_datetime", + ) + + +def downgrade() -> None: + # Zero-width blocks may have been created while the loosened constraint was + # in force, so widen them by a second rather than let the stricter + # constraint fail to validate. A one-second span on a block that covered an + # instant is a smaller lie than a failed downgrade. + op.execute( + "UPDATE transducer_observation_block " + "SET end_datetime = start_datetime + interval '1 second' " + "WHERE end_datetime = start_datetime" + ) + op.drop_constraint( + TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check" + ) + op.create_check_constraint( + LEGACY_TIME_ORDER_CONSTRAINT, + "transducer_observation_block", + "end_datetime > start_datetime", + ) + + op.drop_column("transducer_observation", "note") + op.drop_column("transducer_observation_block", "corrections") + op.drop_column("transducer_observation_block", "source_kind") + op.drop_column("transducer_observation_block", "source_file") diff --git a/api/observation.py b/api/observation.py index d4c7fff78..fb11ca9d7 100644 --- a/api/observation.py +++ b/api/observation.py @@ -28,6 +28,7 @@ session_dependency, amp_admin_dependency, amp_editor_dependency, + amp_staging_dependency, amp_viewer_dependency, ) from db import Observation, Parameter @@ -40,7 +41,12 @@ UpdateGroundwaterLevelObservation, UpdateWaterChemistryObservation, ) -from schemas.transducer import TransducerObservationWithBlockResponse +from schemas.transducer import ( + DeletedTransducerObservationsResponse, + PublishedTransducerBlockResponse, + PublishTransducerBlock, + TransducerObservationWithBlockResponse, +) from schemas.water_level_csv import WaterLevelBulkUploadResponse from services.crud_helper import model_deleter, model_adder from services.observation_helper import ( @@ -50,10 +56,30 @@ get_transducer_observations, ) from services.query_helper import simple_get_by_id +from services.transducer_helper import ( + delete_transducer_observations, + publish_transducer_block, +) from services.water_level_csv import bulk_upload_water_levels router = APIRouter(prefix="/observation", tags=["observation"]) + +def _groundwater_level_parameter_id(session) -> int: + """ + The lexicon id the transducer routes work in. + + Looked up rather than configured so the publish, read, and delete routes + cannot drift onto different parameters. + """ + return ( + session.query(Parameter) + .filter(Parameter.parameter_name == "groundwater level") + .one() + .id + ) + + """ TODO @@ -88,6 +114,37 @@ def add_water_chemistry_observation( return model_adder(session, Observation, obs_data, user=user) +@router.post( + "/transducer-groundwater-level/block", + status_code=HTTP_201_CREATED, + summary="Publish a corrected transducer series as one block", +) +def publish_transducer_groundwater_level_block( + payload: PublishTransducerBlock, + session: session_dependency, + user: amp_staging_dependency, + replace_overlapping: bool = False, +) -> PublishedTransducerBlockResponse: + """ + Publish one corrected logger file as a single observation block. + + The block's time span is derived from the measurements; the client does not + send it. Overlapping an existing block is a 409 listing the collisions -- + pass `replace_overlapping=true` to supersede them, which deletes those + blocks and their readings in the same transaction. + + Written by the hydrograph corrector in OcotilloUI. See + `docs/hydrograph-correction-publish.md`. + """ + return publish_transducer_block( + session, + payload, + parameter_id=_groundwater_level_parameter_id(session), + user=user, + replace_overlapping=replace_overlapping, + ) + + @router.post( "/groundwater-level/bulk-upload", response_model=WaterLevelBulkUploadResponse, @@ -155,17 +212,30 @@ def get_transducer_groundwater_level_observations( thing_id: int | None = None, start_time: datetime | None = None, end_time: datetime | None = None, + sort: str | None = None, + order: str | None = None, ) -> CustomPage[TransducerObservationWithBlockResponse]: + """ + Retrieve transducer groundwater level observations paired with the block + that covers them. - groundwater_parameter_id = ( - session.query(Parameter) - .filter(Parameter.parameter_name == "groundwater level") - .one() - .id - ) - + `sort` accepts `observation_datetime`, `value`, or `id`; `order` accepts + `asc` or `desc`. The default is newest first, so a client that wants the + latest stored reading for a well can ask for size 1. + """ + # Keyword arguments deliberately: the helper's fourth positional parameter + # is `sensor_id`, so the previous positional call passed `start_time` as a + # sensor id (unused, silently dropped), `end_time` as `start_time`, and + # nothing as `end_time` -- an upper bound the caller asked for was ignored + # and the lower bound came from the wrong argument. return get_transducer_observations( - session, thing_id, groundwater_parameter_id, start_time, end_time + session, + thing_id=thing_id, + parameter_id=_groundwater_level_parameter_id(session), + start_time=start_time, + end_time=end_time, + sort=sort, + order=order, ) @@ -302,6 +372,40 @@ def get_observation_by_id( # DELETE ======================================================================= +@router.delete( + "/transducer-groundwater-level", + status_code=HTTP_200_OK, + summary="Delete transducer groundwater level observations in a time range", +) +def delete_transducer_groundwater_level_observations( + session: session_dependency, + user: amp_staging_dependency, + thing_id: int, + start_time: datetime, + end_time: datetime, +) -> DeletedTransducerObservationsResponse: + """ + Delete every transducer groundwater level reading for a well inside a + closed time range, and reconcile the blocks that covered them: a block left + with no readings is deleted, one left with some has its span narrowed to + the survivors. + + All three parameters are required -- there is deliberately no form of this + request that deletes everything for a well. Scoped exactly like the `GET` + on this path, so the set previewed there is the set removed here. + + Irreversible, and it leaves the `transducer_daily_data` materialized view + stale until its next refresh. + """ + return delete_transducer_observations( + session, + thing_id=thing_id, + parameter_id=_groundwater_level_parameter_id(session), + start_time=start_time, + end_time=end_time, + ) + + @router.delete( "/{observation_id}", summary="Delete an observation", diff --git a/core/dependencies.py b/core/dependencies.py index 09e7c3f79..95d11f3c8 100644 --- a/core/dependencies.py +++ b/core/dependencies.py @@ -59,6 +59,21 @@ amp_viewer_function = authenticated(any_of=["AMPAdmin", "AMPEditor", "AMPViewer"]) +# Hydrograph-Corrector Staging Permissions ------------------------------------- +# The hydrograph corrector's publish and range-delete routes write and destroy +# transducer records, and the workbench driving them is still being validated +# against real logger files. `AMP.Staging` is its own group with no tier below +# it and no AMP tier above it -- an AMPAdmin does not satisfy it. Nobody holds +# it until it is granted in Authentik, so the routes ship dark and reachable +# only by whoever is testing them. +# +# This is deliberately not a fourth rung on the AMP ladder. When the workbench +# is trusted, these routes move to `amp_admin_dependency` and the group goes +# away; leaving it as a tier would make that a schema change instead of a +# one-line edit. +amp_staging_function = authenticated(any_of=["AMP.Staging"]) + + # Lexicon-Specific Authentication/Permissions ---------------------------------- lexicon_admin_function = authenticated(any_of=["LexiconAdmin"]) @@ -89,5 +104,7 @@ amp_editor_dependency: TypeAlias = Annotated[dict, Depends(amp_editor_function)] amp_viewer_dependency: TypeAlias = Annotated[dict, Depends(amp_viewer_function)] +amp_staging_dependency: TypeAlias = Annotated[dict, Depends(amp_staging_function)] + no_permission_dependency: TypeAlias = Annotated[dict, Depends(no_permission_function)] # ============= EOF ============================================= diff --git a/db/transducer.py b/db/transducer.py index d109adc58..e129cfe60 100644 --- a/db/transducer.py +++ b/db/transducer.py @@ -28,6 +28,7 @@ Index, UniqueConstraint, ) +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import mapped_column, Mapped, relationship from db import Base, AutoBaseMixin, ReleaseMixin, lexicon_term @@ -62,6 +63,32 @@ class TransducerObservationBlock(Base, AutoBaseMixin, ReleaseMixin): ) comment: Mapped[str] = mapped_column(Text, nullable=True) + + # Publish provenance. A corrected block is derived data -- the numbers in it + # are not what any instrument recorded -- so the file it came from and the + # operations applied to it are part of the record, not metadata about it. A + # reviewer who cannot see that a series was snapped to a manual measurement + # cannot review it. + source_file: Mapped[str] = mapped_column( + String(255), + nullable=True, + comment="Name of the logger file the corrected series was derived from", + ) + source_kind: Mapped[str] = mapped_column( + String(50), + nullable=True, + comment="What the source file measured: water_head or depth_to_water", + ) + # A list of strings in applied order rather than a modelled correction + # entity: the corrector's operation set is still moving, and freezing it + # into columns now would mean a migration per new operation. The strings + # are written by the workbench and read by humans. + corrections: Mapped[list] = mapped_column( + JSONB, + nullable=True, + comment="Corrections applied to the source series, in applied order", + ) + reviewer_id: Mapped[str] = mapped_column( ForeignKey("contact.id", ondelete="CASCADE"), nullable=True, @@ -81,8 +108,13 @@ class TransducerObservationBlock(Base, AutoBaseMixin, ReleaseMixin): "end_datetime", name="uq_transducer_block_thing_status_parameter_time", ), + # Non-strict: a block covering a single instant is legitimate -- a + # published file with one reading, or a block narrowed by a range + # delete until one observation survives. The block reader matches + # observations inclusively on both bounds, so a zero-width block still + # covers its reading. CheckConstraint( - "end_datetime > start_datetime", name="check_transuder_block_time_order" + "end_datetime >= start_datetime", name="check_transducer_block_time_order" ), Index( "ix_transducer_block_time", @@ -137,6 +169,17 @@ class TransducerObservation(Base, AutoBaseMixin, ReleaseMixin): ) value: Mapped[float] = mapped_column(Float, nullable=False) + # Why this reading differs from what the sensor recorded. Present only on + # readings a correction actually moved, so a NULL note means the value is + # as measured -- which is the distinction review needs and which the legacy + # `nma_waterlevelscontinuous_*_notes` columns cannot carry, being scoped to + # one legacy source each. + note: Mapped[str] = mapped_column( + Text, + nullable=True, + comment="Per-reading correction annotation; NULL means the value is as measured", + ) + # How far through review this reading is, on USGS terms: provisional, # in review, approved. Orthogonal to `release_status`, which says who may # see it -- a reading can be public and provisional at once, which one diff --git a/docs/hydrograph-correction-publish.md b/docs/hydrograph-correction-publish.md new file mode 100644 index 000000000..3d8a59fbb --- /dev/null +++ b/docs/hydrograph-correction-publish.md @@ -0,0 +1,144 @@ +# Hydrograph correction — publish and range delete + +The hydrograph corrector in OcotilloUI (`/ocotillo/hydrograph-correction`) +ingests a raw logger file, converts water head to depth below ground surface +against manual measurements, applies corrections, and publishes the result +here. This document records what the API side actually does; the UI-side +proposal it was built from is +`OcotilloUI/docs/hydrograph-correction-upload-contract.md`. + +## Authorization + +Both write routes are gated on **`AMP.Staging`**, a standalone Authentik group. +It is not a fourth rung on the AMP ladder: `AMPAdmin` does not satisfy it, and +it satisfies nothing else. Nobody holds it until it is granted, so the routes +ship dark and are reachable only by whoever is validating the workbench against +real logger files. + +When the workbench is trusted, these routes move to `amp_admin_dependency` and +the group goes away. Leaving it as a tier would make that a schema change +instead of a one-line edit. + +The read route stays on `amp_viewer_dependency` — it was already public to +viewers and publishing does not change who may look. + +## `POST /observation/transducer-groundwater-level/block` + +One corrected logger file becomes one block plus all of its readings, in one +transaction. + +- **The span is derived, not sent.** `start_datetime`/`end_datetime` come from + the min/max measurement timestamp. A client-supplied span wider than the data + would make the block claim readings it does not contain, because nothing links + the observation table to the block table — the reader pairs them by time. +- **`deployment_id` is optional.** Omitted, it is resolved from the deployments + on the well whose installation period covers the span. A NULL installation date + reads as "always installed", a NULL removal date as "still installed". Zero or + more than one match is a 422 telling the client to send it explicitly, because + guessing attributes readings to hardware that did not record them. +- **`data_maturity` is derived from `review_status`**, not sent: a block + published as `not reviewed` is `provisional` on USGS terms. Sending both + separately would let a client store a contradiction. +- **Provenance is part of the record.** `source_file`, `source_kind`, and the + ordered `corrections` list live on the block; `provenance.notes` lands in the + block's existing `comment`. A reviewer who cannot see that a series was + snapped to a manual measurement cannot review it. +- **Per-reading `note`** is set only where a correction moved the value, so NULL + means "as measured" rather than "unknown". + +- **`parameter_id` is validated, not obeyed.** The client states it explicitly, + per the contract, but the route checks it against the parameter the route is + scoped to. The read and delete routes on this path resolve groundwater level + themselves, so a block accepted under any other parameter would be a 201 for + data neither of them could ever list or remove. + +### Concurrency + +Both write paths read state, decide, and then write based on what they read, so +each takes a transaction-scoped advisory lock on `(thing_id, parameter_id)` +first — `pg_advisory_xact_lock`. + +Without it, two publishes with different timestamps but overlapping spans each +see no existing block and both commit: the unique constraints only catch +identical spans and identical readings, and the inclusive reader then has two +blocks claiming the same instants. Two range deletes each compute survivors from +a snapshot the other is invalidating, and the later update can widen a block back +over readings the earlier one removed. + +An advisory lock rather than row locks because on publish there is no row to +lock — the conflict is with a block that does not exist yet — so what needs +guarding is the series, not a row. Both paths take the same key, so they +serialize against each other and cannot deadlock against one another. + +### Overlap + +An existing block for the same well and parameter whose span shares any instant +with the new one is a **409** listing the collisions in +`detail[0].input.overlapping_blocks`. `?replace_overlapping=true` deletes those +blocks **and their readings** in the same transaction and then publishes. + +The readings have to go with the block. Keeping them would leave rows the reader +cannot show — no block covers them — that still occupy the +deployment/parameter/instant the new series is about to claim, so a "replace" +that kept them would fail on the very insert it was asked to make room for. + +Overlap here is **inclusive on both bounds**, unlike +`TransducerObservationBlock.overlaps` on the model, which is half-open. The +reader matches a reading to a block with `start <= t <= end`, so two blocks +sharing an endpoint both claim any reading at that instant — exactly the +ambiguity this check exists to prevent. + +Readings can also survive a block deleted by hand. Those are caught separately +and reported as a 409 naming the earliest colliding timestamp, rather than +letting the insert abort the transaction with a constraint name. + +## `DELETE /observation/transducer-groundwater-level` + +`thing_id`, `start_time`, and `end_time` are all required. There is deliberately +no unbounded form of this request. The scope matches the `GET` on the same path +exactly, so the set a client previews is the set this removes. + +Blocks are reconciled afterwards: one left with no readings is deleted, one left +with some has its span narrowed to the survivors. A block narrowed to a single +reading becomes zero-width, which the `end_datetime >= start_datetime` check +constraint allows on purpose (migration `c3d4e5f6a7b8`) and which the inclusive +reader still covers. + +That same migration renames the constraint from `check_transuder_...` to +`check_transducer_...`. Postgres cannot alter a check in place, so the +drop-and-recreate the relaxation already required was the free moment to fix +the spelling. The downgrade puts the old name back, so anything reaching for +the constraint by name has to pick the spelling that matches the revision it is +running against. + +**This leaves the `transducer_daily_data` materialized view stale** until its +next scheduled refresh. Nothing here refreshes it — a full refresh on every +delete would cost far more than the correctness it buys between nightly runs. + +## Two things fixed in passing + +- The read route was calling `get_transducer_observations` positionally, and the + helper's fourth positional parameter is `sensor_id`. `start_time` was landing + in `sensor_id` (unused, silently dropped), `end_time` was landing in + `start_time`, and `end_time` was never set — so an upper bound a caller asked + for was ignored and the lower bound came from the wrong argument. The call is + keyword-only now. +- The read route honours `sort` (`observation_datetime`, `value`, `id`) and + `order` (`asc`/`desc`), defaulting to newest first. An unrecognised sort field + or order is a 422 rather than being ignored — silently returning a differently + ordered page reads as the data changing, not as a bad request. `order` matters + particularly here: anything other than `asc` used to fall through to + descending, so the near-miss `order=ascending` returned 200 with the rows in + exactly the opposite order to the one asked for. + +## Not built + +Everything in the contract's "Supporting endpoints for Wellntel ingestion" +section is deferred: the `GET /wellntel/readings` proxy and the `sensor_type` +filter on `GET /thing`. Both are blocked on open questions the contract itself +raises — where the Wellntel API key lives and where the wellname→PointID mapping +belongs. The UI already falls back to demo data when they are absent. + +Also open, and unchanged by this work: whether the raw water-head series should +be retained alongside the corrected one, and whether publishing as `provisional` +should feed a review queue. diff --git a/domain/hydrograph.py b/domain/hydrograph.py new file mode 100644 index 000000000..20ad1c7ed --- /dev/null +++ b/domain/hydrograph.py @@ -0,0 +1,163 @@ +# =============================================================================== +# 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. +# =============================================================================== +""" +Rules for publishing and deleting corrected transducer series. + +The hydrograph corrector (OcotilloUI) uploads a whole corrected file as one +block. These functions decide the block's span, whether it collides with what is +already stored, which deployment it belongs to, and what survives a range +delete. They take plain values -- no session, no request -- so the awkward parts +(inclusive vs half-open overlap, a block narrowed to a single instant) can be +tested without a database. + +**Overlap is inclusive on both bounds here**, which differs from +``TransducerObservationBlock.overlaps`` on the model. That method is half-open, +so two blocks sharing an endpoint do not "overlap". The block *reader* +(``services.observation_helper.get_transducer_observations``) matches an +observation to a block with ``start <= t <= end``, so two blocks sharing an +endpoint both claim any reading at that instant and the reader picks whichever +sorts first. That ambiguity is the thing the publish conflict check exists to +prevent, so the check uses the reader's inclusive bounds rather than the +model's. +""" + +from datetime import date, datetime + +# One request per logger file is the expected shape; a 90-day file at a 6-hour +# cadence is 360 rows. The cap is three orders of magnitude above that, high +# enough never to reject real work and low enough that a runaway client cannot +# ask the server to build a million-row transaction. +MAX_MEASUREMENTS = 100_000 + + +class HydrographError(ValueError): + """Base for publish/delete rule violations. A ValueError, per ADR4.""" + + +def derive_block_span( + observation_datetimes: list[datetime], +) -> tuple[datetime, datetime]: + """ + The block's span is the extent of its readings. + + The client does not send ``start_datetime``/``end_datetime``: a span wider + than the data would claim coverage the block does not have, and the reader + would attach unrelated readings to it. + """ + if not observation_datetimes: + raise HydrographError("A block needs at least one measurement") + + return min(observation_datetimes), max(observation_datetimes) + + +def first_out_of_order_index(observation_datetimes: list[datetime]) -> int | None: + """ + Index of the first timestamp that does not advance on its predecessor. + + Strictly increasing, so a repeated timestamp is reported too -- the storage + constraint is one reading per deployment/parameter/instant, and a duplicate + inside one request would abort the whole transaction on insert rather than + be reported against the row that caused it. + + Returns the index so the caller can point at the offending row. + """ + for index in range(1, len(observation_datetimes)): + if observation_datetimes[index] <= observation_datetimes[index - 1]: + return index + + return None + + +def spans_overlap( + a_start: datetime, a_end: datetime, b_start: datetime, b_end: datetime +) -> bool: + """Whether two closed intervals share any instant. See the module docstring.""" + return not (a_end < b_start or a_start > b_end) + + +def resolve_deployment_id( + candidates: list[tuple[int, date | None, date | None]], + span_start: datetime, + span_end: datetime, +) -> int: + """ + Pick the deployment whose installation period covers a block's span. + + ``candidates`` are ``(deployment_id, installation_date, removal_date)`` for + one well. A NULL installation date reads as "always installed" and a NULL + removal date as "still installed" -- that is how the column is used, and + treating an unrecorded date as a closed boundary would exclude the + deployments most likely to be current. + + Ambiguity is an error rather than a choice: two overlapping deployments mean + two sensors could have produced the file, and guessing attributes readings to + hardware that did not record them. + """ + span_start_date = span_start.date() + span_end_date = span_end.date() + + covering = [ + deployment_id + for deployment_id, installation_date, removal_date in candidates + if (installation_date is None or installation_date <= span_start_date) + and (removal_date is None or removal_date >= span_end_date) + ] + + if not covering: + raise HydrographError( + f"No deployment covers {span_start_date} to {span_end_date}; " + "send deployment_id explicitly" + ) + if len(covering) > 1: + joined = ", ".join(str(deployment_id) for deployment_id in sorted(covering)) + raise HydrographError( + f"{len(covering)} deployments cover {span_start_date} to " + f"{span_end_date} ({joined}); send deployment_id explicitly" + ) + + return covering[0] + + +def narrowed_block_span( + surviving_datetimes: list[datetime], +) -> tuple[datetime, datetime] | None: + """ + What a block's span becomes after some of its readings are deleted. + + ``None`` means nothing survived and the block should go with them -- an + empty block is invisible to the reader and would only ever collide with a + later publish. + """ + if not surviving_datetimes: + return None + + return min(surviving_datetimes), max(surviving_datetimes) + + +def validate_delete_range(start_time: datetime, end_time: datetime) -> None: + """ + Reject a delete range that cannot be meant. + + Both bounds are required by the route signature; this covers the ordering. + An inverted range is not silently reordered, because the operation is + irreversible and a transposed pair is as likely to be the wrong pair as the + right one written backwards. + """ + if end_time <= start_time: + raise HydrographError("end_time must be after start_time") + + +# ============= EOF ============================================= diff --git a/schemas/transducer.py b/schemas/transducer.py index f11be79aa..a9c5c4142 100644 --- a/schemas/transducer.py +++ b/schemas/transducer.py @@ -14,10 +14,12 @@ # limitations under the License. # =============================================================================== from datetime import datetime +from typing import Literal -from pydantic import BaseModel +from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, field_validator -from core.enums import DataMaturity, ReviewStatus +from core.enums import DataMaturity, ReleaseStatus, ReviewStatus +from domain.hydrograph import MAX_MEASUREMENTS, first_out_of_order_index from schemas import BaseResponseModel, BaseCreateModel @@ -28,12 +30,22 @@ class TransducerObservationBlockResponse(BaseResponseModel): parameter_id: int # parameter: ParameterResponse + # Publish provenance. Nullable throughout: blocks loaded from the legacy + # AMPAPI transfer predate the corrector and state none of this. + source_file: str | None = None + source_kind: str | None = None + corrections: list[str] | None = None + comment: str | None = None + class TransducerObservationResponse(BaseResponseModel): value: float observation_datetime: datetime parameter_id: int deployment_id: int + # Set only where a correction moved the value, so NULL reads as + # "as measured" rather than "unknown". + note: str | None = None # Nullable: readings loaded before the field existed do not state a # maturity, and asserting one for them would be an invention. data_maturity: DataMaturity | None @@ -53,4 +65,118 @@ class CreateTransducerObservation(BaseCreateModel): data_maturity: DataMaturity | None = None +# ============= Hydrograph correction publish ==================== +# The corrector (OcotilloUI /ocotillo/hydrograph-correction) uploads one +# corrected logger file as one block. See +# docs/hydrograph-correction-publish.md. + + +class TransducerBlockProvenance(BaseModel): + """Where a corrected series came from and what was done to it.""" + + source_file: str = Field(max_length=255) + source_kind: Literal["water_head", "depth_to_water"] | None = None + # Free text in applied order, written by the workbench: "shift (-1.25 ft, + # ...)", "snap_to_manual (+0.42 ft to ..., collected by ...)". + corrections: list[str] = Field(default_factory=list) + notes: str | None = None + + +class CorrectedMeasurement(BaseModel): + """One reading of the corrected series, in feet below ground surface.""" + + # Aware, so a naive timestamp is rejected rather than guessed at. The + # workbench sends UTC; a logger file's local wall time silently read as UTC + # would shift a whole series by hours. + observation_datetime: AwareDatetime + value: float + note: str | None = None + + # NaN and infinity are not measurements. Without this they would validate + # as floats and land in the column. + model_config = ConfigDict(allow_inf_nan=False) + + +class PublishTransducerBlock(BaseCreateModel): + """A whole corrected file: one block plus every reading in it.""" + + thing_id: int + # Optional: resolved server-side from the block span when omitted, and 422 + # when that is ambiguous. See domain.hydrograph.resolve_deployment_id. + deployment_id: int | None = None + parameter_id: int + + # release_status comes from BaseCreateModel, defaulting to "draft". + review_status: ReviewStatus = "not reviewed" + + provenance: TransducerBlockProvenance + measurements: list[CorrectedMeasurement] = Field( + min_length=1, max_length=MAX_MEASUREMENTS + ) + + @field_validator("review_status", mode="before") + @classmethod + def coerce_review_status(cls, v): + if isinstance(v, str): + try: + return ReviewStatus(v) + except ValueError: + raise ValueError(f"Invalid review_status: {v}") + return v + + @field_validator("measurements") + @classmethod + def measurements_strictly_increasing(cls, measurements): + # Reported against the offending row's index so the UI can highlight + # it: the error path becomes + # ["body", "measurements", N, "observation_datetime"]. + index = first_out_of_order_index( + [measurement.observation_datetime for measurement in measurements] + ) + if index is not None: + raise ValueError( + f"measurements must be strictly increasing in time; row {index} " + f"({measurements[index].observation_datetime.isoformat()}) does not " + f"advance on row {index - 1} " + f"({measurements[index - 1].observation_datetime.isoformat()})" + ) + return measurements + + +class PublishedTransducerBlockResponse(BaseModel): + """ + Mirrors the read shape so the client can merge a publish straight into a + ``GET /observation/transducer-groundwater-level`` result set. The + observations are not echoed -- the client just sent them; the count is what + it cannot know. + """ + + block: TransducerObservationBlockResponse + observation_count: int + thing_id: int + deployment_id: int + + +class OverlappingBlock(BaseModel): + """An existing block a publish would collide with, named in the 409 body.""" + + id: int + start_datetime: datetime + end_datetime: datetime + review_status: ReviewStatus + release_status: ReleaseStatus + + +class DeletedTransducerObservationsResponse(BaseModel): + """ + What a range delete removed. ``updated_block_ids`` are blocks that kept + some readings and had their span narrowed to the survivors. + """ + + deleted_observation_count: int + deleted_block_ids: list[int] + updated_block_ids: list[int] + thing_id: int + + # ============= EOF ============================================= diff --git a/services/observation_helper.py b/services/observation_helper.py index 4e1cab5e6..1afe11158 100644 --- a/services/observation_helper.py +++ b/services/observation_helper.py @@ -6,9 +6,9 @@ from fastapi import Request, Query from fastapi_pagination.ext.sqlalchemy import paginate from pydantic import BaseModel -from sqlalchemy import select, desc +from sqlalchemy import asc, select, desc from sqlalchemy.orm import Session -from starlette.status import HTTP_404_NOT_FOUND +from starlette.status import HTTP_404_NOT_FOUND, HTTP_422_UNPROCESSABLE_CONTENT from db import ( Observation, @@ -187,11 +187,75 @@ def transformer(observations): return response_items - query = query.order_by(TransducerObservation.observation_datetime.desc()) + query = _sorted_transducer_query(query, sort, order) return paginate(query=query, conn=session, transformer=transformer) +# Only columns that mean something to a client of this endpoint. A whitelist +# rather than getattr on the model: the latter would expose every column, +# including the legacy `nma_*` ones, as a public sort key. +_TRANSDUCER_SORT_COLUMNS = { + "observation_datetime": TransducerObservation.observation_datetime, + "value": TransducerObservation.value, + "id": TransducerObservation.id, +} + +_TRANSDUCER_SORT_ORDERS = {"asc", "desc"} + + +def _sorted_transducer_query(query, sort: str | None, order: str | None): + """ + Apply `sort`/`order` to a transducer observation query. + + Defaults to newest first, which is what the list view wants and what a + client asking for the latest stored reading relies on. An unrecognised sort + field is rejected rather than ignored -- silently returning a differently + ordered page reads as data changing, not as a bad request. + """ + if sort and sort not in _TRANSDUCER_SORT_COLUMNS: + raise PydanticStyleException( + status_code=HTTP_422_UNPROCESSABLE_CONTENT, + detail=[ + { + "loc": ["query", "sort"], + "msg": ( + f"Cannot sort by '{sort}'. Valid fields: " + f"{', '.join(sorted(_TRANSDUCER_SORT_COLUMNS))}" + ), + "type": "value_error", + "input": sort, + } + ], + ) + + normalized_order = (order or "desc").lower() + if normalized_order not in _TRANSDUCER_SORT_ORDERS: + # Rejected for the same reason an unknown sort field is: anything other + # than `asc` used to fall through to descending, so `order=ascending` + # returned 200 with the opposite of what was asked for. + raise PydanticStyleException( + status_code=HTTP_422_UNPROCESSABLE_CONTENT, + detail=[ + { + "loc": ["query", "order"], + "msg": ( + f"Cannot order by '{order}'. Valid values: " + f"{', '.join(sorted(_TRANSDUCER_SORT_ORDERS))}" + ), + "type": "value_error", + "input": order, + } + ], + ) + + column = _TRANSDUCER_SORT_COLUMNS.get( + sort or "observation_datetime", TransducerObservation.observation_datetime + ) + + return query.order_by(asc(column) if normalized_order == "asc" else desc(column)) + + def get_observations( request: Request, session: Session, diff --git a/services/transducer_helper.py b/services/transducer_helper.py new file mode 100644 index 000000000..99a2e6b5d --- /dev/null +++ b/services/transducer_helper.py @@ -0,0 +1,503 @@ +# =============================================================================== +# 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. +# =============================================================================== +""" +Publish and range-delete for corrected transducer series. + +Orchestration only: load the well, deployment, and colliding blocks, hand the +decisions to ``domain.hydrograph``, persist the result. See +``docs/hydrograph-correction-publish.md`` for the endpoint contract. +""" + +from datetime import datetime + +from sqlalchemy import delete, insert, select, text, update +from sqlalchemy.orm import Session +from starlette.status import ( + HTTP_404_NOT_FOUND, + HTTP_409_CONFLICT, + HTTP_422_UNPROCESSABLE_CONTENT, +) + +from db import Thing +from db.deployment import Deployment +from db.transducer import TransducerObservation, TransducerObservationBlock +from domain.hydrograph import ( + HydrographError, + derive_block_span, + narrowed_block_span, + resolve_deployment_id, + validate_delete_range, +) +from schemas.transducer import ( + DeletedTransducerObservationsResponse, + OverlappingBlock, + PublishedTransducerBlockResponse, + TransducerObservationBlockResponse, +) +from services.exceptions_helper import PydanticStyleException + +# A published block has not been reviewed by anyone yet, which on USGS terms is +# exactly `provisional`. Once a reviewer marks the block `approved` the readings +# follow. Derived rather than sent by the client so the two axes cannot be set +# to contradict each other on the way in. +_MATURITY_FOR_REVIEW_STATUS = {"approved": "approved"} +_DEFAULT_MATURITY = "provisional" + + +def _not_found(field: str, value, message: str): + return PydanticStyleException( + status_code=HTTP_404_NOT_FOUND, + detail=[ + { + "loc": ["body", field], + "msg": message, + "type": "value_error", + "input": value, + } + ], + ) + + +def _unprocessable(loc: list, value, message: str): + return PydanticStyleException( + status_code=HTTP_422_UNPROCESSABLE_CONTENT, + detail=[ + { + "loc": loc, + "msg": message, + "type": "value_error", + "input": value, + } + ], + ) + + +def _enum_value(value): + """Unwrap a lexicon-backed enum member to the term the column stores.""" + return getattr(value, "value", value) + + +def _lock_series(session: Session, thing_id: int, parameter_id: int) -> None: + """ + Serialize writers against one well's series for the rest of the transaction. + + Both the publish overlap check and the delete reconciliation read state, + decide, and then write -- and neither decision survives a concurrent writer. + Two publishes with different timestamps but overlapping spans each see no + existing block and both commit, because the unique constraints only catch + identical spans and identical readings; the inclusive reader then has two + blocks claiming the same instants. Two deletes each compute survivors from + a snapshot the other is invalidating, and the later update can widen a block + back over readings the earlier one removed. + + An advisory lock rather than row locks: on publish there is no row to lock + yet -- the conflict is with a block that does not exist -- so the thing being + guarded is the (well, parameter) series itself, not any row. Transaction + scoped, so it releases on commit or rollback with no unlock path to forget. + Publish and delete take the same key in the same order, so they serialize + against each other and cannot deadlock against one another. + """ + session.execute( + text("SELECT pg_advisory_xact_lock(:thing_id, :parameter_id)"), + {"thing_id": thing_id, "parameter_id": parameter_id}, + ) + + +def _deployment_ids_for_thing(session: Session, thing_id: int) -> list[int]: + return list( + session.scalars( + select(Deployment.id).where(Deployment.thing_id == thing_id) + ).all() + ) + + +def _overlapping_blocks( + session: Session, + thing_id: int, + parameter_id: int, + span_start: datetime, + span_end: datetime, +) -> list[TransducerObservationBlock]: + """ + Blocks whose closed span shares an instant with ``[span_start, span_end]``. + + Inclusive on both ends, matching the block reader rather than + ``TransducerObservationBlock.overlaps``. See ``domain.hydrograph``. + """ + return list( + session.scalars( + select(TransducerObservationBlock) + .where( + TransducerObservationBlock.thing_id == thing_id, + TransducerObservationBlock.parameter_id == parameter_id, + TransducerObservationBlock.start_datetime <= span_end, + TransducerObservationBlock.end_datetime >= span_start, + ) + .order_by(TransducerObservationBlock.start_datetime) + ).all() + ) + + +def publish_transducer_block( + session: Session, + payload, + parameter_id: int, + user=None, + replace_overlapping: bool = False, +) -> PublishedTransducerBlockResponse: + """ + Create one block and all of its readings, or nothing. + + The block's span is derived from the readings -- a client-supplied span + wider than the data would attach unrelated readings to the block. + + ``parameter_id`` is the parameter this route is scoped to, resolved by the + caller; the payload's own ``parameter_id`` is checked against it rather than + trusted. + """ + thing = session.get(Thing, payload.thing_id) + if thing is None: + raise _not_found( + "thing_id", payload.thing_id, f"Thing {payload.thing_id} not found" + ) + + # The read and delete routes on this path resolve the groundwater level + # parameter themselves, so a block published under any other parameter would + # be invisible to both -- a 201 for data that can then never be listed or + # removed here. The field stays in the request because the contract has the + # client state it explicitly rather than inherit a server-side default; it + # is validated, not obeyed. + if payload.parameter_id != parameter_id: + raise _unprocessable( + ["body", "parameter_id"], + payload.parameter_id, + f"This route publishes parameter {parameter_id} only; the read and " + f"delete routes on this path would not see parameter " + f"{payload.parameter_id}", + ) + + span_start, span_end = derive_block_span( + [measurement.observation_datetime for measurement in payload.measurements] + ) + + deployment_id = _resolve_deployment(session, payload, span_start, span_end) + + # Everything from here reads state and then writes based on it, so no other + # writer may touch this series until the transaction ends. + _lock_series(session, payload.thing_id, payload.parameter_id) + + existing = _overlapping_blocks( + session, payload.thing_id, payload.parameter_id, span_start, span_end + ) + if existing and not replace_overlapping: + raise _overlap_conflict(existing) + + if existing: + _delete_superseded(session, payload.thing_id, payload.parameter_id, existing) + + # Readings can outlive the block that covered them -- nothing links the two + # tables, so a block deleted by hand leaves its observations behind, where + # the reader ignores them but the storage constraint still sees them. Insert + # would abort the transaction on the first collision with a message naming + # the constraint, so check first and say what is actually in the way. + _reject_colliding_observations( + session, deployment_id, payload.parameter_id, span_start, span_end + ) + + block = TransducerObservationBlock( + thing_id=payload.thing_id, + parameter_id=payload.parameter_id, + review_status=_enum_value(payload.review_status), + release_status=_enum_value(payload.release_status), + start_datetime=span_start, + end_datetime=span_end, + source_file=payload.provenance.source_file, + source_kind=payload.provenance.source_kind, + corrections=payload.provenance.corrections or None, + comment=payload.provenance.notes, + ) + _stamp_created_by(block, user) + session.add(block) + session.flush() + + review_status = _enum_value(payload.review_status) + data_maturity = _MATURITY_FOR_REVIEW_STATUS.get(review_status, _DEFAULT_MATURITY) + release_status = _enum_value(payload.release_status) + created_by_id, created_by_name = _created_by(user) + + rows = [ + { + "parameter_id": payload.parameter_id, + "deployment_id": deployment_id, + "observation_datetime": measurement.observation_datetime, + "value": measurement.value, + "note": measurement.note, + "data_maturity": data_maturity, + "release_status": release_status, + "created_by_id": created_by_id, + "created_by_name": created_by_name, + } + for measurement in payload.measurements + ] + session.execute(insert(TransducerObservation), rows) + + session.commit() + session.refresh(block) + + return PublishedTransducerBlockResponse( + block=TransducerObservationBlockResponse.model_validate(block), + observation_count=len(rows), + thing_id=payload.thing_id, + deployment_id=deployment_id, + ) + + +def _resolve_deployment(session, payload, span_start, span_end) -> int: + if payload.deployment_id is not None: + deployment = session.get(Deployment, payload.deployment_id) + if deployment is None: + raise _not_found( + "deployment_id", + payload.deployment_id, + f"Deployment {payload.deployment_id} not found", + ) + if deployment.thing_id != payload.thing_id: + raise _unprocessable( + ["body", "deployment_id"], + payload.deployment_id, + f"Deployment {payload.deployment_id} belongs to thing " + f"{deployment.thing_id}, not {payload.thing_id}", + ) + return payload.deployment_id + + candidates = session.execute( + select( + Deployment.id, Deployment.installation_date, Deployment.removal_date + ).where(Deployment.thing_id == payload.thing_id) + ).all() + + try: + return resolve_deployment_id( + [tuple(row) for row in candidates], span_start, span_end + ) + except HydrographError as err: + raise _unprocessable(["body", "deployment_id"], None, str(err)) + + +def _overlap_conflict(blocks: list[TransducerObservationBlock]): + overlapping = [ + OverlappingBlock.model_validate(block, from_attributes=True).model_dump( + mode="json" + ) + for block in blocks + ] + ids = ", ".join(str(block.id) for block in blocks) + return PydanticStyleException( + status_code=HTTP_409_CONFLICT, + detail=[ + { + "loc": ["body", "measurements"], + "msg": ( + f"Time span overlaps existing block(s) {ids}. Retry with " + "?replace_overlapping=true to supersede them." + ), + "type": "value_error", + "input": {"overlapping_blocks": overlapping}, + } + ], + ) + + +def _delete_superseded( + session: Session, + thing_id: int, + parameter_id: int, + blocks: list[TransducerObservationBlock], +) -> None: + """ + Drop the blocks a replacing publish supersedes, and their readings. + + The readings go too. Keeping them would leave rows the reader cannot show + (no block covers them) that still occupy the deployment/parameter/instant + the new series is about to claim -- so "replace" that kept them would fail + on the very insert it was asked to make room for. + """ + deployment_ids = _deployment_ids_for_thing(session, thing_id) + if deployment_ids: + for block in blocks: + session.execute( + delete(TransducerObservation).where( + TransducerObservation.deployment_id.in_(deployment_ids), + TransducerObservation.parameter_id == parameter_id, + TransducerObservation.observation_datetime >= block.start_datetime, + TransducerObservation.observation_datetime <= block.end_datetime, + ) + ) + + session.execute( + delete(TransducerObservationBlock).where( + TransducerObservationBlock.id.in_([block.id for block in blocks]) + ) + ) + session.flush() + + +def _reject_colliding_observations( + session: Session, + deployment_id: int, + parameter_id: int, + span_start: datetime, + span_end: datetime, +) -> None: + collisions = session.scalar( + select(TransducerObservation.observation_datetime) + .where( + TransducerObservation.deployment_id == deployment_id, + TransducerObservation.parameter_id == parameter_id, + TransducerObservation.observation_datetime >= span_start, + TransducerObservation.observation_datetime <= span_end, + ) + .order_by(TransducerObservation.observation_datetime) + .limit(1) + ) + if collisions is None: + return + + raise PydanticStyleException( + status_code=HTTP_409_CONFLICT, + detail=[ + { + "loc": ["body", "measurements"], + "msg": ( + f"Deployment {deployment_id} already has readings in this " + f"time span (earliest {collisions.isoformat()}) that no block " + "covers. Delete them by range before publishing." + ), + "type": "value_error", + "input": {"deployment_id": deployment_id}, + } + ], + ) + + +def delete_transducer_observations( + session: Session, + thing_id: int, + parameter_id: int, + start_time: datetime, + end_time: datetime, +) -> DeletedTransducerObservationsResponse: + """ + Delete every reading for a well inside a closed time range, then reconcile + the blocks that covered them. + + Scoped exactly like the ``GET`` on the same path, so the set a client + previews is the set this removes. There is deliberately no unbounded form. + """ + thing = session.get(Thing, thing_id) + if thing is None: + raise _not_found("thing_id", thing_id, f"Thing {thing_id} not found") + + try: + validate_delete_range(start_time, end_time) + except HydrographError as err: + raise _unprocessable(["query", "end_time"], end_time.isoformat(), str(err)) + + _lock_series(session, thing_id, parameter_id) + + deployment_ids = _deployment_ids_for_thing(session, thing_id) + if not deployment_ids: + return DeletedTransducerObservationsResponse( + deleted_observation_count=0, + deleted_block_ids=[], + updated_block_ids=[], + thing_id=thing_id, + ) + + # Read the affected blocks before deleting: after the readings are gone + # there is nothing left to identify which blocks covered them. + affected_blocks = _overlapping_blocks( + session, thing_id, parameter_id, start_time, end_time + ) + + deleted = session.execute( + delete(TransducerObservation).where( + TransducerObservation.deployment_id.in_(deployment_ids), + TransducerObservation.parameter_id == parameter_id, + TransducerObservation.observation_datetime >= start_time, + TransducerObservation.observation_datetime <= end_time, + ) + ) + deleted_observation_count = deleted.rowcount or 0 + session.flush() + + deleted_block_ids: list[int] = [] + updated_block_ids: list[int] = [] + + for block in affected_blocks: + surviving = list( + session.scalars( + select(TransducerObservation.observation_datetime).where( + TransducerObservation.deployment_id.in_(deployment_ids), + TransducerObservation.parameter_id == parameter_id, + TransducerObservation.observation_datetime >= block.start_datetime, + TransducerObservation.observation_datetime <= block.end_datetime, + ) + ).all() + ) + span = narrowed_block_span(surviving) + + if span is None: + deleted_block_ids.append(block.id) + session.execute( + delete(TransducerObservationBlock).where( + TransducerObservationBlock.id == block.id + ) + ) + continue + + new_start, new_end = span + if (new_start, new_end) != (block.start_datetime, block.end_datetime): + updated_block_ids.append(block.id) + session.execute( + update(TransducerObservationBlock) + .where(TransducerObservationBlock.id == block.id) + .values(start_datetime=new_start, end_datetime=new_end) + ) + + session.commit() + + return DeletedTransducerObservationsResponse( + deleted_observation_count=deleted_observation_count, + deleted_block_ids=deleted_block_ids, + updated_block_ids=updated_block_ids, + thing_id=thing_id, + ) + + +def _created_by(user) -> tuple[str | None, str | None]: + if isinstance(user, dict): + return user.get("sub"), user.get("name") + return None, None + + +def _stamp_created_by(obj, user) -> None: + created_by_id, created_by_name = _created_by(user) + obj.created_by_id = created_by_id + obj.created_by_name = created_by_name + + +# ============= EOF ============================================= diff --git a/tests/test_authorization.py b/tests/test_authorization.py index 97608ae9c..02ce5caee 100644 --- a/tests/test_authorization.py +++ b/tests/test_authorization.py @@ -68,6 +68,7 @@ dependencies.amp_admin_function, dependencies.amp_editor_function, dependencies.amp_viewer_function, + dependencies.amp_staging_function, dependencies.lexicon_admin_function, dependencies.lexicon_editor_function, dependencies.no_permission_function, diff --git a/tests/test_domain_hydrograph.py b/tests/test_domain_hydrograph.py new file mode 100644 index 000000000..70e857e13 --- /dev/null +++ b/tests/test_domain_hydrograph.py @@ -0,0 +1,179 @@ +# =============================================================================== +# 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. +# =============================================================================== +"""Rules behind publishing and range-deleting corrected transducer series.""" + +from datetime import date, datetime, timedelta, timezone + +import pytest + +from domain.hydrograph import ( + HydrographError, + derive_block_span, + first_out_of_order_index, + narrowed_block_span, + resolve_deployment_id, + spans_overlap, + validate_delete_range, +) + +T0 = datetime(2025, 1, 15, tzinfo=timezone.utc) + + +def _times(*hours): + return [T0 + timedelta(hours=h) for h in hours] + + +# -------------------------------------------------------------------------- +# derive_block_span +# -------------------------------------------------------------------------- +def test_span_is_the_extent_of_the_readings(): + assert derive_block_span(_times(0, 6, 12)) == (T0, T0 + timedelta(hours=12)) + + +def test_span_does_not_assume_the_readings_arrived_sorted(): + assert derive_block_span(_times(12, 0, 6)) == (T0, T0 + timedelta(hours=12)) + + +def test_a_single_reading_gives_a_zero_width_span(): + # Allowed on purpose: the block check constraint is `end >= start` and the + # reader matches inclusively, so one reading still gets covered. + assert derive_block_span(_times(0)) == (T0, T0) + + +def test_empty_series_is_rejected(): + with pytest.raises(HydrographError): + derive_block_span([]) + + +# -------------------------------------------------------------------------- +# first_out_of_order_index +# -------------------------------------------------------------------------- +def test_increasing_series_is_in_order(): + assert first_out_of_order_index(_times(0, 6, 12)) is None + + +def test_going_backwards_is_reported_at_the_offending_row(): + assert first_out_of_order_index(_times(0, 12, 6)) == 2 + + +def test_a_repeated_timestamp_is_reported_too(): + # Not merely untidy: two readings at one instant collide on the + # deployment/parameter/datetime constraint and would abort the insert. + assert first_out_of_order_index(_times(0, 6, 6)) == 2 + + +def test_a_single_row_cannot_be_out_of_order(): + assert first_out_of_order_index(_times(0)) is None + + +# -------------------------------------------------------------------------- +# spans_overlap +# -------------------------------------------------------------------------- +def test_disjoint_spans_do_not_overlap(): + assert not spans_overlap( + T0, T0 + timedelta(1), T0 + timedelta(2), T0 + timedelta(3) + ) + + +def test_nested_span_overlaps(): + assert spans_overlap( + T0 + timedelta(1), T0 + timedelta(2), T0, T0 + timedelta(days=10) + ) + + +def test_spans_touching_at_one_instant_overlap(): + # The distinguishing case. `TransducerObservationBlock.overlaps` is + # half-open and would call this clear; the reader is inclusive, so both + # blocks would claim a reading at the shared instant. + assert spans_overlap(T0, T0 + timedelta(1), T0 + timedelta(1), T0 + timedelta(2)) + + +# -------------------------------------------------------------------------- +# resolve_deployment_id +# -------------------------------------------------------------------------- +SPAN_START = datetime(2025, 3, 1, tzinfo=timezone.utc) +SPAN_END = datetime(2025, 3, 31, tzinfo=timezone.utc) + + +def test_the_one_covering_deployment_is_chosen(): + candidates = [ + (1, date(2020, 1, 1), date(2021, 1, 1)), + (2, date(2025, 1, 1), None), + ] + assert resolve_deployment_id(candidates, SPAN_START, SPAN_END) == 2 + + +def test_a_null_installation_date_reads_as_always_installed(): + assert resolve_deployment_id([(7, None, None)], SPAN_START, SPAN_END) == 7 + + +def test_a_deployment_removed_mid_span_does_not_cover_it(): + with pytest.raises(HydrographError, match="No deployment covers"): + resolve_deployment_id( + [(1, date(2025, 1, 1), date(2025, 3, 15))], SPAN_START, SPAN_END + ) + + +def test_no_deployments_at_all_is_an_error_naming_the_way_out(): + with pytest.raises(HydrographError, match="send deployment_id explicitly"): + resolve_deployment_id([], SPAN_START, SPAN_END) + + +def test_two_covering_deployments_are_ambiguous_rather_than_a_coin_flip(): + with pytest.raises(HydrographError, match="2 deployments cover"): + resolve_deployment_id( + [(1, date(2024, 1, 1), None), (2, None, None)], SPAN_START, SPAN_END + ) + + +# -------------------------------------------------------------------------- +# narrowed_block_span +# -------------------------------------------------------------------------- +def test_a_block_with_nothing_left_is_marked_for_deletion(): + assert narrowed_block_span([]) is None + + +def test_a_block_narrows_to_what_survived(): + assert narrowed_block_span(_times(6, 12)) == ( + T0 + timedelta(hours=6), + T0 + timedelta(hours=12), + ) + + +def test_a_block_down_to_one_reading_becomes_zero_width(): + assert narrowed_block_span(_times(6)) == ( + T0 + timedelta(hours=6), + T0 + timedelta(hours=6), + ) + + +# -------------------------------------------------------------------------- +# validate_delete_range +# -------------------------------------------------------------------------- +def test_a_forward_range_is_accepted(): + assert validate_delete_range(T0, T0 + timedelta(1)) is None + + +def test_an_inverted_range_is_rejected_not_reordered(): + # The operation is irreversible; a transposed pair is as likely to be the + # wrong pair as the right one written backwards. + with pytest.raises(HydrographError, match="after start_time"): + validate_delete_range(T0 + timedelta(1), T0) + + +def test_a_zero_width_range_is_rejected(): + with pytest.raises(HydrographError): + validate_delete_range(T0, T0) diff --git a/tests/test_transducer_publish.py b/tests/test_transducer_publish.py new file mode 100644 index 000000000..44e325b06 --- /dev/null +++ b/tests/test_transducer_publish.py @@ -0,0 +1,675 @@ +# =============================================================================== +# 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 hydrograph corrector's publish and range-delete endpoints. + +Both are gated on `AMP.Staging`, which nobody holds in Authentik yet -- these +tests override that dependency, so they cover the behaviour, not the grant. +""" + +import threading +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import select, text + +from core.dependencies import amp_staging_function, amp_viewer_function +from db import Deployment, Sensor, Thing, TransducerObservation +from db.engine import engine, session_ctx +from db.transducer import TransducerObservationBlock +from main import app +from tests import client, get_parameter_id, override_authentication + +PUBLISH_URL = "/observation/transducer-groundwater-level/block" +READ_URL = "/observation/transducer-groundwater-level" + +T0 = datetime(2025, 1, 15, tzinfo=timezone.utc) + + +def _groundwater_level_parameter_id() -> int: + return get_parameter_id("groundwater level", "Field Parameter") + + +@pytest.fixture(scope="module", autouse=True) +def override_authentication_dependency_fixture(): + app.dependency_overrides[amp_staging_function] = override_authentication( + default={"name": "foobar", "sub": "1234567890"} + ) + app.dependency_overrides[amp_viewer_function] = override_authentication() + + yield + + app.dependency_overrides = {} + + +@pytest.fixture() +def published_well(): + """A well with one deployment and nothing stored yet.""" + with session_ctx() as session: + thing = Thing( + name=f"Hydrograph Publish Well {datetime.now().timestamp()}", + first_visit_date="2023-03-03", + thing_type="water well", + release_status="draft", + well_depth=200, + hole_depth=200, + well_casing_diameter=5.0, + well_casing_depth=200.0, + ) + sensor = Sensor( + name=f"Hydrograph Publish Sensor {datetime.now().timestamp()}", + sensor_type="Pressure Transducer", + model="Model X", + serial_no=f"serial-{datetime.now().timestamp()}", + pcn_number=f"pcn-{datetime.now().timestamp()}", + owner_agency="NMBGMR", + sensor_status="In Service", + release_status="draft", + ) + session.add_all([thing, sensor]) + session.flush() + + deployment = Deployment( + sensor_id=sensor.id, + thing_id=thing.id, + installation_date="2020-01-01", + removal_date=None, + recording_interval=6, + recording_interval_units="hour", + ) + session.add(deployment) + session.commit() + + thing_id, deployment_id, sensor_id = thing.id, deployment.id, sensor.id + + yield thing_id, deployment_id + + with session_ctx() as session: + deployment_ids = session.scalars( + select(Deployment.id).where(Deployment.thing_id == thing_id) + ).all() + if deployment_ids: + for observation in session.scalars( + select(TransducerObservation).where( + TransducerObservation.deployment_id.in_(deployment_ids) + ) + ).all(): + session.delete(observation) + for block in session.scalars( + select(TransducerObservationBlock).where( + TransducerObservationBlock.thing_id == thing_id + ) + ).all(): + session.delete(block) + session.flush() + for model, pk in ((Deployment, deployment_id), (Thing, thing_id)): + obj = session.get(model, pk) + if obj is not None: + session.delete(obj) + session.flush() + sensor = session.get(Sensor, sensor_id) + if sensor is not None: + session.delete(sensor) + session.commit() + + +def _payload(thing_id, hours=(0, 6, 12), **overrides): + payload = { + "thing_id": thing_id, + "parameter_id": _groundwater_level_parameter_id(), + "release_status": "provisional", + "review_status": "not reviewed", + "provenance": { + "source_file": "SO-0167_20250115.csv", + "source_kind": "water_head", + "corrections": ["convert_water_head (drift corrected)"], + "notes": "Snapped to 2025-01-15 manual measurement.", + }, + "measurements": [ + { + "observation_datetime": (T0 + timedelta(hours=h)).isoformat(), + "value": 42.5 + index * 0.01, + } + for index, h in enumerate(hours) + ], + } + payload.update(overrides) + return payload + + +# -------------------------------------------------------------------------- +# publish +# -------------------------------------------------------------------------- +def test_publish_creates_one_block_and_all_of_its_readings(published_well): + thing_id, deployment_id = published_well + + response = client.post(PUBLISH_URL, json=_payload(thing_id)) + + assert response.status_code == 201, response.text + body = response.json() + assert body["observation_count"] == 3 + assert body["thing_id"] == thing_id + # Deployment resolved server-side: the payload never named one. + assert body["deployment_id"] == deployment_id + + block = body["block"] + assert block["release_status"] == "provisional" + assert block["review_status"] == "not reviewed" + assert block["source_file"] == "SO-0167_20250115.csv" + assert block["corrections"] == ["convert_water_head (drift corrected)"] + assert block["comment"] == "Snapped to 2025-01-15 manual measurement." + + # Span is derived from the data, not sent by the client. + assert block["start_datetime"] == "2025-01-15T00:00:00Z" + assert block["end_datetime"] == "2025-01-15T12:00:00Z" + + +def test_published_readings_come_back_from_the_read_endpoint(published_well): + thing_id, _ = published_well + client.post(PUBLISH_URL, json=_payload(thing_id)) + + response = client.get(READ_URL, params={"thing_id": thing_id}) + + assert response.status_code == 200 + items = response.json()["items"] + assert len(items) == 3 + # Newest first by default, so a client can ask for the latest with size 1. + assert items[0]["observation"]["observation_datetime"] == "2025-01-15T12:00:00Z" + # Unreviewed on publish is provisional on USGS terms. + assert items[0]["observation"]["data_maturity"] == "provisional" + + +def test_per_reading_notes_are_persisted(published_well): + thing_id, _ = published_well + payload = _payload(thing_id) + payload["measurements"][1]["note"] = "spurious reflection removed" + + client.post(PUBLISH_URL, json=payload) + + response = client.get(READ_URL, params={"thing_id": thing_id, "order": "asc"}) + observations = [item["observation"] for item in response.json()["items"]] + assert observations[0]["note"] is None + assert observations[1]["note"] == "spurious reflection removed" + + +def test_a_single_reading_publishes_as_a_zero_width_block(published_well): + thing_id, _ = published_well + + response = client.post(PUBLISH_URL, json=_payload(thing_id, hours=(0,))) + + assert response.status_code == 201, response.text + block = response.json()["block"] + assert block["start_datetime"] == block["end_datetime"] + + +def test_overlapping_publish_is_rejected_and_names_the_blocks(published_well): + thing_id, _ = published_well + first = client.post(PUBLISH_URL, json=_payload(thing_id)) + first_block_id = first.json()["block"]["id"] + + response = client.post(PUBLISH_URL, json=_payload(thing_id, hours=(6, 18))) + + assert response.status_code == 409 + detail = response.json()["detail"][0] + assert str(first_block_id) in detail["msg"] + assert detail["input"]["overlapping_blocks"][0]["id"] == first_block_id + + +def test_replace_overlapping_supersedes_the_old_block_and_its_readings( + published_well, +): + thing_id, _ = published_well + first = client.post(PUBLISH_URL, json=_payload(thing_id)) + first_block_id = first.json()["block"]["id"] + + response = client.post( + PUBLISH_URL, + params={"replace_overlapping": "true"}, + json=_payload(thing_id, hours=(6, 18)), + ) + + assert response.status_code == 201, response.text + assert response.json()["block"]["id"] != first_block_id + + with session_ctx() as session: + assert session.get(TransducerObservationBlock, first_block_id) is None + + # Only the replacing series survives -- the superseded readings went with + # their block rather than being left where no block covers them. + items = client.get(READ_URL, params={"thing_id": thing_id}).json()["items"] + assert len(items) == 2 + + +def test_publish_is_atomic_when_a_reading_collides(published_well): + thing_id, _ = published_well + client.post(PUBLISH_URL, json=_payload(thing_id)) + + # Delete the block but leave its readings, which is what a hand-deleted + # block leaves behind: rows the reader ignores but storage still holds. + with session_ctx() as session: + for block in session.scalars( + select(TransducerObservationBlock).where( + TransducerObservationBlock.thing_id == thing_id + ) + ).all(): + session.delete(block) + session.commit() + + response = client.post(PUBLISH_URL, json=_payload(thing_id)) + + assert response.status_code == 409 + assert "no block covers" in response.json()["detail"][0]["msg"] + + with session_ctx() as session: + blocks = session.scalars( + select(TransducerObservationBlock).where( + TransducerObservationBlock.thing_id == thing_id + ) + ).all() + assert blocks == [] + + +def test_unknown_thing_is_a_404(published_well): + response = client.post(PUBLISH_URL, json=_payload(-1)) + assert response.status_code == 404 + + +def test_out_of_order_measurements_point_at_the_offending_row(published_well): + thing_id, _ = published_well + payload = _payload(thing_id, hours=(0, 12, 6)) + + response = client.post(PUBLISH_URL, json=payload) + + assert response.status_code == 422 + assert any( + error["loc"][:3] == ["body", "measurements"] + for error in response.json()["detail"] + ) + + +def test_naive_timestamps_are_rejected(published_well): + thing_id, _ = published_well + payload = _payload(thing_id) + payload["measurements"][0]["observation_datetime"] = "2025-01-15T00:00:00" + + response = client.post(PUBLISH_URL, json=payload) + + assert response.status_code == 422 + + +def test_an_empty_series_is_rejected(published_well): + thing_id, _ = published_well + response = client.post(PUBLISH_URL, json=_payload(thing_id, hours=())) + assert response.status_code == 422 + + +def test_a_deployment_on_another_well_is_rejected(published_well, sensor): + thing_id, _ = published_well + with session_ctx() as session: + other = Thing( + name=f"Hydrograph Other Well {datetime.now().timestamp()}", + first_visit_date="2023-03-03", + thing_type="water well", + release_status="draft", + ) + session.add(other) + session.flush() + other_deployment = Deployment( + sensor_id=sensor.id, thing_id=other.id, installation_date="2020-01-01" + ) + session.add(other_deployment) + session.commit() + other_deployment_id, other_thing_id = other_deployment.id, other.id + + try: + response = client.post( + PUBLISH_URL, + json=_payload(thing_id, deployment_id=other_deployment_id), + ) + assert response.status_code == 422 + assert "belongs to thing" in response.json()["detail"][0]["msg"] + finally: + with session_ctx() as session: + session.delete(session.get(Deployment, other_deployment_id)) + session.flush() + session.delete(session.get(Thing, other_thing_id)) + session.commit() + + +# -------------------------------------------------------------------------- +# range delete +# -------------------------------------------------------------------------- +def test_deleting_the_whole_span_removes_the_block_too(published_well): + thing_id, _ = published_well + block_id = client.post(PUBLISH_URL, json=_payload(thing_id)).json()["block"]["id"] + + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": thing_id, + "start_time": T0.isoformat(), + "end_time": (T0 + timedelta(hours=12)).isoformat(), + }, + ) + + assert response.status_code == 200, response.text + body = response.json() + assert body["deleted_observation_count"] == 3 + assert body["deleted_block_ids"] == [block_id] + assert body["updated_block_ids"] == [] + assert client.get(READ_URL, params={"thing_id": thing_id}).json()["items"] == [] + + +def test_a_partial_delete_narrows_the_block_to_the_survivors(published_well): + thing_id, _ = published_well + block_id = client.post(PUBLISH_URL, json=_payload(thing_id)).json()["block"]["id"] + + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": thing_id, + "start_time": (T0 + timedelta(hours=6)).isoformat(), + "end_time": (T0 + timedelta(hours=12)).isoformat(), + }, + ) + + assert response.status_code == 200, response.text + body = response.json() + assert body["deleted_observation_count"] == 2 + assert body["deleted_block_ids"] == [] + assert body["updated_block_ids"] == [block_id] + + with session_ctx() as session: + block = session.get(TransducerObservationBlock, block_id) + assert block.start_datetime == T0 + assert block.end_datetime == T0 + + # The survivor is still readable, which it would not be if the narrowed + # block no longer covered it. + items = client.get(READ_URL, params={"thing_id": thing_id}).json()["items"] + assert len(items) == 1 + + +def test_an_inverted_delete_range_is_rejected(published_well): + thing_id, _ = published_well + + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": thing_id, + "start_time": (T0 + timedelta(hours=12)).isoformat(), + "end_time": T0.isoformat(), + }, + ) + + assert response.status_code == 422 + + +def test_delete_without_a_bound_is_rejected(published_well): + thing_id, _ = published_well + + response = client.request("DELETE", READ_URL, params={"thing_id": thing_id}) + + assert response.status_code == 422 + + +def test_delete_for_an_unknown_well_is_a_404(): + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": -1, + "start_time": T0.isoformat(), + "end_time": (T0 + timedelta(hours=12)).isoformat(), + }, + ) + + assert response.status_code == 404 + + +# -------------------------------------------------------------------------- +# read ordering +# -------------------------------------------------------------------------- +def test_ascending_order_is_honoured(published_well): + thing_id, _ = published_well + client.post(PUBLISH_URL, json=_payload(thing_id)) + + items = client.get(READ_URL, params={"thing_id": thing_id, "order": "asc"}).json()[ + "items" + ] + + assert items[0]["observation"]["observation_datetime"] == "2025-01-15T00:00:00Z" + + +def test_an_unknown_sort_field_is_rejected_rather_than_ignored(published_well): + thing_id, _ = published_well + + response = client.get(READ_URL, params={"thing_id": thing_id, "sort": "nonsense"}) + + assert response.status_code == 422 + + +def test_an_unknown_order_is_rejected_rather_than_silently_descending(published_well): + # `ascending` is the obvious near-miss for `asc`, and it used to return 200 + # with the rows in exactly the opposite order to the one asked for. + thing_id, _ = published_well + + response = client.get(READ_URL, params={"thing_id": thing_id, "order": "ascending"}) + + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["query", "order"] + + +# -------------------------------------------------------------------------- +# deployment resolution at the service boundary +# +# domain.hydrograph covers the rule itself; these cover the translation of its +# verdicts into responses, and the path where the client names a deployment +# and the rule never runs. +# -------------------------------------------------------------------------- +def test_an_explicitly_named_deployment_is_used_as_given(published_well): + thing_id, deployment_id = published_well + + response = client.post( + PUBLISH_URL, json=_payload(thing_id, deployment_id=deployment_id) + ) + + assert response.status_code == 201, response.text + assert response.json()["deployment_id"] == deployment_id + + +def test_unknown_deployment_is_a_404(published_well): + thing_id, _ = published_well + + response = client.post(PUBLISH_URL, json=_payload(thing_id, deployment_id=-1)) + + assert response.status_code == 404 + assert response.json()["detail"][0]["loc"] == ["body", "deployment_id"] + + +def test_a_parameter_this_route_does_not_serve_is_rejected(published_well): + # A block published under another parameter would be invisible to the read + # and delete routes on this path, which resolve groundwater level + # themselves -- a 201 for data that could then never be listed or removed + # here. + thing_id, _ = published_well + other_parameter_id = get_parameter_id("pH", "Field Parameter") + + response = client.post( + PUBLISH_URL, json=_payload(thing_id, parameter_id=other_parameter_id) + ) + + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "parameter_id"] + + +def test_an_unknown_parameter_is_rejected_too(published_well): + thing_id, _ = published_well + + response = client.post(PUBLISH_URL, json=_payload(thing_id, parameter_id=-1)) + + assert response.status_code == 422 + assert response.json()["detail"][0]["loc"] == ["body", "parameter_id"] + + +def test_two_covering_deployments_are_a_422_not_a_coin_flip(published_well, sensor): + thing_id, _ = published_well + with session_ctx() as session: + second = Deployment( + sensor_id=sensor.id, + thing_id=thing_id, + installation_date="2019-01-01", + removal_date=None, + ) + session.add(second) + session.commit() + second_id = second.id + + try: + response = client.post(PUBLISH_URL, json=_payload(thing_id)) + + assert response.status_code == 422 + detail = response.json()["detail"][0] + assert detail["loc"] == ["body", "deployment_id"] + assert "2 deployments cover" in detail["msg"] + finally: + with session_ctx() as session: + session.delete(session.get(Deployment, second_id)) + session.commit() + + +def test_a_span_no_deployment_covers_is_a_422(published_well): + thing_id, deployment_id = published_well + # Retire the well's only deployment before the series was recorded. + with session_ctx() as session: + session.get(Deployment, deployment_id).removal_date = "2021-01-01" + session.commit() + + try: + response = client.post(PUBLISH_URL, json=_payload(thing_id)) + + assert response.status_code == 422 + assert "No deployment covers" in response.json()["detail"][0]["msg"] + finally: + with session_ctx() as session: + session.get(Deployment, deployment_id).removal_date = None + session.commit() + + +def test_deleting_from_a_well_with_no_deployments_removes_nothing(): + with session_ctx() as session: + thing = Thing( + name=f"Hydrograph Bare Well {datetime.now().timestamp()}", + first_visit_date="2023-03-03", + thing_type="water well", + release_status="draft", + ) + session.add(thing) + session.commit() + thing_id = thing.id + + try: + response = client.request( + "DELETE", + READ_URL, + params={ + "thing_id": thing_id, + "start_time": T0.isoformat(), + "end_time": (T0 + timedelta(hours=12)).isoformat(), + }, + ) + + assert response.status_code == 200, response.text + assert response.json() == { + "deleted_observation_count": 0, + "deleted_block_ids": [], + "updated_block_ids": [], + "thing_id": thing_id, + } + finally: + with session_ctx() as session: + session.delete(session.get(Thing, thing_id)) + session.commit() + + +def test_audit_stamping_survives_a_non_dict_user(): + # `authenticated()` yields the token claims, but the development bypass + # yields `True`. Reachable in any environment running with + # AUTHENTIK_DISABLE_AUTHENTICATION, where a publish must still write rather + # than fail reaching into a bool for a subject id. + from services.transducer_helper import _created_by + + assert _created_by(True) == (None, None) + assert _created_by({"sub": "1234567890", "name": "foobar"}) == ( + "1234567890", + "foobar", + ) + + +def test_a_review_status_outside_the_lexicon_is_rejected(published_well): + # `review_status` is lexicon-backed, so an unknown term would otherwise + # reach the column and fail on a foreign key rather than as a bad request. + thing_id, _ = published_well + + response = client.post( + PUBLISH_URL, json=_payload(thing_id, review_status="mostly reviewed") + ) + + assert response.status_code == 422 + assert any( + error["loc"][:2] == ["body", "review_status"] + for error in response.json()["detail"] + ) + + +def test_publish_waits_for_a_concurrent_writer_on_the_same_series(published_well): + """ + The overlap check reads state and then writes based on it, so it is only + correct if no other writer can slip between the two. Hold the series lock + from another connection and the publish must block rather than proceed on a + snapshot that is already stale. + """ + thing_id, _ = published_well + parameter_id = _groundwater_level_parameter_id() + result = {} + + def publish(): + result["response"] = client.post(PUBLISH_URL, json=_payload(thing_id)) + + blocker = engine.connect() + try: + blocker.execute( + text("SELECT pg_advisory_xact_lock(:thing_id, :parameter_id)"), + {"thing_id": thing_id, "parameter_id": parameter_id}, + ) + + worker = threading.Thread(target=publish, daemon=True) + worker.start() + worker.join(timeout=3) + assert worker.is_alive(), "publish did not wait for the series lock" + + # Releasing the transaction releases the lock. + blocker.rollback() + + worker.join(timeout=30) + assert not worker.is_alive(), "publish never completed after the lock lifted" + finally: + blocker.close() + + assert result["response"].status_code == 201, result["response"].text