From 8dc3f71aaf9211212511c6a1c0358d49ec5a6b88 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Mon, 24 Aug 2026 10:43:25 +0200 Subject: [PATCH] M8-R08: add atomic meter close and binding transfer --- app/api/routes/api/meter_sources.py | 84 +++++- app/api/routes/api/meters.py | 68 ++++- app/schemas/meter.py | 6 + app/schemas/meter_source.py | 11 + app/services/energy_cost.py | 9 +- app/services/meter_sources.py | 113 ++++++- app/services/meters.py | 78 ++++- frontend/src/api/schema.d.ts | 136 +++++++++ openapi/openapi.json | 195 ++++++++++++ openapi/openapi.yaml | 124 ++++++++ tests/test_api_meters.py | 443 +++++++++++++++++++++++++++- tests/test_energy_cost.py | 25 ++ tests/test_meter_source_api.py | 176 ++++++++++- tests/test_meter_source_services.py | 120 +++++++- tests/test_meters.py | 114 +++++++ 15 files changed, 1667 insertions(+), 35 deletions(-) diff --git a/app/api/routes/api/meter_sources.py b/app/api/routes/api/meter_sources.py index d001dfb..e629327 100644 --- a/app/api/routes/api/meter_sources.py +++ b/app/api/routes/api/meter_sources.py @@ -16,6 +16,7 @@ from app.models.energy import DsmrReading, Meter from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading from app.schemas.meter_source import ( BindingCreate, BindingListResponse, BindingPatch, BindingResponse, ChannelBindingSummaryResponse, + BindingTransferRequest, BindingTransferResponse, ChannelReadingResponse, ChannelReadingsResponse, CommoditiesResponse, CommodityResponse, DiscoverResponse, DiscoverChannelResponse, @@ -29,8 +30,9 @@ from app.services.dsmr_ingest import apply_dsmr_subscription from app.services.meter_sources import ( BindingNotFoundError, ChannelNotFoundError, MeterNotFoundError, MeterSourceError, SourceDeleteRestrictedError, SourceNotFoundError, create_binding, - create_source, delete_source, list_bindings, list_sources, update_binding, update_source, + create_source, delete_source, list_bindings, list_sources, transfer_binding, update_binding, update_source, ) +from app.services.energy_cost import recompute_range as electricity_recompute_range from app.services import timezone as _tz_mod from app.services.warmtelink_worker import warmtelink_worker_manager @@ -104,6 +106,25 @@ def _binding_error(exc: MeterSourceError) -> HTTPException: return HTTPException(status_code=422, detail=str(exc)) +def _recompute_binding_commodity(db: Session, commodity: str, start: datetime) -> None: + end = datetime.now(UTC) + if start >= end: + return + if commodity == "electricity": + electricity_recompute_range(db, start, end, commit=False, strict=True) + else: + from app.services.meter_cost import recompute_range + recompute_range(db, start, end, commit=False) + + +def _republish_after_commit(db: Session) -> None: + try: + from app.services.ha_discovery import publish_discovery + publish_discovery(db) + except Exception: + pass + + @router.get("/source-profiles", response_model=SourceProfilesResponse) def source_profiles(_auth: AuthenticatedSession = Depends(require_session)) -> SourceProfilesResponse: """Return profile metadata; default secrets are never populated with stored values.""" @@ -300,12 +321,19 @@ def post_meter_binding(meter_id: int, body: BindingCreate, db: Session = Depends try: binding = create_binding(db, meter_id=meter_id, channel_id=channel.id, started_at=_as_utc(body.started_at), ended_at=_as_utc(body.ended_at) if body.ended_at else None) + meter = db.get(Meter, meter_id) + db.flush() + _recompute_binding_commodity(db, meter.commodity, _as_utc(body.started_at)) db.commit() db.refresh(binding) + _republish_after_commit(db) return binding_response(binding) except MeterSourceError as exc: db.rollback() raise _binding_error(exc) from exc + except Exception: + db.rollback() + raise @router.patch("/bindings/{binding_uuid}", response_model=BindingResponse) @@ -323,10 +351,64 @@ def patch_binding(binding_uuid: str, body: BindingPatch, db: Session = Depends(g changes["started_at"] = _as_utc(body.started_at) if body.started_at is not None else None if "ended_at" in body.model_fields_set: changes["ended_at"] = _as_utc(body.ended_at) if body.ended_at is not None else None + old_started_at = _as_utc(binding.started_at) + old_ended_at = _as_utc(binding.ended_at) if binding.ended_at is not None else None updated = update_binding(db, binding.id, **changes) + meter = db.get(Meter, updated.meter_id) + if "started_at" in changes: + earliest = min(old_started_at, _as_utc(updated.started_at)) + elif "ended_at" in changes: + new_ended_at = _as_utc(updated.ended_at) if updated.ended_at is not None else None + changed_ends = [value for value in (old_ended_at, new_ended_at) if value is not None] + earliest = min(changed_ends) if changed_ends else old_started_at + else: + earliest = old_started_at + db.flush() + _recompute_binding_commodity(db, meter.commodity, earliest) db.commit() db.refresh(updated) + _republish_after_commit(db) return binding_response(updated) except MeterSourceError as exc: db.rollback() raise _binding_error(exc) from exc + except Exception: + db.rollback() + raise + + +@router.post("/meters/{meter_id}/bindings/transfer", response_model=BindingTransferResponse) +def post_binding_transfer( + meter_id: int, body: BindingTransferRequest, db: Session = Depends(get_db), + _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf), +) -> BindingTransferResponse: + source = db.execute(select(MeterSourceBinding).where( + MeterSourceBinding.uuid == body.from_binding_uuid + )).scalar_one_or_none() + channel = db.execute(select(MeterSourceChannel).where( + MeterSourceChannel.uuid == body.to_source_channel_uuid + )).scalar_one_or_none() + if source is None or channel is None: + raise HTTPException(status_code=404, detail="Meter source binding or channel not found.") + effective_at = _as_utc(body.effective_at) + try: + closed, created = transfer_binding(db, target_meter_id=meter_id, from_binding_id=source.id, + to_channel_id=channel.id, effective_at=effective_at) + meter = db.get(Meter, meter_id) + if closed.meter_id == meter.id: + earliest = effective_at + else: + earliest = min(_as_utc(closed.ended_at), effective_at) + db.flush() + _recompute_binding_commodity(db, meter.commodity, earliest) + db.commit() + db.refresh(closed) + db.refresh(created) + except MeterSourceError as exc: + db.rollback() + raise _binding_error(exc) from exc + except Exception: + db.rollback() + raise + _republish_after_commit(db) + return BindingTransferResponse(closed_binding=binding_response(closed), created_binding=binding_response(created)) diff --git a/app/api/routes/api/meters.py b/app/api/routes/api/meters.py index 80069f7..c20690a 100644 --- a/app/api/routes/api/meters.py +++ b/app/api/routes/api/meters.py @@ -55,6 +55,7 @@ from app.dependencies import get_db from app.models.energy import Meter from app.models.meter_source import MeterSourceChannel from app.schemas.meter import ( + MeterCloseRequest, MeterDeclareRequest, MeterBindingSummary, MeterListResponse, @@ -66,6 +67,7 @@ from app.services.meter_sources import ( MeterSourceError, create_binding, create_binding_for_meter_swap, + close_open_bindings_for_meter, ) from app.services import timezone as _tz_mod from app.services.auth import AuthenticatedSession @@ -74,6 +76,7 @@ from app.services.meters import ( MeterIntervalError, MeterOverlapError, declare_meter, + close_meter, list_meters, update_meter, ) @@ -151,7 +154,7 @@ def _trigger_recompute(db: Session, start: datetime, label: str) -> int: # started_at is in the future — nothing to recompute. logger.info("%s: started_at (%s) is in the future, skipping recompute.", label, start) return 0 - n = recompute_range(db, start, end, commit=False) + n = recompute_range(db, start, end, commit=False, strict=True) logger.info( "%s: recomputed %d period(s) in window [%s, %s).", label, @@ -162,6 +165,17 @@ def _trigger_recompute(db: Session, start: datetime, label: str) -> int: return n +def _recompute_commodity(db: Session, commodity: str, start: datetime, label: str) -> int: + if commodity == "electricity": + return _trigger_recompute(db, start, label) + from app.services.meter_cost import recompute_range as thermal_recompute_range + + end = datetime.now(UTC) + if start >= end: + return 0 + return thermal_recompute_range(db, start, end, commit=False) + + def _meter_response(meter: Meter) -> MeterResponse: """Serialize meter plus binding summaries without exposing source config.""" response = MeterResponse.model_validate(meter) @@ -248,6 +262,16 @@ def declare_energy_meter( note=body.note, ) db.flush() # assign PK before an optional binding and recompute + # A closed predecessor must never retain an open interval. For a + # meter swap with no selected channel we can safely hand off exactly + # one compatible open channel; ambiguity is fail-closed. + auto_channel = None + if body.source_channel_uuid is None and old_meter is not None and body.reason.value == "meter_swap": + candidates = [b for b in old_meter.source_bindings if b.ended_at is None and b.channel.unit == {"electricity": "kWh", "heating": "GJ", "hot_water": "m³"}.get(body.commodity)] + if len(candidates) > 1: + raise MeterSourceError("Meter swap has ambiguous open bindings; select a channel explicitly.") + if len(candidates) == 1: + auto_channel = candidates[0].channel if body.source_channel_uuid is not None: channel = db.execute( select(MeterSourceChannel).where(MeterSourceChannel.uuid == body.source_channel_uuid) @@ -269,14 +293,20 @@ def declare_energy_meter( channel_id=channel.id, started_at=started_at_utc, ) + elif auto_channel is not None: + create_binding_for_meter_swap(db, old_meter_id=old_meter.id, new_meter_id=new_meter.id, + channel_id=auto_channel.id, started_at=started_at_utc) + if old_meter is not None: + close_open_bindings_for_meter(db, old_meter.id, ended_at=started_at_utc) # Keep recompute in this transaction: a failure must not leave a new # meter, its predecessor, or either binding at a half-applied boundary. now = datetime.now(UTC) if started_at_utc < now: - _trigger_recompute(db, started_at_utc, "POST /api/energy/meters") + db.flush() + _recompute_commodity(db, body.commodity, started_at_utc, "POST /api/energy/meters") db.commit() - except (MeterOverlapError, MeterSourceError) as exc: + except (MeterIntervalError, MeterOverlapError, MeterSourceError) as exc: db.rollback() raise HTTPException( status_code=(status.HTTP_404_NOT_FOUND if isinstance(exc, ChannelNotFoundError) @@ -304,6 +334,30 @@ def declare_energy_meter( return _meter_response(new_meter) +@router.post("/meters/{meter_id}/close", response_model=MeterResponse) +def close_energy_meter( + meter_id: int, body: MeterCloseRequest, db: Session = Depends(get_db), + _auth: AuthenticatedSession = Depends(require_session), _csrf: None = Depends(require_csrf), +) -> MeterResponse: + meter = _get_meter_or_404(db, meter_id) + boundary = _localize_started_at(body.ended_at) + try: + close_meter(db, meter, ended_at=boundary) + close_open_bindings_for_meter(db, meter.id, ended_at=boundary) + db.flush() + _recompute_commodity(db, meter.commodity, boundary, f"POST /api/energy/meters/{meter_id}/close") + db.commit() + except (MeterIntervalError, MeterSourceError) as exc: + db.rollback() + raise HTTPException(status_code=422, detail=str(exc)) from exc + except Exception: + db.rollback() + raise + db.refresh(meter) + _trigger_discovery_republish(db) + return _meter_response(meter) + + # --------------------------------------------------------------------------- # PATCH /api/energy/meters/{id} # --------------------------------------------------------------------------- @@ -361,7 +415,13 @@ def patch_energy_meter( # Window = [min(old, new), now) — covers all periods whose attribution # may have changed due to the boundary shift in either direction. window_start = min(old_started_at, new_started_at_utc) - _trigger_recompute(db, window_start, f"PATCH /api/energy/meters/{meter_id}") + db.flush() + _recompute_commodity( + db, + meter.commodity, + window_start, + f"PATCH /api/energy/meters/{meter_id}", + ) db.commit() except MeterIntervalError as exc: diff --git a/app/schemas/meter.py b/app/schemas/meter.py index 20de70a..be09783 100644 --- a/app/schemas/meter.py +++ b/app/schemas/meter.py @@ -146,3 +146,9 @@ class MeterPatchRequest(BaseModel): "Triggers billing recompute over the affected window." ), ) + + +class MeterCloseRequest(BaseModel): + """Close the active meter epoch at an exclusive end boundary.""" + + ended_at: datetime diff --git a/app/schemas/meter_source.py b/app/schemas/meter_source.py index 9e47780..19c2a2a 100644 --- a/app/schemas/meter_source.py +++ b/app/schemas/meter_source.py @@ -135,6 +135,12 @@ class BindingPatch(BaseModel): ended_at: datetime | None = None +class BindingTransferRequest(BaseModel): + from_binding_uuid: str = Field(..., min_length=1, max_length=36) + to_source_channel_uuid: str = Field(..., min_length=1, max_length=36) + effective_at: datetime + + class BindingResponse(BaseModel): uuid: str meter_id: int @@ -149,3 +155,8 @@ class BindingResponse(BaseModel): class BindingListResponse(BaseModel): items: list[BindingResponse] total: int + + +class BindingTransferResponse(BaseModel): + closed_binding: BindingResponse + created_binding: BindingResponse diff --git a/app/services/energy_cost.py b/app/services/energy_cost.py index fd1cfdd..cc99558 100644 --- a/app/services/energy_cost.py +++ b/app/services/energy_cost.py @@ -666,7 +666,7 @@ def compute_closed_periods(session: Session) -> int: def recompute_range( - session: Session, start: datetime, end: datetime, *, commit: bool = True + session: Session, start: datetime, end: datetime, *, commit: bool = True, strict: bool = False ) -> int: """Recompute (overwrite) all 15-minute periods in ``[start, end)``. @@ -693,6 +693,11 @@ def recompute_range( When true (the default), commit after all periods have been processed. Callers composing this recompute with other writes may pass false and own the surrounding transaction themselves. + strict: + When true, propagate a failed period computation to the caller. This + is for lifecycle transactions which must roll back their meter/binding + mutation together with the cost recompute. The default remains + best-effort for existing background and standalone callers. start: Inclusive start datetime (floored to the nearest quarter-hour internally). end: @@ -723,6 +728,8 @@ def recompute_range( if did_write: written += 1 except Exception: + if strict: + raise logger.exception( "recompute_range: unexpected error for t0=%s — continuing.", t0.isoformat(), diff --git a/app/services/meter_sources.py b/app/services/meter_sources.py index 0af6a7f..4ed4f2e 100644 --- a/app/services/meter_sources.py +++ b/app/services/meter_sources.py @@ -250,7 +250,7 @@ def _validate_binding( channel_id: int, started_at: datetime, ended_at: datetime | None, - excluding_id: int | None = None, + excluding_ids: set[int] | None = None, ) -> None: meter = _get_meter(session, meter_id) channel = get_channel(session, channel_id) @@ -264,7 +264,19 @@ def _validate_binding( ) if ended_at is not None and _as_utc(ended_at) <= _as_utc(started_at): raise BindingValidationError("Binding ended_at must be strictly after started_at.") + if _as_utc(started_at) < _as_utc(meter.started_at): + raise BindingValidationError("Binding must not start before its meter epoch.") + if meter.ended_at is None: + if ended_at is not None: + # A historical binding on an active epoch is valid, but it must be + # wholly within that epoch (whose upper bound is open). + pass + else: + meter_end = _as_utc(meter.ended_at) + if ended_at is None or _as_utc(ended_at) > meter_end: + raise BindingValidationError("Closed meter bindings must end within the meter epoch.") + excluded = excluding_ids or set() candidates = session.execute( select(MeterSourceBinding).where( or_( @@ -274,7 +286,7 @@ def _validate_binding( ) ).scalars() for existing in candidates: - if existing.id == excluding_id: + if existing.id in excluded: continue if half_open_intervals_overlap( _as_utc(started_at), @@ -295,6 +307,9 @@ def create_binding( ended_at: datetime | None = None, ) -> MeterSourceBinding: """Create a compatible non-overlapping half-open source binding.""" + now = _utc_now() + if _as_utc(started_at) > now or (ended_at is not None and _as_utc(ended_at) > now): + raise BindingValidationError("Binding boundaries must not be in the future.") _validate_binding( session, meter_id=meter_id, @@ -302,7 +317,6 @@ def create_binding( started_at=started_at, ended_at=ended_at, ) - now = _utc_now() binding = MeterSourceBinding( meter_id=meter_id, channel_id=channel_id, @@ -403,13 +417,16 @@ def update_binding( new_channel_id = binding.channel_id if channel_id is None else channel_id new_started_at = binding.started_at if started_at is None else started_at new_ended_at = binding.ended_at if ended_at is _UNSET else ended_at + now = _utc_now() + if _as_utc(new_started_at) > now or (new_ended_at is not None and _as_utc(new_ended_at) > now): + raise BindingValidationError("Binding boundaries must not be in the future.") _validate_binding( session, meter_id=new_meter_id, channel_id=new_channel_id, started_at=new_started_at, ended_at=new_ended_at, - excluding_id=binding.id, + excluding_ids={binding.id}, ) binding.meter_id = new_meter_id binding.channel_id = new_channel_id @@ -422,3 +439,91 @@ def update_binding( def close_binding(session: Session, binding_id: int, *, ended_at: datetime) -> MeterSourceBinding: """Close an existing binding at its exclusive end boundary.""" return update_binding(session, binding_id, ended_at=ended_at) + + +def close_open_bindings_for_meter(session: Session, meter_id: int, *, ended_at: datetime) -> list[MeterSourceBinding]: + """Close every open binding on a meter at one shared boundary.""" + bindings = list(session.execute( + select(MeterSourceBinding).where( + MeterSourceBinding.meter_id == meter_id, MeterSourceBinding.ended_at.is_(None) + ) + ).scalars()) + for binding in bindings: + update_binding(session, binding.id, ended_at=ended_at) + return bindings + + +def transfer_binding( + session: Session, *, target_meter_id: int, from_binding_id: int, to_channel_id: int, + effective_at: datetime, +) -> tuple[MeterSourceBinding, MeterSourceBinding]: + """Atomically close a binding and open its replacement on the target meter.""" + source = session.get(MeterSourceBinding, from_binding_id) + if source is None: + raise BindingNotFoundError(f"Meter source binding {from_binding_id} was not found.") + target = _get_meter(session, target_meter_id) + old_meter = _get_meter(session, source.meter_id) + effective_at = _as_utc(effective_at) + now = _utc_now() + if effective_at > now: + raise BindingValidationError("Binding transfer effective_at must not be in the future.") + if old_meter.commodity != target.commodity: + raise BindingValidationError("Binding transfer meters must have the same commodity.") + if source.ended_at is not None: + raise BindingValidationError("Only an open binding can be transferred.") + if old_meter.id == target.id: + close_at = effective_at + else: + # Recovery is deliberately narrow: the source meter must be the one + # and only most-recent closed predecessor in this commodity's timeline. + # A manually closed meter may leave an intentional epoch gap before the + # target is declared, so adjacency is not required. + if old_meter.ended_at is None: + raise BindingValidationError("Source binding must belong to a closed predecessor meter.") + timeline = list(session.execute( + select(Meter).where(Meter.commodity == target.commodity) + ).scalars()) + predecessors = [ + meter for meter in timeline + if meter.id != target.id + and meter.ended_at is not None + and _as_utc(meter.ended_at) <= _as_utc(target.started_at) + ] + if not predecessors: + raise BindingValidationError("Source meter is not the unique immediately preceding meter.") + latest_end = max(_as_utc(meter.ended_at) for meter in predecessors) + latest = [meter for meter in predecessors if _as_utc(meter.ended_at) == latest_end] + if len(latest) != 1 or latest[0].id != old_meter.id: + raise BindingValidationError("Source meter is not the unique immediately preceding meter.") + # Reject any overlapping epoch around either endpoint. A separate + # meter inside the gap is already excluded by the predecessor check; + # one extending into either endpoint is an ambiguous timeline too. + for meter in timeline: + if meter.id in {old_meter.id, target.id}: + continue + meter_end = _as_utc(meter.ended_at) if meter.ended_at is not None else None + if ( + half_open_intervals_overlap( + _as_utc(old_meter.started_at), _as_utc(old_meter.ended_at), + _as_utc(meter.started_at), meter_end, + ) + or half_open_intervals_overlap( + _as_utc(target.started_at), + _as_utc(target.ended_at) if target.ended_at is not None else None, + _as_utc(meter.started_at), meter_end, + ) + ): + raise BindingValidationError("Source meter has an ambiguous commodity timeline.") + close_at = _as_utc(old_meter.ended_at) + if effective_at < _as_utc(target.started_at): + raise BindingValidationError("Transfer effective_at must be within the target meter epoch.") + if effective_at < _as_utc(source.started_at): + raise BindingValidationError("Transfer effective_at precedes the source binding.") + # Validate the target before mutating the old row, then close/create in one session. + _validate_binding(session, meter_id=target.id, channel_id=to_channel_id, + started_at=effective_at, ended_at=None, + excluding_ids={source.id}) + update_binding(session, source.id, ended_at=close_at) + created = create_binding(session, meter_id=target.id, channel_id=to_channel_id, + started_at=effective_at) + return source, created diff --git a/app/services/meters.py b/app/services/meters.py index b6dc0cb..0d1297b 100644 --- a/app/services/meters.py +++ b/app/services/meters.py @@ -53,6 +53,7 @@ from sqlalchemy import select from sqlalchemy.orm import Session from app.models.energy import Meter +from app.models.meter_source import MeterSourceBinding logger = logging.getLogger(__name__) @@ -104,6 +105,20 @@ class MeterIntervalError(MeterError): """ +def close_meter(session: Session, meter: Meter, *, ended_at: datetime) -> Meter: + """Close an active meter at a valid, non-future exclusive boundary.""" + boundary = _as_utc(ended_at) + if meter.ended_at is not None: + raise MeterIntervalError("Only an active meter can be closed.") + if boundary <= _as_utc(meter.started_at): + raise MeterIntervalError("Meter ended_at must be strictly after started_at.") + if boundary > datetime.now(UTC): + raise MeterIntervalError("Meter ended_at must not be in the future.") + _validate_bindings_fit_meter_end(session, meter, boundary) + meter.ended_at = boundary + return meter + + # --------------------------------------------------------------------------- # Internal query helpers # --------------------------------------------------------------------------- @@ -121,6 +136,23 @@ def _active_meter(session: Session, commodity: str) -> Optional[Meter]: ).scalar_one_or_none() +def _validate_bindings_fit_meter_end(session: Session, meter: Meter, boundary: datetime) -> None: + """Reject an epoch close that would put any retained binding out of bounds. + + Closed binding history is immutable here. Open bindings may subsequently + be closed by the caller at the shared meter boundary, but only when that + produces a non-empty interval. + """ + for binding in session.execute( + select(MeterSourceBinding).where(MeterSourceBinding.meter_id == meter.id) + ).scalars(): + if binding.ended_at is None: + if _as_utc(binding.started_at) >= boundary: + raise MeterIntervalError("Open binding cannot be closed within the proposed meter epoch.") + elif _as_utc(binding.ended_at) > boundary: + raise MeterIntervalError("Closed binding extends beyond the proposed meter epoch.") + + def _meter_before(session: Session, meter: Meter) -> Optional[Meter]: """Return the meter whose ``ended_at`` equals *meter*'s ``started_at``. @@ -296,6 +328,8 @@ def declare_meter( If *started_at* is strictly earlier than the current active meter's ``started_at`` (chronological backdate below the active epoch's start). """ + if _as_utc(started_at) > datetime.now(UTC): + raise MeterIntervalError("Meter started_at must not be in the future.") active = _active_meter(session, commodity) if active is not None: @@ -308,6 +342,9 @@ def declare_meter( "Declare a started_at on or after the active meter's start to avoid " "a chronologically inconsistent epoch ordering." ) + # Validate before changing the epoch: retained closed binding history + # must never be silently truncated by a later declaration. + _validate_bindings_fit_meter_end(session, active, _as_utc(started_at)) # Close the current active meter at the swap point (contiguous handoff). active.ended_at = started_at logger.info( @@ -400,6 +437,9 @@ def update_meter( invert). b. It must be **strictly before** this meter's ``ended_at`` (if set), so this meter's epoch remains non-empty. + c. Every binding on this meter and its affected predecessor must remain + wholly inside its proposed epoch. The service rejects the correction + rather than rewriting binding history. Note: triggering a billing recompute (``recompute_range``) after a retroactive ``started_at`` change is **out of scope** for this service @@ -429,6 +469,14 @@ def update_meter( If the new ``started_at`` would produce an invalid (empty or inverted) epoch for this meter or the immediately preceding one. """ + # Validate the proposed epoch boundary before touching *any* mutable + # field. PATCH accepts label/note together with started_at, so doing this + # first keeps an invalid future timestamp from leaking a partial in-session + # update before the API's rollback boundary is reached. + proposed_started_at = _as_utc(started_at) if started_at is not None else None + if proposed_started_at is not None and proposed_started_at > datetime.now(UTC): + raise MeterIntervalError("Meter started_at must not be in the future.") + if label is not None: meter.label = label logger.info("Updated meter id=%d label=%r", meter.id, label) @@ -439,10 +487,11 @@ def update_meter( if started_at is not None: old_started_at = meter.started_at + assert proposed_started_at is not None # --- Validate upper bound: new started_at must be < this meter's ended_at (if set). if meter.ended_at is not None: - if _as_utc(started_at) >= _as_utc(meter.ended_at): + if proposed_started_at >= _as_utc(meter.ended_at): raise MeterIntervalError( f"New started_at ({started_at.isoformat()}) must be strictly before " f"this meter's ended_at ({meter.ended_at.isoformat()}). " @@ -454,12 +503,37 @@ def update_meter( # --- Validate lower bound: new started_at must be strictly after prev's started_at. if prev is not None: - if _as_utc(started_at) <= _as_utc(prev.started_at): + if proposed_started_at <= _as_utc(prev.started_at): raise MeterIntervalError( f"New started_at ({started_at.isoformat()}) must be strictly after " f"the previous meter's started_at ({prev.started_at.isoformat()}). " "Moving the boundary that far back would collapse the previous meter's epoch." ) + # A boundary correction changes both adjacent meter epochs. Fail closed + # rather than silently rewriting binding history: every existing binding + # must still fit in its proposed epoch before either Meter is mutated. + affected_meters = [ + (meter, proposed_started_at, _as_utc(meter.ended_at) if meter.ended_at is not None else None) + ] + if prev is not None: + affected_meters.append((prev, _as_utc(prev.started_at), proposed_started_at)) + for affected_meter, proposed_start, proposed_end in affected_meters: + bindings = session.execute( + select(MeterSourceBinding).where(MeterSourceBinding.meter_id == affected_meter.id) + ).scalars() + for binding in bindings: + if _as_utc(binding.started_at) < proposed_start: + raise MeterIntervalError( + f"Binding {binding.id} starts before meter {affected_meter.id}'s epoch." + ) + if proposed_end is not None and ( + binding.ended_at is None or _as_utc(binding.ended_at) > proposed_end + ): + raise MeterIntervalError( + f"Binding {binding.id} would fall outside meter {affected_meter.id}'s epoch." + ) + + if prev is not None: # Maintain continuity: update the previous meter's ended_at to match the new start. prev.ended_at = started_at logger.info( diff --git a/frontend/src/api/schema.d.ts b/frontend/src/api/schema.d.ts index 7c8afa6..f8d2e20 100644 --- a/frontend/src/api/schema.d.ts +++ b/frontend/src/api/schema.d.ts @@ -663,6 +663,23 @@ export interface paths { patch?: never; trace?: never; }; + "/api/energy/meters/{meter_id}/close": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Close Energy Meter */ + post: operations["close_energy_meter_api_energy_meters__meter_id__close_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/energy/meters/{meter_id}": { parameters: { query?: never; @@ -858,6 +875,23 @@ export interface paths { patch: operations["patch_binding_api_energy_bindings__binding_uuid__patch"]; trace?: never; }; + "/api/energy/meters/{meter_id}/bindings/transfer": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** Post Binding Transfer */ + post: operations["post_binding_transfer_api_energy_meters__meter_id__bindings_transfer_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/expose": { parameters: { query?: never; @@ -1504,6 +1538,23 @@ export interface components { */ updated_at: string; }; + /** BindingTransferRequest */ + BindingTransferRequest: { + /** From Binding Uuid */ + from_binding_uuid: string; + /** To Source Channel Uuid */ + to_source_channel_uuid: string; + /** + * Effective At + * Format: date-time + */ + effective_at: string; + }; + /** BindingTransferResponse */ + BindingTransferResponse: { + closed_binding: components["schemas"]["BindingResponse"]; + created_binding: components["schemas"]["BindingResponse"]; + }; /** * CatalogEntrySchema * @description An entity from the catalog with its current toggle state. @@ -2022,6 +2073,17 @@ export interface components { /** Ended At */ ended_at: string | null; }; + /** + * MeterCloseRequest + * @description Close the active meter epoch at an exclusive end boundary. + */ + MeterCloseRequest: { + /** + * Ended At + * Format: date-time + */ + ended_at: string; + }; /** * MeterCostPeriodSchema * @description One auditable thermal ledger row; all Decimal values are JSON strings. @@ -4021,6 +4083,43 @@ export interface operations { }; }; }; + close_energy_meter_api_energy_meters__meter_id__close_post: { + parameters: { + query?: never; + header?: { + "X-CSRF-Token"?: string | null; + }; + path: { + meter_id: number; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["MeterCloseRequest"]; + }; + }; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MeterResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; patch_energy_meter_api_energy_meters__meter_id__patch: { parameters: { query?: never; @@ -4457,6 +4556,43 @@ export interface operations { }; }; }; + post_binding_transfer_api_energy_meters__meter_id__bindings_transfer_post: { + parameters: { + query?: never; + header?: { + "X-CSRF-Token"?: string | null; + }; + path: { + meter_id: number; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["BindingTransferRequest"]; + }; + }; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["BindingTransferResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; get_expose_api_expose_get: { parameters: { query?: never; diff --git a/openapi/openapi.json b/openapi/openapi.json index 4b41874..2a7186b 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -1777,6 +1777,74 @@ } } }, + "/api/energy/meters/{meter_id}/close": { + "post": { + "tags": [ + "api-energy-meters" + ], + "summary": "Close Energy Meter", + "operationId": "close_energy_meter_api_energy_meters__meter_id__close_post", + "parameters": [ + { + "name": "meter_id", + "in": "path", + "required": true, + "schema": { + "type": "integer", + "title": "Meter Id" + } + }, + { + "name": "X-CSRF-Token", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "X-Csrf-Token" + } + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/MeterCloseRequest" + } + } + } + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/MeterResponse" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, "/api/energy/meters/{meter_id}": { "patch": { "tags": [ @@ -2497,6 +2565,74 @@ } } }, + "/api/energy/meters/{meter_id}/bindings/transfer": { + "post": { + "tags": [ + "api-energy-meter-sources" + ], + "summary": "Post Binding Transfer", + "operationId": "post_binding_transfer_api_energy_meters__meter_id__bindings_transfer_post", + "parameters": [ + { + "name": "meter_id", + "in": "path", + "required": true, + "schema": { + "type": "integer", + "title": "Meter Id" + } + }, + { + "name": "X-CSRF-Token", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "X-Csrf-Token" + } + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/BindingTransferRequest" + } + } + } + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/BindingTransferResponse" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, "/api/expose": { "get": { "tags": [ @@ -3762,6 +3898,50 @@ ], "title": "BindingResponse" }, + "BindingTransferRequest": { + "properties": { + "from_binding_uuid": { + "type": "string", + "maxLength": 36, + "minLength": 1, + "title": "From Binding Uuid" + }, + "to_source_channel_uuid": { + "type": "string", + "maxLength": 36, + "minLength": 1, + "title": "To Source Channel Uuid" + }, + "effective_at": { + "type": "string", + "format": "date-time", + "title": "Effective At" + } + }, + "type": "object", + "required": [ + "from_binding_uuid", + "to_source_channel_uuid", + "effective_at" + ], + "title": "BindingTransferRequest" + }, + "BindingTransferResponse": { + "properties": { + "closed_binding": { + "$ref": "#/components/schemas/BindingResponse" + }, + "created_binding": { + "$ref": "#/components/schemas/BindingResponse" + } + }, + "type": "object", + "required": [ + "closed_binding", + "created_binding" + ], + "title": "BindingTransferResponse" + }, "CatalogEntrySchema": { "properties": { "entity": { @@ -4907,6 +5087,21 @@ "title": "MeterBindingSummary", "description": "Stable, non-sensitive binding identity embedded in meter responses." }, + "MeterCloseRequest": { + "properties": { + "ended_at": { + "type": "string", + "format": "date-time", + "title": "Ended At" + } + }, + "type": "object", + "required": [ + "ended_at" + ], + "title": "MeterCloseRequest", + "description": "Close the active meter epoch at an exclusive end boundary." + }, "MeterCostPeriodSchema": { "properties": { "commodity": { diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index 60ea221..445ece9 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -1360,6 +1360,46 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /api/energy/meters/{meter_id}/close: + post: + tags: + - api-energy-meters + summary: Close Energy Meter + operationId: close_energy_meter_api_energy_meters__meter_id__close_post + parameters: + - name: meter_id + in: path + required: true + schema: + type: integer + title: Meter Id + - name: X-CSRF-Token + in: header + required: false + schema: + anyOf: + - type: string + - type: 'null' + title: X-Csrf-Token + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/MeterCloseRequest' + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/MeterResponse' + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /api/energy/meters/{meter_id}: patch: tags: @@ -1802,6 +1842,46 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /api/energy/meters/{meter_id}/bindings/transfer: + post: + tags: + - api-energy-meter-sources + summary: Post Binding Transfer + operationId: post_binding_transfer_api_energy_meters__meter_id__bindings_transfer_post + parameters: + - name: meter_id + in: path + required: true + schema: + type: integer + title: Meter Id + - name: X-CSRF-Token + in: header + required: false + schema: + anyOf: + - type: string + - type: 'null' + title: X-Csrf-Token + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/BindingTransferRequest' + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/BindingTransferResponse' + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /api/expose: get: tags: @@ -2783,6 +2863,39 @@ components: - created_at - updated_at title: BindingResponse + BindingTransferRequest: + properties: + from_binding_uuid: + type: string + maxLength: 36 + minLength: 1 + title: From Binding Uuid + to_source_channel_uuid: + type: string + maxLength: 36 + minLength: 1 + title: To Source Channel Uuid + effective_at: + type: string + format: date-time + title: Effective At + type: object + required: + - from_binding_uuid + - to_source_channel_uuid + - effective_at + title: BindingTransferRequest + BindingTransferResponse: + properties: + closed_binding: + $ref: '#/components/schemas/BindingResponse' + created_binding: + $ref: '#/components/schemas/BindingResponse' + type: object + required: + - closed_binding + - created_binding + title: BindingTransferResponse CatalogEntrySchema: properties: entity: @@ -3618,6 +3731,17 @@ components: - ended_at title: MeterBindingSummary description: Stable, non-sensitive binding identity embedded in meter responses. + MeterCloseRequest: + properties: + ended_at: + type: string + format: date-time + title: Ended At + type: object + required: + - ended_at + title: MeterCloseRequest + description: Close the active meter epoch at an exclusive end boundary. MeterCostPeriodSchema: properties: commodity: diff --git a/tests/test_api_meters.py b/tests/test_api_meters.py index 0a2e7fd..6c476e3 100644 --- a/tests/test_api_meters.py +++ b/tests/test_api_meters.py @@ -31,7 +31,8 @@ Retroactive recompute integration from __future__ import annotations -from datetime import UTC, datetime, timedelta +from datetime import UTC, datetime, timedelta, timezone +from decimal import Decimal from unittest.mock import patch import pytest @@ -39,7 +40,7 @@ from fastapi.testclient import TestClient from sqlalchemy import create_engine, event, select from sqlalchemy.orm import Session -from app.models.energy import EnergyCostPeriod, Meter +from app.models.energy import EnergyCostPeriod, Meter, MeterCostPeriod from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel # --------------------------------------------------------------------------- @@ -68,8 +69,15 @@ def _declare_payload(**overrides) -> dict: return base -def _add_bound_channel(engine, *, meter_id: int, started_at: datetime) -> str: - """Persist one test-only DSMR channel binding and return its public UUID.""" +def _add_bound_channel( + engine, + *, + meter_id: int, + started_at: datetime, + ended_at: datetime | None = None, + unit: str = "kWh", +) -> str: + """Persist one test-only channel binding and return its public UUID.""" with Session(engine) as session: source = MeterSource( name="Test DSMR", @@ -84,9 +92,9 @@ def _add_bound_channel(engine, *, meter_id: int, started_at: datetime) -> str: session.flush() channel = MeterSourceChannel( source_id=source.id, - channel_key="electricity-total", - label="Electricity total", - unit="kWh", + channel_key=f"test-total-{meter_id}", + label="Test total", + unit=unit, created_at=started_at, updated_at=started_at, ) @@ -97,6 +105,7 @@ def _add_bound_channel(engine, *, meter_id: int, started_at: datetime) -> str: meter_id=meter_id, channel_id=channel.id, started_at=started_at, + ended_at=ended_at, created_at=started_at, updated_at=started_at, ) @@ -595,6 +604,7 @@ def test_declare_meter_retroactive_triggers_recompute(meters_client): # recompute_range should have been called with start == t_past assert mock_recompute.called assert mock_recompute.call_args.kwargs["commit"] is False + assert mock_recompute.call_args.kwargs["strict"] is True call_args = mock_recompute.call_args recompute_start = call_args[0][1] # positional arg index 1 (session is 0) # Normalise for comparison @@ -735,6 +745,7 @@ def test_patch_meter_started_at_retroactive_triggers_recompute(meters_client): # recompute should be triggered assert mock_recompute.called assert mock_recompute.call_args.kwargs["commit"] is False + assert mock_recompute.call_args.kwargs["strict"] is True call_args = mock_recompute.call_args recompute_start = call_args[0][1] if recompute_start.tzinfo is None: @@ -776,6 +787,59 @@ def test_patch_meter_started_at_interval_violation_returns_422(meters_client): assert resp.status_code == 422 +@pytest.mark.parametrize("shift", ["later", "earlier"]) +def test_patch_meter_started_at_rejects_boundary_shift_that_strands_binding( + meters_client, mock_publish_discovery, shift +): + """Rejected boundary shifts leave adjacent meters/bindings untouched and emit no side effects.""" + client, engine = meters_client + _login(client) + t0 = datetime(2024, 1, 1, tzinfo=UTC) + boundary = datetime(2025, 1, 1, tzinfo=UTC) + with patch("app.api.routes.api.meters.recompute_range", return_value=0): + meter_a = client.post( + "/api/energy/meters", json=_declare_payload(label="A", started_at=t0.isoformat()), + headers={"X-CSRF-Token": _CSRF}, + ).json() + meter_b = client.post( + "/api/energy/meters", json=_declare_payload( + label="B", started_at=boundary.isoformat(), reason="meter_swap" + ), headers={"X-CSRF-Token": _CSRF}, + ).json() + if shift == "later": + _add_bound_channel(engine, meter_id=meter_b["id"], started_at=boundary) + proposed = boundary + timedelta(days=1) + else: + _add_bound_channel(engine, meter_id=meter_a["id"], started_at=t0) + with Session(engine) as session: + binding = session.scalar(select(MeterSourceBinding)) + assert binding is not None + binding.ended_at = boundary + session.commit() + proposed = boundary - timedelta(days=1) + + mock_publish_discovery.reset_mock() + with patch("app.api.routes.api.meters.recompute_range", return_value=0) as recompute: + response = client.patch( + f"/api/energy/meters/{meter_b['id']}", json={"started_at": proposed.isoformat()}, + headers={"X-CSRF-Token": _CSRF}, + ) + assert response.status_code == 422 + recompute.assert_not_called() + mock_publish_discovery.assert_not_called() + + with Session(engine) as observer: + assert observer.get(Meter, meter_a["id"]).ended_at.replace(tzinfo=UTC) == boundary + assert observer.get(Meter, meter_b["id"]).started_at.replace(tzinfo=UTC) == boundary + binding = observer.scalar(select(MeterSourceBinding)) + assert binding is not None + if shift == "later": + assert binding.meter_id == meter_b["id"] and binding.ended_at is None + else: + assert binding.meter_id == meter_a["id"] + assert binding.ended_at.replace(tzinfo=UTC) == boundary + + def test_patch_meter_no_recompute_when_started_at_not_changed(meters_client): """PATCH that only changes label does NOT trigger recompute.""" client, _ = meters_client @@ -801,6 +865,201 @@ def test_patch_meter_no_recompute_when_started_at_not_changed(meters_client): assert not mock_recompute.called +@pytest.mark.parametrize("representation", ["aware_utc", "naive_local"]) +def test_patch_meter_rejects_future_started_at_before_any_side_effect( + meters_client, mock_publish_discovery, monkeypatch, representation +): + """Future aware and local-naive starts leave all persisted state untouched.""" + client, engine = meters_client + _login(client) + started = datetime.now(UTC) - timedelta(hours=2) + with patch("app.api.routes.api.meters.recompute_range", return_value=0): + declared = client.post( + "/api/energy/meters", + json=_declare_payload( + label="Original", note="Original note", started_at=started.isoformat() + ), + headers={"X-CSRF-Token": _CSRF}, + ) + meter_id = declared.json()["id"] + mock_publish_discovery.reset_mock() + future = datetime.now(UTC) + timedelta(hours=2) + if representation == "aware_utc": + proposed = future.isoformat() + else: + monkeypatch.setattr( + "app.services.timezone.local_tz", lambda: timezone(timedelta(hours=2)) + ) + proposed = (future + timedelta(hours=2)).replace(tzinfo=None).isoformat() + + with patch("app.api.routes.api.meters.recompute_range", return_value=0) as recompute: + response = client.patch( + f"/api/energy/meters/{meter_id}", + json={"label": "Changed", "note": "Changed note", "started_at": proposed}, + headers={"X-CSRF-Token": _CSRF}, + ) + + assert response.status_code == 422 + recompute.assert_not_called() + mock_publish_discovery.assert_not_called() + with Session(engine) as observer: + meter = observer.get(Meter, meter_id) + assert meter is not None + assert meter.label == "Original" + assert meter.note == "Original note" + assert meter.started_at.replace(tzinfo=UTC) == started + + +@pytest.mark.parametrize("commodity", ["heating", "hot_water"]) +def test_patch_thermal_started_at_uses_thermal_recompute(meters_client, monkeypatch, commodity): + """A successful thermal correction never routes through electricity recompute.""" + from app.services import meter_cost + + client, _ = meters_client + _login(client) + started = datetime.now(UTC) - timedelta(hours=3) + calls = [] + monkeypatch.setattr( + meter_cost, + "recompute_range", + lambda db, start, end, *, commit: calls.append((start, end, commit)) or 0, + ) + with patch("app.api.routes.api.meters.recompute_range", return_value=0) as electricity: + declared = client.post( + "/api/energy/meters", + json=_declare_payload( + commodity=commodity, + label=f"{commodity} meter", + started_at=started.isoformat(), + ), + headers={"X-CSRF-Token": _CSRF}, + ) + assert declared.status_code == 201 + calls.clear() + electricity.reset_mock() + response = client.patch( + f"/api/energy/meters/{declared.json()['id']}", + json={"started_at": (started + timedelta(minutes=30)).isoformat()}, + headers={"X-CSRF-Token": _CSRF}, + ) + + assert response.status_code == 200 + assert len(calls) == 1 + assert calls[0][2] is False + electricity.assert_not_called() + + +def _create_adjacent_thermal_patch_state(engine, commodity: str): + """Create an editable thermal boundary with two untouched bindings.""" + old_start = datetime.now(UTC) - timedelta(hours=5) + boundary = old_start + timedelta(hours=1) + shifted = boundary + timedelta(minutes=30) + current_binding_start = shifted + timedelta(minutes=30) + with Session(engine) as session: + previous = Meter( + label="Previous thermal meter", + commodity=commodity, + started_at=old_start, + ended_at=boundary, + reason="initial", + created_at=old_start, + ) + current = Meter( + label="Current thermal meter", + commodity=commodity, + started_at=boundary, + reason="meter_swap", + created_at=boundary, + ) + session.add_all([previous, current]) + session.commit() + previous_id, current_id = previous.id, current.id + unit = {"heating": "GJ", "hot_water": "m³"}[commodity] + _add_bound_channel( + engine, meter_id=previous_id, started_at=old_start, ended_at=boundary, unit=unit + ) + _add_bound_channel(engine, meter_id=current_id, started_at=current_binding_start, unit=unit) + return previous_id, current_id, boundary, shifted, current_binding_start + + +@pytest.mark.parametrize("commodity", ["heating", "hot_water"]) +@pytest.mark.parametrize("failure", ["recompute", "flush", "commit"]) +def test_patch_thermal_failure_rolls_back_lifecycle_and_cost_state( + meters_client, mock_publish_discovery, monkeypatch, commodity, failure +): + """Thermal PATCH failures roll back meters, bindings, cost writes, and HA.""" + from app.services import meter_cost + + client, engine = meters_client + _login(client) + previous_id, current_id, boundary, shifted, binding_start = _create_adjacent_thermal_patch_state( + engine, commodity + ) + + def recompute_with_uncommitted_cost(db, start, end, *, commit): + assert commit is False + db.add( + MeterCostPeriod( + commodity=commodity, + period_start=shifted, + period_end=shifted + timedelta(minutes=15), + quantity=Decimal("0"), + cost=Decimal("0"), + currency="EUR", + cost_breakdown={}, + pricing_snapshot={}, + quality="invalid", + degraded=True, + degraded_reason="test rollback", + created_at=shifted, + updated_at=shifted, + ) + ) + if failure == "recompute": + raise RuntimeError("thermal recompute failed") + return 0 + + monkeypatch.setattr(meter_cost, "recompute_range", recompute_with_uncommitted_cost) + if failure == "flush": + monkeypatch.setattr( + Session, + "flush", + lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("thermal flush failed")), + ) + if failure == "commit": + def fail_commit(_session: Session) -> None: + raise RuntimeError("thermal commit failed") + + event.listen(Session, "before_commit", fail_commit) + + try: + expected = f"thermal {failure} failed" + with pytest.raises(RuntimeError, match=expected): + client.patch( + f"/api/energy/meters/{current_id}", + json={"started_at": shifted.isoformat()}, + headers={"X-CSRF-Token": _CSRF}, + ) + finally: + if failure == "commit": + event.remove(Session, "before_commit", fail_commit) + if failure == "flush": + monkeypatch.undo() + + mock_publish_discovery.assert_not_called() + with Session(engine) as observer: + previous = observer.get(Meter, previous_id) + current = observer.get(Meter, current_id) + assert previous is not None and previous.ended_at.replace(tzinfo=UTC) == boundary + assert current is not None and current.started_at.replace(tzinfo=UTC) == boundary + bindings = observer.execute( + select(MeterSourceBinding).order_by(MeterSourceBinding.meter_id) + ).scalars().all() + assert bindings[0].ended_at.replace(tzinfo=UTC) == boundary + assert bindings[1].started_at.replace(tzinfo=UTC) == binding_start + assert observer.execute(select(MeterCostPeriod)).scalars().all() == [] + + # --------------------------------------------------------------------------- # Timeline continuity (recompute mocked to avoid slow computation over empty quarters) # --------------------------------------------------------------------------- @@ -994,3 +1253,173 @@ def test_declare_meter_succeeds_when_publish_discovery_raises(meters_client): # The meter must be created successfully despite the discovery failure. assert resp.status_code == 201 assert resp.json()["label"] == "Best Effort Meter" + + +@pytest.mark.parametrize("commodity", ["heating", "hot_water"]) +def test_thermal_declare_and_close_use_meter_cost_recompute(meters_client, monkeypatch, commodity): + """Thermal lifecycle routes use the meter-cost helper's actual signature.""" + from app.services import meter_cost + + client, _ = meters_client + _login(client) + calls = [] + monkeypatch.setattr(meter_cost, "recompute_range", lambda db, start, end, *, commit: calls.append((start, end, commit)) or 0) + started = datetime.now(UTC) - timedelta(hours=2) + declared = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ + "label": f"{commodity} meter", "started_at": started.isoformat(), "reason": "initial", "commodity": commodity, + }) + assert declared.status_code == 201 + closed = client.post(f"/api/energy/meters/{declared.json()['id']}/close", headers={"X-CSRF-Token": _CSRF}, json={ + "ended_at": (started + timedelta(hours=1)).isoformat(), + }) + assert closed.status_code == 200 + assert len(calls) == 2 and all(call[2] is False for call in calls) + + +def test_close_meter_closes_open_bindings_and_enforces_auth_csrf(meters_client): + client, engine = meters_client + started = datetime.now(UTC) - timedelta(hours=2) + _login(client) + with patch("app.api.routes.api.meters.recompute_range", return_value=0): + declared = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=_declare_payload( + label="closable", started_at=started.isoformat(), reason="initial", + )) + assert declared.status_code == 201 + _add_bound_channel(engine, meter_id=declared.json()["id"], started_at=started) + boundary = started + timedelta(hours=1) + assert client.post(f"/api/energy/meters/{declared.json()['id']}/close", json={"ended_at": boundary.isoformat()}).status_code == 403 + with patch("app.api.routes.api.meters.recompute_range", return_value=0): + closed = client.post(f"/api/energy/meters/{declared.json()['id']}/close", headers={"X-CSRF-Token": _CSRF}, json={ + "ended_at": boundary.isoformat(), + }) + assert closed.status_code == 200 + with Session(engine) as session: + binding = session.scalar(select(MeterSourceBinding)) + assert binding is not None and binding.ended_at.replace(tzinfo=UTC) == boundary + + +@pytest.mark.parametrize("commodity,unit", [("electricity", "kWh"), ("heating", "GJ"), ("hot_water", "m³")]) +@pytest.mark.parametrize("operation", ["close", "declare"]) +def test_lifecycle_rejects_retained_closed_binding_beyond_proposed_end( + meters_client, mock_publish_discovery, commodity, unit, operation, +): + """Close and declare fail closed before recompute or HA for every commodity.""" + client, engine = meters_client + _login(client) + start = datetime.now(UTC) - timedelta(hours=3) + boundary = start + timedelta(hours=1) + retained_end = start + timedelta(hours=2) + declared = client.post( + "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, + json=_declare_payload(label="retained-history", started_at=start.isoformat(), reason="initial", commodity=commodity), + ) + assert declared.status_code == 201 + meter_id = declared.json()["id"] + _add_bound_channel(engine, meter_id=meter_id, started_at=start, ended_at=retained_end, unit=unit) + mock_publish_discovery.reset_mock() + + with patch("app.api.routes.api.meters._recompute_commodity", side_effect=AssertionError("must not recompute")): + if operation == "close": + response = client.post( + f"/api/energy/meters/{meter_id}/close", headers={"X-CSRF-Token": _CSRF}, + json={"ended_at": boundary.isoformat()}, + ) + else: + response = client.post( + "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, + json=_declare_payload( + label="replacement", started_at=boundary.isoformat(), reason="meter_swap", commodity=commodity, + ), + ) + assert response.status_code == 422 + assert mock_publish_discovery.call_count == 0 + with Session(engine) as observer: + meter = observer.get(Meter, meter_id) + binding = observer.scalar(select(MeterSourceBinding).where(MeterSourceBinding.meter_id == meter_id)) + assert meter is not None and meter.ended_at is None + assert binding is not None and binding.ended_at.replace(tzinfo=UTC) == retained_end + assert observer.scalars(select(Meter).where(Meter.commodity == commodity)).all() == [meter] + + +def test_close_flushes_lifecycle_boundary_before_strict_recompute(meters_client): + """The strict recompute query observes the just-closed meter and binding.""" + client, engine = meters_client + _login(client) + started = datetime.now(UTC) - timedelta(hours=2) + boundary = started + timedelta(hours=1) + with patch("app.api.routes.api.meters.recompute_range", return_value=0): + declared = client.post( + "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, + json=_declare_payload(label="flush-visible", started_at=started.isoformat(), reason="initial"), + ) + meter_id = declared.json()["id"] + _add_bound_channel(engine, meter_id=meter_id, started_at=started) + + def observe(session, *_args, **kwargs): + assert kwargs == {"commit": False, "strict": True} + observed_meter = session.get(Meter, meter_id) + observed_binding = session.scalar(select(MeterSourceBinding)) + assert observed_meter is not None and observed_meter.ended_at is not None + assert observed_binding is not None and observed_binding.ended_at is not None + return 0 + + with patch("app.api.routes.api.meters.recompute_range", side_effect=observe): + response = client.post( + f"/api/energy/meters/{meter_id}/close", headers={"X-CSRF-Token": _CSRF}, + json={"ended_at": boundary.isoformat()}, + ) + assert response.status_code == 200 + + +def test_close_strict_compute_failure_rolls_back_persisted_lifecycle_state(meters_client, monkeypatch): + """A real per-period strict failure rolls back the close in a fresh Session.""" + from app.services import energy_cost + + client, engine = meters_client + _login(client) + started = datetime.now(UTC) - timedelta(hours=2) + boundary = started + timedelta(hours=1) + with patch("app.api.routes.api.meters.recompute_range", return_value=0): + declared = client.post( + "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, + json=_declare_payload(label="strict-rollback", started_at=started.isoformat(), reason="initial"), + ) + meter_id = declared.json()["id"] + _add_bound_channel(engine, meter_id=meter_id, started_at=started) + monkeypatch.setattr( + energy_cost, "compute_period", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("strict boom")) + ) + + with pytest.raises(RuntimeError, match="strict boom"): + client.post( + f"/api/energy/meters/{meter_id}/close", headers={"X-CSRF-Token": _CSRF}, + json={"ended_at": boundary.isoformat()}, + ) + with Session(engine) as observer: + assert observer.get(Meter, meter_id).ended_at is None + assert observer.scalar(select(MeterSourceBinding)).ended_at is None + + +def test_close_flush_failure_rolls_back_persisted_lifecycle_state(meters_client, monkeypatch): + """The mandatory pre-recompute flush shares the route rollback boundary.""" + client, engine = meters_client + _login(client) + started = datetime.now(UTC) - timedelta(hours=2) + with patch("app.api.routes.api.meters.recompute_range", return_value=0): + declared = client.post( + "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, + json=_declare_payload(label="flush-rollback", started_at=started.isoformat(), reason="initial"), + ) + meter_id = declared.json()["id"] + _add_bound_channel(engine, meter_id=meter_id, started_at=started) + + monkeypatch.setattr(Session, "flush", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("flush boom"))) + with pytest.raises(RuntimeError, match="flush boom"): + client.post( + f"/api/energy/meters/{meter_id}/close", headers={"X-CSRF-Token": _CSRF}, + json={"ended_at": (started + timedelta(hours=1)).isoformat()}, + ) + monkeypatch.undo() + with Session(engine) as observer: + assert observer.get(Meter, meter_id).ended_at is None + assert observer.scalar(select(MeterSourceBinding)).ended_at is None diff --git a/tests/test_energy_cost.py b/tests/test_energy_cost.py index a3ae6e1..722787c 100644 --- a/tests/test_energy_cost.py +++ b/tests/test_energy_cost.py @@ -2176,6 +2176,31 @@ class TestComputeClosedPeriods: class TestRecomputeRange: + def test_strict_mode_propagates_period_failure_without_committing( + self, energy_db: Session, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Lifecycle callers can own one rollback for writes and recompute.""" + from app.services import energy_cost + + monkeypatch.setattr( + energy_cost, "compute_period", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("boom")) + ) + + with pytest.raises(RuntimeError, match="boom"): + recompute_range(energy_db, _T0, _T1, commit=False, strict=True) + + def test_default_mode_keeps_best_effort_period_failure( + self, energy_db: Session, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Existing workers retain their tolerant recompute behaviour by default.""" + from app.services import energy_cost + + monkeypatch.setattr( + energy_cost, "compute_period", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("boom")) + ) + + assert recompute_range(energy_db, _T0, _T1, commit=False) == 0 + def test_default_commit_is_visible_to_a_new_session(self, energy_db: Session) -> None: """The public recompute API retains its standalone commit behaviour.""" _setup_manual_scenario(energy_db) diff --git a/tests/test_meter_source_api.py b/tests/test_meter_source_api.py index 36968b1..febb2a5 100644 --- a/tests/test_meter_source_api.py +++ b/tests/test_meter_source_api.py @@ -10,6 +10,7 @@ import threading import time from unittest.mock import patch +import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine, select from sqlalchemy.orm import Session @@ -53,6 +54,13 @@ def _add_channel(engine, source_uuid: str, *, key: str = "electricity") -> str: return channel.uuid +@pytest.fixture(autouse=True) +def _mock_lifecycle_recompute(monkeypatch): + """Keep lifecycle API contracts on synthetic DBs; cost engines have their own tests.""" + monkeypatch.setattr("app.api.routes.api.meters.recompute_range", lambda *args, **kwargs: 0) + monkeypatch.setattr("app.api.routes.api.meter_sources.electricity_recompute_range", lambda *args, **kwargs: 0) + + def test_source_profiles_and_crud_mask_secrets(auth_database): client, engine = _client(auth_database) with client: @@ -268,7 +276,7 @@ def test_binding_routes_and_atomic_meter_declaration(auth_database): channel_uuid = channel.uuid declaration = { - "label": "Bound meter", "started_at": "2030-01-01T00:00:00Z", "reason": "initial", + "label": "Bound meter", "started_at": "2025-01-01T00:00:00Z", "reason": "initial", "commodity": "electricity", "source_channel_uuid": channel_uuid, } created = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=declaration) @@ -278,7 +286,7 @@ def test_binding_routes_and_atomic_meter_declaration(auth_database): assert client.get(f"/api/energy/meters/{meter_id}/bindings").json()["total"] == 1 assert client.get(f"/api/energy/sources/{source_uuid}/channels").json()["items"][0]["binding_count"] == 1 - invalid = dict(declaration, label="Must roll back", started_at="2030-02-01T00:00:00Z", source_channel_uuid="missing-channel") + invalid = dict(declaration, label="Must roll back", started_at="2025-02-01T00:00:00Z", source_channel_uuid="missing-channel") assert client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=invalid).status_code == 404 assert client.get("/api/energy/meters").json()["total"] == 1 with Session(engine) as session: @@ -403,12 +411,12 @@ def test_management_reads_require_auth_and_mutations_require_csrf(auth_database) source = _create_source(client) channel_uuid = _add_channel(engine, source["uuid"]) meter = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ - "label": "CSRF meter", "started_at": "2030-01-01T00:00:00Z", "reason": "initial", + "label": "CSRF meter", "started_at": "2025-01-01T00:00:00Z", "reason": "initial", }) assert meter.status_code == 201 meter_id = meter.json()["id"] binding = client.post(f"/api/energy/meters/{meter_id}/bindings", headers={"X-CSRF-Token": _CSRF}, json={ - "source_channel_uuid": channel_uuid, "started_at": "2030-01-01T00:00:00Z", + "source_channel_uuid": channel_uuid, "started_at": "2025-01-01T00:00:00Z", }) assert binding.status_code == 201 @@ -467,11 +475,11 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth assert channels.json()["source_status"] == "online" meter = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ - "label": "Contract meter", "started_at": "2030-01-01T00:00:00Z", "reason": "initial", + "label": "Contract meter", "started_at": "2025-01-01T00:00:00Z", "reason": "initial", }) assert meter.status_code == 201 binding = client.post(f"/api/energy/meters/{meter.json()['id']}/bindings", headers={"X-CSRF-Token": _CSRF}, json={ - "source_channel_uuid": channel_uuid, "started_at": "2030-01-01T00:00:00Z", + "source_channel_uuid": channel_uuid, "started_at": "2025-01-01T00:00:00Z", }) assert binding.status_code == 201 binding_item = client.get(f"/api/energy/meters/{meter.json()['id']}/bindings").json()["items"][0] @@ -492,38 +500,178 @@ def test_binding_patch_omitted_null_and_adjacent_half_open_boundaries(auth_datab source = _create_source(client) channel_uuid = _add_channel(engine, source["uuid"]) meter = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ - "label": "Timeline meter", "started_at": "2030-01-01T00:00:00Z", "reason": "initial", + "label": "Timeline meter", "started_at": "2025-01-01T00:00:00Z", "reason": "initial", }) assert meter.status_code == 201 meter_id = meter.json()["id"] first = client.post(f"/api/energy/meters/{meter_id}/bindings", headers={"X-CSRF-Token": _CSRF}, json={ - "source_channel_uuid": channel_uuid, "started_at": "2030-01-01T00:00:00Z", - "ended_at": "2030-02-01T00:00:00Z", + "source_channel_uuid": channel_uuid, "started_at": "2025-01-01T00:00:00Z", + "ended_at": "2025-02-01T00:00:00Z", }) assert first.status_code == 201 first_uuid = first.json()["uuid"] corrected = client.patch(f"/api/energy/bindings/{first_uuid}", headers={"X-CSRF-Token": _CSRF}, json={ - "started_at": "2030-01-02T00:00:00Z", + "started_at": "2025-01-02T00:00:00Z", }) assert corrected.status_code == 200 - assert corrected.json()["ended_at"] == "2030-02-01T00:00:00" + assert corrected.json()["ended_at"] == "2025-02-01T00:00:00" unchanged = client.patch(f"/api/energy/bindings/{first_uuid}", headers={"X-CSRF-Token": _CSRF}, json={}) assert unchanged.status_code == 200 - assert unchanged.json()["ended_at"] == "2030-02-01T00:00:00" + assert unchanged.json()["ended_at"] == "2025-02-01T00:00:00" reopened = client.patch(f"/api/energy/bindings/{first_uuid}", headers={"X-CSRF-Token": _CSRF}, json={"ended_at": None}) assert reopened.status_code == 200 assert reopened.json()["ended_at"] is None reclosed = client.patch(f"/api/energy/bindings/{first_uuid}", headers={"X-CSRF-Token": _CSRF}, json={ - "ended_at": "2030-02-01T00:00:00Z", + "ended_at": "2025-02-01T00:00:00Z", }) assert reclosed.status_code == 200 adjacent = client.post(f"/api/energy/meters/{meter_id}/bindings", headers={"X-CSRF-Token": _CSRF}, json={ - "source_channel_uuid": channel_uuid, "started_at": "2030-02-01T00:00:00Z", + "source_channel_uuid": channel_uuid, "started_at": "2025-02-01T00:00:00Z", }) assert adjacent.status_code == 201 engine.dispose() +def test_binding_create_and_meter_declare_reject_future_boundaries(auth_database): + client, engine = _client(auth_database) + with client: + _login(client) + source = _create_source(client) + channel_uuid = _add_channel(engine, source["uuid"]) + future = (datetime.now(UTC) + timedelta(minutes=5)).isoformat() + assert client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ + "label": "future", "started_at": future, "reason": "initial", + }).status_code == 422 + past = (datetime.now(UTC) - timedelta(hours=1)).isoformat() + meter = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ + "label": "past", "started_at": past, "reason": "initial", + }) + assert meter.status_code == 201 + assert client.post(f"/api/energy/meters/{meter.json()['id']}/bindings", headers={"X-CSRF-Token": _CSRF}, json={ + "source_channel_uuid": channel_uuid, "started_at": future, + }).status_code == 422 + engine.dispose() + + +def test_transfer_recovers_stranded_previous_meter_same_channel(auth_database): + """A retained open row on a closed predecessor is recoverable in one request.""" + client, engine = _client(auth_database) + with client: + _login(client) + source = _create_source(client) + channel_uuid = _add_channel(engine, source["uuid"]) + start = datetime.now(UTC) - timedelta(days=2) + boundary = start + timedelta(days=1) + with Session(engine) as session: + channel = session.scalar(select(MeterSourceChannel).where(MeterSourceChannel.uuid == channel_uuid)) + assert channel is not None + old = Meter(label="old", commodity="electricity", started_at=start, ended_at=boundary, + reason="meter_swap", created_at=start) + target = Meter(label="target", commodity="electricity", started_at=boundary, + reason="meter_swap", created_at=boundary) + session.add_all([old, target]) + session.flush() + stranded = MeterSourceBinding(meter_id=old.id, channel_id=channel.id, started_at=start, + created_at=start, updated_at=start) + session.add(stranded) + session.commit() + target_id, stranded_uuid = target.id, stranded.uuid + response = client.post(f"/api/energy/meters/{target_id}/bindings/transfer", headers={"X-CSRF-Token": _CSRF}, json={ + "from_binding_uuid": stranded_uuid, "to_source_channel_uuid": channel_uuid, + "effective_at": (boundary + timedelta(hours=2)).isoformat(), + }) + assert response.status_code == 200 + assert response.json()["closed_binding"]["ended_at"] is not None + assert response.json()["created_binding"]["started_at"].startswith((boundary + timedelta(hours=2)).isoformat()[:19]) + engine.dispose() + + +def test_transfer_recovers_unique_gapped_predecessor_after_commit(auth_database, monkeypatch): + """Recovery closes at the old epoch end and recomputes from that earliest boundary.""" + from app.api.routes.api import meter_sources + + client, engine = _client(auth_database) + calls: list[tuple[datetime, datetime, bool, bool]] = [] + published: list[bool] = [] + monkeypatch.setattr( + meter_sources, "electricity_recompute_range", + lambda _db, start, end, *, commit, strict: calls.append((start, end, commit, strict)) or 0, + ) + monkeypatch.setattr(meter_sources, "_republish_after_commit", lambda _db: published.append(True)) + with client: + _login(client) + source = _create_source(client) + channel_uuid = _add_channel(engine, source["uuid"]) + start = datetime.now(UTC) - timedelta(days=3) + old_end = start + timedelta(days=1) + target_start = old_end + timedelta(hours=3) + effective_at = target_start + timedelta(hours=1) + with Session(engine) as session: + channel = session.scalar(select(MeterSourceChannel).where(MeterSourceChannel.uuid == channel_uuid)) + assert channel is not None + old = Meter(label="old", commodity="electricity", started_at=start, ended_at=old_end, + reason="meter_swap", created_at=start) + target = Meter(label="target", commodity="electricity", started_at=target_start, + reason="initial", created_at=target_start) + session.add_all([old, target]) + session.flush() + stranded = MeterSourceBinding(meter_id=old.id, channel_id=channel.id, started_at=start, + created_at=start, updated_at=start) + session.add(stranded) + session.commit() + target_id, stranded_uuid = target.id, stranded.uuid + response = client.post( + f"/api/energy/meters/{target_id}/bindings/transfer", headers={"X-CSRF-Token": _CSRF}, + json={"from_binding_uuid": stranded_uuid, "to_source_channel_uuid": channel_uuid, + "effective_at": effective_at.isoformat()}, + ) + assert response.status_code == 200 + assert response.json()["closed_binding"]["ended_at"].startswith(old_end.isoformat()[:19]) + assert response.json()["created_binding"]["started_at"].startswith(effective_at.isoformat()[:19]) + assert calls and calls[0][0] == old_end and calls[0][2:] == (False, True) + assert published == [True] + engine.dispose() + + +def test_transfer_rejects_intervening_meter_without_changing_stranded_binding(auth_database): + """A non-predecessor recovery request is fail-closed and rolls back cleanly.""" + client, engine = _client(auth_database) + with client: + _login(client) + source = _create_source(client) + channel_uuid = _add_channel(engine, source["uuid"]) + start = datetime.now(UTC) - timedelta(days=4) + old_end = start + timedelta(days=1) + target_start = old_end + timedelta(days=2) + with Session(engine) as session: + channel = session.scalar(select(MeterSourceChannel).where(MeterSourceChannel.uuid == channel_uuid)) + assert channel is not None + old = Meter(label="old", commodity="electricity", started_at=start, ended_at=old_end, + reason="meter_swap", created_at=start) + intervening = Meter(label="intervening", commodity="electricity", started_at=old_end, + ended_at=target_start, reason="other", created_at=old_end) + target = Meter(label="target", commodity="electricity", started_at=target_start, + reason="initial", created_at=target_start) + session.add_all([old, intervening, target]) + session.flush() + stranded = MeterSourceBinding(meter_id=old.id, channel_id=channel.id, started_at=start, + created_at=start, updated_at=start) + session.add(stranded) + session.commit() + target_id, stranded_uuid, stranded_id = target.id, stranded.uuid, stranded.id + response = client.post( + f"/api/energy/meters/{target_id}/bindings/transfer", headers={"X-CSRF-Token": _CSRF}, + json={"from_binding_uuid": stranded_uuid, "to_source_channel_uuid": channel_uuid, + "effective_at": (target_start + timedelta(hours=1)).isoformat()}, + ) + assert response.status_code == 422 + with Session(engine) as observer: + binding = observer.get(MeterSourceBinding, stranded_id) + assert binding is not None and binding.ended_at is None + assert observer.scalars(select(MeterSourceBinding)).all() == [binding] + engine.dispose() + + def test_warmtelink_discover_and_minute_history_are_bounded_and_private(auth_database, monkeypatch): """Discover delegates to the manager; readings expose accepted minute samples only.""" from app.api.routes.api import meter_sources diff --git a/tests/test_meter_source_services.py b/tests/test_meter_source_services.py index 9787ed2..44efd87 100644 --- a/tests/test_meter_source_services.py +++ b/tests/test_meter_source_services.py @@ -26,6 +26,7 @@ from app.services.meter_sources import ( create_binding_for_meter_swap, create_source, delete_source, + transfer_binding, upsert_discovered_channel, ) @@ -184,7 +185,7 @@ def test_meter_swap_hands_off_only_the_previous_meter_binding(session): boundary = start + timedelta(days=1) old_meter = _meter(session, "heating", "old") old_meter.started_at = start - old_meter.ended_at = boundary + old_meter.ended_at = None new_meter = Meter( label="new", commodity="heating", @@ -198,6 +199,7 @@ def test_meter_swap_hands_off_only_the_previous_meter_binding(session): old_binding = create_binding( session, meter_id=old_meter.id, channel_id=channel.id, started_at=start ) + old_meter.ended_at = boundary session.flush() new_binding = create_binding_for_meter_swap( @@ -241,7 +243,7 @@ def test_meter_swap_rejects_ambiguous_channel_without_closing_any_binding(sessio boundary = start + timedelta(days=1) old_meter = _meter(session, "heating", "old") old_meter.started_at = start - old_meter.ended_at = boundary + old_meter.ended_at = None new_meter = Meter( label="new", commodity="heating", @@ -254,6 +256,7 @@ def test_meter_swap_rejects_ambiguous_channel_without_closing_any_binding(sessio session.flush() _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") old_binding = create_binding(session, meter_id=old_meter.id, channel_id=channel.id, started_at=start) + old_meter.ended_at = boundary session.add( MeterSourceBinding( meter_id=other_meter.id, @@ -304,6 +307,119 @@ def test_meter_swap_rejects_incompatible_channel(session): ) +def test_transfer_closes_and_opens_at_shared_boundary(session): + start = datetime(2026, 8, 22, tzinfo=UTC) + meter = _meter(session, "heating") + meter.started_at = start + _, old_channel = _source_and_channel(session, "warmtelink_serial", "GJ") + _, new_channel = _source_and_channel(session, "warmtelink_serial", "GJ") + old = create_binding(session, meter_id=meter.id, channel_id=old_channel.id, started_at=start) + session.flush() + + closed, created = transfer_binding( + session, target_meter_id=meter.id, from_binding_id=old.id, + to_channel_id=new_channel.id, effective_at=start + timedelta(hours=1), + ) + + assert closed.ended_at == created.started_at == start + timedelta(hours=1) + assert created.channel_id == new_channel.id + + +def test_transfer_rejects_future_without_mutating_source_binding(session): + start = datetime.now(UTC) - timedelta(hours=2) + meter = _meter(session, "heating") + meter.started_at = start + _, old_channel = _source_and_channel(session, "warmtelink_serial", "GJ") + _, new_channel = _source_and_channel(session, "warmtelink_serial", "GJ") + old = create_binding(session, meter_id=meter.id, channel_id=old_channel.id, started_at=start) + session.flush() + + with pytest.raises(BindingValidationError, match="future"): + transfer_binding(session, target_meter_id=meter.id, from_binding_id=old.id, + to_channel_id=new_channel.id, effective_at=datetime.now(UTC) + timedelta(minutes=1)) + assert old.ended_at is None + + +def test_cross_meter_transfer_recovers_stranded_same_channel(session): + start = datetime(2026, 8, 20, tzinfo=UTC) + boundary = start + timedelta(days=1) + old = _meter(session, "heating", "old") + old.started_at, old.ended_at = start, None + new = Meter(label="new", commodity="heating", started_at=boundary, reason="meter_swap", created_at=boundary) + session.add(new) + session.flush() + _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") + stranded = create_binding(session, meter_id=old.id, channel_id=channel.id, started_at=start) + old.ended_at = boundary + session.flush() + + closed, created = transfer_binding( + session, target_meter_id=new.id, from_binding_id=stranded.id, + to_channel_id=channel.id, effective_at=boundary + timedelta(hours=2), + ) + + assert closed.ended_at == boundary + assert created.started_at == boundary + timedelta(hours=2) + assert created.channel_id == channel.id + + +def test_cross_meter_transfer_recovers_unique_gapped_predecessor(session): + """A deliberate no-meter gap does not make the latest predecessor ambiguous.""" + start = datetime(2026, 8, 20, tzinfo=UTC) + old_end = start + timedelta(days=1) + target_start = old_end + timedelta(hours=3) + old = _meter(session, "heating", "old") + old.started_at, old.ended_at = start, old_end + target = Meter(label="target", commodity="heating", started_at=target_start, + reason="initial", created_at=target_start) + session.add(target) + session.flush() + _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") + stranded = MeterSourceBinding( + meter_id=old.id, channel_id=channel.id, started_at=start, + created_at=start, updated_at=start, + ) + session.add(stranded) + session.flush() + + closed, created = transfer_binding( + session, target_meter_id=target.id, from_binding_id=stranded.id, + to_channel_id=channel.id, effective_at=target_start + timedelta(hours=1), + ) + + assert closed.ended_at == old_end + assert created.started_at == target_start + timedelta(hours=1) + + +def test_cross_meter_transfer_rejects_closed_source_and_ambiguous_predecessor(session): + start = datetime(2026, 8, 20, tzinfo=UTC) + boundary = start + timedelta(days=1) + old = _meter(session, "heating", "old") + old.started_at, old.ended_at = start, None + target = Meter(label="target", commodity="heating", started_at=boundary, reason="meter_swap", created_at=boundary) + session.add(target) + session.flush() + _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") + source = create_binding(session, meter_id=old.id, channel_id=channel.id, started_at=start) + old.ended_at = boundary + session.flush() + close_binding(session, source.id, ended_at=boundary - timedelta(hours=1)) + with pytest.raises(BindingValidationError, match="open binding"): + transfer_binding(session, target_meter_id=target.id, from_binding_id=source.id, + to_channel_id=channel.id, effective_at=boundary) + assert source.ended_at == boundary - timedelta(hours=1) + + source.ended_at = None # synthetic retained bad row, exactly the recovery input. + duplicate = Meter(label="duplicate", commodity="heating", started_at=start + timedelta(hours=1), + ended_at=boundary, reason="other", created_at=start) + session.add(duplicate) + session.flush() + with pytest.raises(BindingValidationError, match="unique immediately preceding"): + transfer_binding(session, target_meter_id=target.id, from_binding_id=source.id, + to_channel_id=channel.id, effective_at=boundary) + assert source.ended_at is None + + def test_source_delete_is_restricted_by_discovered_channel(session): source, _ = _source_and_channel(session, "dsmr_mqtt", "kWh") with pytest.raises(SourceDeleteRestrictedError): diff --git a/tests/test_meters.py b/tests/test_meters.py index d4e04ee..d0c4c09 100644 --- a/tests/test_meters.py +++ b/tests/test_meters.py @@ -25,7 +25,10 @@ from sqlalchemy import create_engine, event as sa_event from sqlalchemy.orm import Session from app.models.energy import Meter +from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel from app.services.meters import ( + _as_utc, + close_meter, MeterIntervalError, MeterOverlapError, declare_meter, @@ -35,6 +38,62 @@ from app.services.meters import ( ) +def test_declare_and_close_reject_future_boundaries(session: Session): + future = datetime.now(UTC) + timedelta(minutes=5) + with pytest.raises(MeterIntervalError, match="future"): + declare_meter(session, label="future", started_at=future, reason="initial") + meter = _make_meter(session, started_at=datetime.now(UTC) - timedelta(hours=1)) + with pytest.raises(MeterIntervalError, match="future"): + close_meter(session, meter, ended_at=future) + assert meter.ended_at is None + + +def test_close_and_declare_reject_boundary_before_retained_closed_binding(session: Session): + """Lifecycle writes must not silently shorten immutable binding history.""" + start = datetime.now(UTC) - timedelta(hours=3) + proposed_end = start + timedelta(hours=1) + meter = _make_meter(session, started_at=start) + binding = _make_binding(session, meter, started_at=start, ended_at=start + timedelta(hours=2)) + + with pytest.raises(MeterIntervalError, match="Closed binding extends"): + close_meter(session, meter, ended_at=proposed_end) + assert meter.ended_at is None + assert binding.ended_at == start + timedelta(hours=2) + + with pytest.raises(MeterIntervalError, match="Closed binding extends"): + declare_meter(session, label="replacement", started_at=proposed_end, reason="meter_swap") + assert meter.ended_at is None + assert session.query(Meter).count() == 1 + + +def test_update_rejects_future_started_at_before_mutating_other_fields(session: Session): + """A future correction must not leak label/note changes into the Session.""" + meter = _make_meter( + session, + started_at=datetime.now(UTC) - timedelta(hours=1), + label="Original", + note="Original note", + ) + session.commit() + meter_id = meter.id + + with pytest.raises(MeterIntervalError, match="future"): + update_meter( + session, + meter, + label="Changed", + note="Changed note", + started_at=datetime.now(UTC) + timedelta(minutes=5), + ) + session.rollback() + + with Session(session.bind) as observer: + unchanged = observer.get(Meter, meter_id) + assert unchanged is not None + assert unchanged.label == "Original" + assert unchanged.note == "Original note" + + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @@ -112,6 +171,28 @@ def _make_meter( return m +def _make_binding(session: Session, meter: Meter, *, started_at: datetime, ended_at: datetime | None = None): + source = MeterSource( + name=f"source-{meter.id}", kind="dsmr_mqtt", enabled=True, config={}, status="online", + created_at=started_at, updated_at=started_at, + ) + session.add(source) + session.flush() + channel = MeterSourceChannel( + source_id=source.id, channel_key=f"channel-{meter.id}", label="Total", unit="kWh", + created_at=started_at, updated_at=started_at, + ) + session.add(channel) + session.flush() + binding = MeterSourceBinding( + meter_id=meter.id, channel_id=channel.id, started_at=started_at, ended_at=ended_at, + created_at=started_at, updated_at=started_at, + ) + session.add(binding) + session.flush() + return binding + + # --------------------------------------------------------------------------- # 1. meter_at — half-open interval semantics # --------------------------------------------------------------------------- @@ -637,3 +718,36 @@ class TestUpdateMeter: from app.services.meters import _as_utc fetched = session.get(Meter, m.id) assert _as_utc(fetched.started_at) == _as_utc(earlier) + + @pytest.mark.parametrize("shift", ["later", "earlier"]) + def test_update_started_at_rejects_boundary_shift_that_strands_binding(self, session: Session, shift: str): + """A correction must not create an out-of-epoch binding on either adjacent meter.""" + boundary = _T0 + timedelta(days=10) + prev = _make_meter(session, started_at=_T0, ended_at=boundary, label="Prev") + current = _make_meter(session, started_at=boundary, ended_at=None, label="Current") + if shift == "later": + binding = _make_binding(session, current, started_at=boundary) + proposed = boundary + timedelta(days=1) + else: + binding = _make_binding(session, prev, started_at=_T0, ended_at=boundary) + proposed = boundary - timedelta(days=1) + session.commit() + + prev_id = prev.id + current_id = current.id + binding_id = binding.id + binding_started_at = binding.started_at + binding_ended_at = binding.ended_at + with pytest.raises(MeterIntervalError, match="Binding"): + update_meter(session, current, started_at=proposed) + session.rollback() + + with Session(session.bind) as observer: + assert _as_utc(observer.get(Meter, prev_id).ended_at) == boundary + assert _as_utc(observer.get(Meter, current_id).started_at) == boundary + observed_binding = observer.get(MeterSourceBinding, binding_id) + assert observed_binding is not None + assert _as_utc(observed_binding.started_at) == _as_utc(binding_started_at) + assert ( + _as_utc(observed_binding.ended_at) if observed_binding.ended_at is not None else None + ) == (_as_utc(binding_ended_at) if binding_ended_at is not None else None)