640 lines
26 KiB
Python
640 lines
26 KiB
Python
"""Tests for M7-T02: Meter service layer.
|
|
|
|
Coverage
|
|
--------
|
|
1. ``meter_at`` — half-open interval boundary semantics.
|
|
2. ``declare_meter`` — first declaration (no active meter), normal swap, backdate rejection.
|
|
3. Mutual exclusion — each commodity has at most one active meter after swaps.
|
|
4. Interval continuity — old meter's ended_at == new meter's started_at after swap.
|
|
5. Different commodities are independent (electricity swap doesn't touch gas meters).
|
|
6. ``list_meters`` — ordering and commodity filtering.
|
|
7. ``update_meter`` — label/note update, started_at retroactive correction with
|
|
interval consistency, and validation errors.
|
|
8. ``update_meter`` — first-meter (no previous) retroactive started_at change.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine, event as sa_event
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.energy import Meter
|
|
from app.services.meters import (
|
|
MeterIntervalError,
|
|
MeterOverlapError,
|
|
declare_meter,
|
|
list_meters,
|
|
meter_at,
|
|
update_meter,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_app_alembic_config(database_url: str) -> Config:
|
|
cfg = Config("alembic_app.ini")
|
|
cfg.set_main_option("sqlalchemy.url", database_url)
|
|
return cfg
|
|
|
|
|
|
def _engine_with_fk(db_url: str):
|
|
"""Create a SQLAlchemy engine with SQLite FK enforcement enabled."""
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
|
|
@sa_event.listens_for(engine, "connect")
|
|
def _enable_fk(dbapi_conn, _rec):
|
|
cursor = dbapi_conn.cursor()
|
|
cursor.execute("PRAGMA foreign_keys = ON")
|
|
cursor.close()
|
|
|
|
return engine
|
|
|
|
|
|
@pytest.fixture()
|
|
def meter_db(tmp_path: Path):
|
|
"""Temporary SQLite DB upgraded to the current Alembic head with FK enforcement."""
|
|
db_path = tmp_path / "meter_service_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
alembic_cfg = _make_app_alembic_config(db_url)
|
|
command.upgrade(alembic_cfg, "head")
|
|
engine = _engine_with_fk(db_url)
|
|
yield engine
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.fixture()
|
|
def session(meter_db):
|
|
"""Provide a single SQLAlchemy session for a test, auto-rolling back on exit."""
|
|
with Session(meter_db) as s:
|
|
yield s
|
|
# Tests that commit explicitly are fine; for read-only tests the context
|
|
# manager handles cleanup.
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_T0 = datetime(2026, 1, 1, 12, 0, 0, tzinfo=UTC) # base timestamp for tests
|
|
|
|
|
|
def _make_meter(
|
|
session: Session,
|
|
*,
|
|
started_at: datetime,
|
|
ended_at: datetime | None = None,
|
|
label: str = "Test meter",
|
|
commodity: str = "electricity",
|
|
reason: str = "initial",
|
|
note: str | None = None,
|
|
) -> Meter:
|
|
"""Insert a Meter row directly (bypassing service logic) for test setup."""
|
|
m = Meter(
|
|
label=label,
|
|
commodity=commodity,
|
|
started_at=started_at,
|
|
ended_at=ended_at,
|
|
reason=reason,
|
|
note=note,
|
|
created_at=datetime.now(UTC),
|
|
)
|
|
session.add(m)
|
|
session.flush()
|
|
return m
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. meter_at — half-open interval semantics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMeterAt:
|
|
def test_returns_none_when_no_meters(self, session: Session):
|
|
"""meter_at must return None when the table is empty."""
|
|
assert meter_at(session, _T0) is None
|
|
|
|
def test_returns_active_meter_for_ts_after_start(self, session: Session):
|
|
"""An active meter (ended_at IS NULL) covers any ts ≥ started_at."""
|
|
m = _make_meter(session, started_at=_T0, ended_at=None)
|
|
result = meter_at(session, _T0 + timedelta(hours=1))
|
|
assert result is not None
|
|
assert result.id == m.id
|
|
|
|
def test_exact_started_at_is_inclusive(self, session: Session):
|
|
"""ts == started_at must be covered by that meter (left-closed boundary)."""
|
|
m = _make_meter(session, started_at=_T0, ended_at=None)
|
|
result = meter_at(session, _T0)
|
|
assert result is not None
|
|
assert result.id == m.id
|
|
|
|
def test_ts_before_started_at_returns_none(self, session: Session):
|
|
"""ts < started_at must not be covered."""
|
|
_make_meter(session, started_at=_T0, ended_at=None)
|
|
result = meter_at(session, _T0 - timedelta(seconds=1))
|
|
assert result is None
|
|
|
|
def test_exact_ended_at_is_exclusive(self, session: Session):
|
|
"""ts == ended_at must NOT be covered by the closed meter (right-open boundary)."""
|
|
t1 = _T0 + timedelta(hours=2)
|
|
_make_meter(session, started_at=_T0, ended_at=t1, label="Old meter")
|
|
# The new active meter starts exactly at t1.
|
|
new = _make_meter(session, started_at=t1, ended_at=None, label="New meter")
|
|
|
|
result = meter_at(session, t1)
|
|
assert result is not None
|
|
assert result.id == new.id
|
|
|
|
def test_ts_just_before_ended_at_is_covered(self, session: Session):
|
|
"""ts just before ended_at must still be covered by the closing meter."""
|
|
t1 = _T0 + timedelta(hours=2)
|
|
m = _make_meter(session, started_at=_T0, ended_at=t1)
|
|
result = meter_at(session, t1 - timedelta(seconds=1))
|
|
assert result is not None
|
|
assert result.id == m.id
|
|
|
|
def test_commodity_filter(self, session: Session):
|
|
"""meter_at must only return the meter for the requested commodity."""
|
|
m_elec = _make_meter(session, started_at=_T0, commodity="electricity")
|
|
_make_meter(session, started_at=_T0, commodity="gas")
|
|
|
|
result = meter_at(session, _T0, commodity="electricity")
|
|
assert result is not None
|
|
assert result.id == m_elec.id
|
|
|
|
result_gas = meter_at(session, _T0, commodity="gas")
|
|
assert result_gas is not None
|
|
assert result_gas.commodity == "gas"
|
|
|
|
def test_no_meter_for_unknown_commodity(self, session: Session):
|
|
"""meter_at returns None when no meter exists for the requested commodity."""
|
|
_make_meter(session, started_at=_T0, commodity="electricity")
|
|
assert meter_at(session, _T0, commodity="heating") is None
|
|
|
|
def test_two_contiguous_epochs_correct_routing(self, session: Session):
|
|
"""With two contiguous meters, meter_at routes each ts to the correct epoch."""
|
|
t1 = _T0 + timedelta(hours=3)
|
|
m0 = _make_meter(session, started_at=_T0, ended_at=t1, label="Meter 0")
|
|
m1 = _make_meter(session, started_at=t1, ended_at=None, label="Meter 1")
|
|
|
|
# ts in first epoch
|
|
assert meter_at(session, _T0 + timedelta(hours=1)).id == m0.id
|
|
# ts exactly at boundary → second epoch
|
|
assert meter_at(session, t1).id == m1.id
|
|
# ts in second epoch
|
|
assert meter_at(session, t1 + timedelta(hours=1)).id == m1.id
|
|
|
|
def test_equal_started_at_swap_meter_at_still_returns_new_active(self, session: Session):
|
|
"""Regression: after equal-timestamp swap, meter_at must return the new active meter.
|
|
|
|
When declare_meter is called with started_at == active.started_at (the
|
|
"equal-timestamp replace" allowed by §3.5), the old meter becomes a
|
|
zero-width epoch [T0, T0). A previous bug caused meter_at to select
|
|
the zero-width row first (same started_at, lower rowid) and then fail
|
|
the upper-bound check, returning None for *any* ts >= T0. This test
|
|
pins the correct behaviour: meter_at(T0) and meter_at(T0+δ) must both
|
|
return the new active meter, not None.
|
|
"""
|
|
# Declare first meter at T0.
|
|
declare_meter(session, label="M1 (original)", started_at=_T0, reason="initial")
|
|
session.commit()
|
|
|
|
# Declare second meter at *the same* T0 — equal-timestamp swap.
|
|
new_m = declare_meter(
|
|
session, label="M2 (replacement)", started_at=_T0, reason="meter_swap"
|
|
)
|
|
session.commit()
|
|
|
|
# meter_at at exactly T0 must return the new active meter.
|
|
result_at_T0 = meter_at(session, _T0)
|
|
assert result_at_T0 is not None, (
|
|
"meter_at(T0) returned None after equal-timestamp swap; "
|
|
"the new active meter should cover T0"
|
|
)
|
|
assert result_at_T0.id == new_m.id, (
|
|
f"meter_at(T0) returned meter id={result_at_T0.id} (label={result_at_T0.label!r}), "
|
|
f"expected id={new_m.id} (the new active meter)"
|
|
)
|
|
|
|
# meter_at slightly after T0 must also return the new active meter.
|
|
result_after_T0 = meter_at(session, _T0 + timedelta(seconds=1))
|
|
assert result_after_T0 is not None, (
|
|
"meter_at(T0+1s) returned None after equal-timestamp swap"
|
|
)
|
|
assert result_after_T0.id == new_m.id, (
|
|
f"meter_at(T0+1s) returned meter id={result_after_T0.id}, "
|
|
f"expected id={new_m.id} (the new active meter)"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2 & 3. declare_meter — first declaration, swap, mutual exclusion
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeclareMeter:
|
|
def test_first_declaration_no_active(self, session: Session):
|
|
"""Declaring the first meter must create an active meter with ended_at IS NULL."""
|
|
m = declare_meter(
|
|
session,
|
|
label="Initial meter",
|
|
started_at=_T0,
|
|
reason="initial",
|
|
)
|
|
session.commit()
|
|
|
|
fetched = session.get(Meter, m.id)
|
|
assert fetched is not None
|
|
assert fetched.ended_at is None
|
|
assert fetched.commodity == "electricity"
|
|
assert fetched.reason == "initial"
|
|
|
|
def test_swap_closes_old_meter(self, session: Session):
|
|
"""Declaring a second meter must close the previous active meter."""
|
|
first = declare_meter(
|
|
session, label="First meter", started_at=_T0, reason="initial"
|
|
)
|
|
session.commit()
|
|
first_id = first.id
|
|
|
|
t1 = _T0 + timedelta(days=30)
|
|
second = declare_meter(
|
|
session, label="Second meter", started_at=t1, reason="meter_swap"
|
|
)
|
|
session.commit()
|
|
|
|
# First meter must now be closed at exactly t1.
|
|
closed = session.get(Meter, first_id)
|
|
assert closed.ended_at is not None
|
|
from app.services.meters import _as_utc
|
|
assert _as_utc(closed.ended_at) == _as_utc(t1)
|
|
|
|
# Second meter must be active.
|
|
assert second.ended_at is None
|
|
|
|
def test_swap_interval_contiguous(self, session: Session):
|
|
"""Old meter's ended_at must exactly equal new meter's started_at after swap."""
|
|
declare_meter(session, label="M1", started_at=_T0, reason="initial")
|
|
session.commit()
|
|
|
|
t1 = _T0 + timedelta(days=10)
|
|
new_m = declare_meter(session, label="M2", started_at=t1, reason="meter_swap")
|
|
session.commit()
|
|
|
|
# Query the old (now-closed) meter.
|
|
meters = list_meters(session, commodity="electricity")
|
|
assert len(meters) == 2
|
|
old_m = next(m for m in meters if m.id != new_m.id)
|
|
|
|
from app.services.meters import _as_utc
|
|
assert _as_utc(old_m.ended_at) == _as_utc(new_m.started_at), (
|
|
"Timeline gap or overlap: old ended_at must == new started_at"
|
|
)
|
|
|
|
def test_at_most_one_active_per_commodity_after_multiple_swaps(self, session: Session):
|
|
"""After N swaps, exactly one meter per commodity must be active."""
|
|
declare_meter(session, label="M1", started_at=_T0, reason="initial")
|
|
session.commit()
|
|
declare_meter(
|
|
session, label="M2", started_at=_T0 + timedelta(days=10), reason="meter_swap"
|
|
)
|
|
session.commit()
|
|
declare_meter(
|
|
session, label="M3", started_at=_T0 + timedelta(days=20), reason="meter_swap"
|
|
)
|
|
session.commit()
|
|
|
|
active_meters = [m for m in list_meters(session, commodity="electricity") if m.ended_at is None]
|
|
assert len(active_meters) == 1, f"Expected 1 active meter, got {len(active_meters)}"
|
|
|
|
def test_backdate_rejected(self, session: Session):
|
|
"""started_at strictly before active meter's started_at must raise MeterOverlapError."""
|
|
declare_meter(session, label="Current", started_at=_T0, reason="initial")
|
|
session.commit()
|
|
|
|
with pytest.raises(MeterOverlapError):
|
|
declare_meter(
|
|
session,
|
|
label="Too early",
|
|
started_at=_T0 - timedelta(hours=1),
|
|
reason="meter_swap",
|
|
)
|
|
|
|
def test_backdate_raises_before_any_db_write(self, session: Session):
|
|
"""When backdate is rejected, no new meter row must be written."""
|
|
declare_meter(session, label="Current", started_at=_T0, reason="initial")
|
|
session.commit()
|
|
|
|
meters_before = list_meters(session, commodity="electricity")
|
|
count_before = len(meters_before)
|
|
|
|
with pytest.raises(MeterOverlapError):
|
|
declare_meter(
|
|
session,
|
|
label="Bad meter",
|
|
started_at=_T0 - timedelta(minutes=5),
|
|
reason="meter_swap",
|
|
)
|
|
# Rollback the failed operation explicitly (simulating what the caller would do).
|
|
session.rollback()
|
|
|
|
# Re-open session to verify state.
|
|
with Session(session.get_bind()) as s2:
|
|
meters_after = list_meters(s2, commodity="electricity")
|
|
assert len(meters_after) == count_before, (
|
|
"No extra meter must be written when backdate is rejected"
|
|
)
|
|
|
|
def test_equal_started_at_allowed(self, session: Session):
|
|
"""started_at == active meter's started_at must NOT raise (equal is allowed)."""
|
|
declare_meter(session, label="M1", started_at=_T0, reason="initial")
|
|
session.commit()
|
|
|
|
# Should not raise — equal timestamps are valid (replaces meter at same instant).
|
|
new_m = declare_meter(
|
|
session, label="M2", started_at=_T0, reason="meter_swap"
|
|
)
|
|
session.commit()
|
|
assert new_m.ended_at is None
|
|
|
|
def test_note_and_commodity_stored(self, session: Session):
|
|
"""declare_meter must persist note and commodity correctly."""
|
|
m = declare_meter(
|
|
session,
|
|
label="Gas meter",
|
|
started_at=_T0,
|
|
reason="initial",
|
|
commodity="gas",
|
|
note="Rotameter serial XYZ",
|
|
)
|
|
session.commit()
|
|
|
|
fetched = session.get(Meter, m.id)
|
|
assert fetched.commodity == "gas"
|
|
assert fetched.note == "Rotameter serial XYZ"
|
|
|
|
def test_declare_meter_generates_uuid(self, session: Session):
|
|
"""declare_meter must auto-generate a non-empty uuid via ORM default."""
|
|
import re
|
|
UUID4_RE = re.compile(
|
|
r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
m = declare_meter(
|
|
session,
|
|
label="Meter with UUID",
|
|
started_at=_T0,
|
|
reason="initial",
|
|
)
|
|
session.commit()
|
|
|
|
fetched = session.get(Meter, m.id)
|
|
assert fetched is not None
|
|
assert fetched.uuid is not None, "uuid must not be None after declare_meter"
|
|
assert fetched.uuid != "", "uuid must not be empty"
|
|
assert UUID4_RE.match(fetched.uuid), (
|
|
f"uuid {fetched.uuid!r} does not look like a valid UUID v4"
|
|
)
|
|
|
|
def test_declare_meter_each_gets_distinct_uuid(self, session: Session):
|
|
"""Each declared meter must receive a distinct UUID (not duplicated)."""
|
|
m1 = declare_meter(session, label="M1", started_at=_T0, reason="initial")
|
|
session.commit()
|
|
|
|
t1 = _T0 + timedelta(days=10)
|
|
m2 = declare_meter(session, label="M2", started_at=t1, reason="meter_swap")
|
|
session.commit()
|
|
|
|
assert m1.uuid != m2.uuid, (
|
|
f"Two declared meters must have distinct UUIDs; both got {m1.uuid!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. Different commodities are independent
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCommodityIsolation:
|
|
def test_electricity_swap_does_not_affect_gas_meter(self, session: Session):
|
|
"""Swapping the electricity meter must not touch the gas meter's active state."""
|
|
declare_meter(session, label="Elec 1", started_at=_T0, reason="initial", commodity="electricity")
|
|
declare_meter(session, label="Gas 1", started_at=_T0, reason="initial", commodity="gas")
|
|
session.commit()
|
|
|
|
declare_meter(
|
|
session,
|
|
label="Elec 2",
|
|
started_at=_T0 + timedelta(days=5),
|
|
reason="meter_swap",
|
|
commodity="electricity",
|
|
)
|
|
session.commit()
|
|
|
|
# Gas meter must still be active.
|
|
gas_meters = list_meters(session, commodity="gas")
|
|
active_gas = [m for m in gas_meters if m.ended_at is None]
|
|
assert len(active_gas) == 1, "Gas meter must remain active after electricity swap"
|
|
assert active_gas[0].label == "Gas 1"
|
|
|
|
# Electricity: exactly one active.
|
|
elec_meters = list_meters(session, commodity="electricity")
|
|
active_elec = [m for m in elec_meters if m.ended_at is None]
|
|
assert len(active_elec) == 1
|
|
assert active_elec[0].label == "Elec 2"
|
|
|
|
def test_different_commodity_backdate_is_independent(self, session: Session):
|
|
"""Backdate validation is per-commodity: gas meter start does not constrain electricity."""
|
|
# Declare gas meter at a later time.
|
|
declare_meter(
|
|
session, label="Gas 1", started_at=_T0 + timedelta(days=10), reason="initial", commodity="gas"
|
|
)
|
|
session.commit()
|
|
|
|
# Declaring an electricity meter at an earlier time must succeed (no gas constraint).
|
|
m = declare_meter(
|
|
session, label="Elec 1", started_at=_T0, reason="initial", commodity="electricity"
|
|
)
|
|
session.commit()
|
|
assert m is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. list_meters
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListMeters:
|
|
def test_returns_empty_list_when_no_meters(self, session: Session):
|
|
assert list_meters(session) == []
|
|
|
|
def test_ordered_by_started_at_asc(self, session: Session):
|
|
"""list_meters must return meters in ascending started_at order."""
|
|
t1 = _T0 + timedelta(days=10)
|
|
t2 = _T0 + timedelta(days=20)
|
|
declare_meter(session, label="M1", started_at=_T0, reason="initial")
|
|
session.commit()
|
|
declare_meter(session, label="M2", started_at=t1, reason="meter_swap")
|
|
session.commit()
|
|
declare_meter(session, label="M3", started_at=t2, reason="meter_swap")
|
|
session.commit()
|
|
|
|
meters = list_meters(session, commodity="electricity")
|
|
assert [m.label for m in meters] == ["M1", "M2", "M3"]
|
|
|
|
def test_commodity_filter_returns_only_matching(self, session: Session):
|
|
"""list_meters with commodity kwarg must filter correctly."""
|
|
declare_meter(session, label="Elec", started_at=_T0, reason="initial", commodity="electricity")
|
|
declare_meter(session, label="Gas", started_at=_T0, reason="initial", commodity="gas")
|
|
session.commit()
|
|
|
|
elec = list_meters(session, commodity="electricity")
|
|
gas = list_meters(session, commodity="gas")
|
|
all_meters = list_meters(session)
|
|
|
|
assert len(elec) == 1 and elec[0].commodity == "electricity"
|
|
assert len(gas) == 1 and gas[0].commodity == "gas"
|
|
assert len(all_meters) == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 7 & 8. update_meter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUpdateMeter:
|
|
def test_update_label(self, session: Session):
|
|
"""update_meter must update label without touching other fields."""
|
|
m = _make_meter(session, started_at=_T0)
|
|
update_meter(session, m, label="New label")
|
|
session.commit()
|
|
|
|
fetched = session.get(Meter, m.id)
|
|
assert fetched.label == "New label"
|
|
assert fetched.note is None # unchanged
|
|
|
|
def test_update_note(self, session: Session):
|
|
"""update_meter must update note without touching other fields."""
|
|
m = _make_meter(session, started_at=_T0)
|
|
update_meter(session, m, note="Some note")
|
|
session.commit()
|
|
|
|
fetched = session.get(Meter, m.id)
|
|
assert fetched.note == "Some note"
|
|
assert fetched.label == "Test meter" # unchanged
|
|
|
|
def test_update_label_and_note_simultaneously(self, session: Session):
|
|
"""update_meter must update both label and note in a single call."""
|
|
m = _make_meter(session, started_at=_T0)
|
|
update_meter(session, m, label="Updated label", note="Updated note")
|
|
session.commit()
|
|
|
|
fetched = session.get(Meter, m.id)
|
|
assert fetched.label == "Updated label"
|
|
assert fetched.note == "Updated note"
|
|
|
|
def test_update_started_at_first_meter(self, session: Session):
|
|
"""Changing started_at of the first (only) active meter must work without a previous meter."""
|
|
m = _make_meter(session, started_at=_T0, ended_at=None)
|
|
new_start = _T0 + timedelta(hours=2)
|
|
|
|
update_meter(session, m, started_at=new_start)
|
|
session.commit()
|
|
|
|
fetched = session.get(Meter, m.id)
|
|
from app.services.meters import _as_utc
|
|
assert _as_utc(fetched.started_at) == _as_utc(new_start)
|
|
|
|
def test_update_started_at_updates_previous_ended_at(self, session: Session):
|
|
"""Retroactive started_at change must propagate to the previous meter's ended_at."""
|
|
t1 = _T0 + timedelta(days=10)
|
|
prev = _make_meter(session, started_at=_T0, ended_at=t1, label="Prev meter")
|
|
curr = _make_meter(session, started_at=t1, ended_at=None, label="Curr meter")
|
|
session.commit()
|
|
|
|
new_start = _T0 + timedelta(days=7) # shift boundary 3 days earlier
|
|
update_meter(session, curr, started_at=new_start)
|
|
session.commit()
|
|
|
|
from app.services.meters import _as_utc
|
|
fetched_prev = session.get(Meter, prev.id)
|
|
fetched_curr = session.get(Meter, curr.id)
|
|
|
|
# The previous meter's ended_at must now equal the new started_at.
|
|
assert _as_utc(fetched_prev.ended_at) == _as_utc(new_start), (
|
|
"Previous meter's ended_at must be updated to maintain continuity"
|
|
)
|
|
# The current meter's started_at must reflect the change.
|
|
assert _as_utc(fetched_curr.started_at) == _as_utc(new_start)
|
|
|
|
def test_update_started_at_continuity_maintained(self, session: Session):
|
|
"""After retroactive started_at change, prev.ended_at == curr.started_at (no gap)."""
|
|
t1 = _T0 + timedelta(days=10)
|
|
prev = _make_meter(session, started_at=_T0, ended_at=t1, label="Prev")
|
|
curr = _make_meter(session, started_at=t1, ended_at=None, label="Curr")
|
|
session.commit()
|
|
|
|
new_start = _T0 + timedelta(days=12) # shift boundary 2 days later
|
|
update_meter(session, curr, started_at=new_start)
|
|
session.commit()
|
|
|
|
from app.services.meters import _as_utc
|
|
fetched_prev = session.get(Meter, prev.id)
|
|
fetched_curr = session.get(Meter, curr.id)
|
|
|
|
assert _as_utc(fetched_prev.ended_at) == _as_utc(fetched_curr.started_at), (
|
|
"Timeline must remain contiguous after retroactive started_at shift"
|
|
)
|
|
|
|
def test_update_started_at_rejects_at_or_past_ended_at(self, session: Session):
|
|
"""New started_at ≥ ended_at must raise MeterIntervalError (empty/inverted epoch)."""
|
|
t1 = _T0 + timedelta(days=10)
|
|
m = _make_meter(session, started_at=_T0, ended_at=t1)
|
|
|
|
with pytest.raises(MeterIntervalError):
|
|
update_meter(session, m, started_at=t1) # == ended_at → empty epoch
|
|
|
|
with pytest.raises(MeterIntervalError):
|
|
update_meter(session, m, started_at=t1 + timedelta(hours=1)) # > ended_at
|
|
|
|
def test_update_started_at_rejects_at_or_before_prev_started_at(self, session: Session):
|
|
"""New started_at ≤ prev.started_at must raise MeterIntervalError."""
|
|
t1 = _T0 + timedelta(days=10)
|
|
_make_meter(session, started_at=_T0, ended_at=t1, label="Prev")
|
|
curr = _make_meter(session, started_at=t1, ended_at=None, label="Curr")
|
|
session.commit()
|
|
|
|
with pytest.raises(MeterIntervalError):
|
|
# exactly at prev.started_at — would collapse prev epoch to zero
|
|
update_meter(session, curr, started_at=_T0)
|
|
|
|
with pytest.raises(MeterIntervalError):
|
|
# strictly before prev.started_at — inverts prev epoch
|
|
update_meter(session, curr, started_at=_T0 - timedelta(hours=1))
|
|
|
|
def test_noop_call_does_not_raise(self, session: Session):
|
|
"""Calling update_meter with all None args must succeed without error."""
|
|
m = _make_meter(session, started_at=_T0)
|
|
result = update_meter(session, m)
|
|
assert result is m
|
|
|
|
def test_update_started_at_no_previous_meter_shift_backward(self, session: Session):
|
|
"""For the first meter (no previous), shifting started_at backward must succeed."""
|
|
m = _make_meter(session, started_at=_T0, ended_at=None)
|
|
earlier = _T0 - timedelta(days=5)
|
|
update_meter(session, m, started_at=earlier)
|
|
session.commit()
|
|
|
|
from app.services.meters import _as_utc
|
|
fetched = session.get(Meter, m.id)
|
|
assert _as_utc(fetched.started_at) == _as_utc(earlier)
|