tliu93
7e266a21eb
- app/integrations/tibber/client.py: fetch_price_range/fetch_current_price (priceInfoRange QUARTER_HOURLY, startsAt->UTC, no fixed node count assumption), TibberError/TibberAuthError; token never logged or put in exception text. - app/services/tibber_prices.py: refresh_prices upserts by starts_at (idempotent), no-op unless an active kind=tibber contract + token exist. - main.py: hourly tibber-refresh job (existing jobs/MQTT untouched). Tests added.
395 lines
13 KiB
Python
395 lines
13 KiB
Python
"""Tests for M6-T05: Tibber price refresh service (app/services/tibber_prices.py).
|
|
|
|
Covers:
|
|
1. With active tibber contract + non-empty token: ``refresh_prices`` calls the
|
|
client, upserts returned price points, and returns the count.
|
|
2. Idempotency: running ``refresh_prices`` twice does not double-insert rows;
|
|
rows are updated in-place (``starts_at`` remains unique).
|
|
3. No active contract → no-op (returns 0, no DB writes).
|
|
4. Active contract with kind="manual" (not tibber) → no-op.
|
|
5. Token is empty string → no-op.
|
|
6. Token is whitespace-only → no-op.
|
|
7. Client errors propagate (not swallowed by the service).
|
|
|
|
All Tibber API calls are intercepted via monkeypatching ``fetch_price_range``
|
|
so no real network access is needed.
|
|
"""
|
|
|
|
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, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.integrations.tibber.client import PricePoint, TibberError
|
|
from app.models.energy import EnergyContract, EnergyContractVersion, TibberPrice
|
|
from app.services.tibber_prices import refresh_prices
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
|
|
|
|
@pytest.fixture()
|
|
def energy_db(tmp_path: Path):
|
|
"""Temporary SQLite DB upgraded to head (has all energy tables)."""
|
|
db_path = tmp_path / "tibber_prices_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
alembic_cfg = _make_app_alembic_config(db_url)
|
|
command.upgrade(alembic_cfg, "head")
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
session = Session(engine)
|
|
yield session
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_UTC = UTC
|
|
|
|
|
|
def _ts(hour: int, minute: int = 0) -> datetime:
|
|
"""Return a UTC datetime on 2026-06-23 at the given hour:minute."""
|
|
return datetime(2026, 6, 23, hour, minute, 0, tzinfo=_UTC)
|
|
|
|
|
|
def _make_price_points(count: int = 3) -> list[PricePoint]:
|
|
"""Build a list of *count* fake PricePoint objects 15 minutes apart."""
|
|
base = _ts(10, 0)
|
|
points = []
|
|
for i in range(count):
|
|
starts_at = base + timedelta(minutes=15 * i)
|
|
points.append(
|
|
PricePoint(
|
|
starts_at=starts_at,
|
|
total=0.20 + i * 0.01,
|
|
energy=0.16 + i * 0.01,
|
|
tax=0.04,
|
|
currency="EUR",
|
|
level="NORMAL",
|
|
resolution="QUARTER_HOURLY",
|
|
)
|
|
)
|
|
return points
|
|
|
|
|
|
class _FakeSettings:
|
|
"""Minimal settings stand-in with Tibber fields."""
|
|
|
|
def __init__(self, token: str = "valid-token", home_id: str = ""):
|
|
self.tibber_api_token = token
|
|
self.tibber_home_id = home_id
|
|
|
|
|
|
def _create_active_tibber_contract(session: Session) -> EnergyContract:
|
|
"""Insert and commit an active tibber contract + one open version."""
|
|
now = datetime.now(UTC)
|
|
contract = EnergyContract(
|
|
name="Tibber Test",
|
|
kind="tibber",
|
|
active=True,
|
|
currency="EUR",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(contract)
|
|
session.flush()
|
|
version = EnergyContractVersion(
|
|
contract_id=contract.id,
|
|
effective_from=now,
|
|
effective_to=None,
|
|
values={"energy": {"energy_tax": 0.1108, "sell_adjust": 0.0}, "standing": {}, "credits": {}},
|
|
created_at=now,
|
|
)
|
|
session.add(version)
|
|
session.commit()
|
|
return contract
|
|
|
|
|
|
def _create_active_manual_contract(session: Session) -> EnergyContract:
|
|
"""Insert and commit an active manual contract."""
|
|
now = datetime.now(UTC)
|
|
contract = EnergyContract(
|
|
name="Manual Test",
|
|
kind="manual",
|
|
active=True,
|
|
currency="EUR",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(contract)
|
|
session.flush()
|
|
version = EnergyContractVersion(
|
|
contract_id=contract.id,
|
|
effective_from=now,
|
|
effective_to=None,
|
|
values={},
|
|
created_at=now,
|
|
)
|
|
session.add(version)
|
|
session.commit()
|
|
return contract
|
|
|
|
|
|
def _count_tibber_price_rows(session: Session) -> int:
|
|
"""Return the current number of rows in tibber_price."""
|
|
return len(session.execute(select(TibberPrice)).scalars().all())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test: happy path (active tibber contract + valid token)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_refresh_prices_upserts_price_points(energy_db, monkeypatch):
|
|
"""Active tibber contract + token → price points are upserted and count returned."""
|
|
session = energy_db
|
|
_create_active_tibber_contract(session)
|
|
points = _make_price_points(3)
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.tibber_prices.fetch_price_range",
|
|
lambda token, home_id_or_none: points,
|
|
)
|
|
|
|
count = refresh_prices(session, _FakeSettings())
|
|
|
|
assert count == 3
|
|
rows = session.execute(select(TibberPrice).order_by(TibberPrice.starts_at)).scalars().all()
|
|
assert len(rows) == 3
|
|
|
|
for row, point in zip(rows, points):
|
|
# starts_at is stored UTC-aware; SQLite may return naive — normalise.
|
|
stored_starts_at = row.starts_at
|
|
if stored_starts_at.tzinfo is None:
|
|
stored_starts_at = stored_starts_at.replace(tzinfo=UTC)
|
|
assert stored_starts_at == point.starts_at
|
|
assert row.total == pytest.approx(point.total)
|
|
assert row.energy == pytest.approx(point.energy)
|
|
assert row.tax == pytest.approx(point.tax)
|
|
assert row.currency == point.currency
|
|
assert row.resolution == "QUARTER_HOURLY"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test: idempotency (running twice must not double-insert)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_refresh_prices_is_idempotent(energy_db, monkeypatch):
|
|
"""Running refresh_prices twice does not increase the row count."""
|
|
session = energy_db
|
|
_create_active_tibber_contract(session)
|
|
points = _make_price_points(4)
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def _fake_fetch(token, home_id_or_none):
|
|
call_count["n"] += 1
|
|
return points
|
|
|
|
monkeypatch.setattr("app.services.tibber_prices.fetch_price_range", _fake_fetch)
|
|
|
|
count1 = refresh_prices(session, _FakeSettings())
|
|
count2 = refresh_prices(session, _FakeSettings())
|
|
|
|
assert call_count["n"] == 2 # fetch was called twice
|
|
assert count1 == 4
|
|
assert count2 == 4 # same number of "upserted" (all updated in-place)
|
|
|
|
# DB must still have exactly 4 rows (not 8).
|
|
assert _count_tibber_price_rows(session) == 4
|
|
|
|
|
|
def test_refresh_prices_idempotent_updates_fields(energy_db, monkeypatch):
|
|
"""On second run, existing rows are updated (not duplicated)."""
|
|
session = energy_db
|
|
_create_active_tibber_contract(session)
|
|
points = _make_price_points(2)
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.tibber_prices.fetch_price_range",
|
|
lambda token, home_id_or_none: points,
|
|
)
|
|
|
|
refresh_prices(session, _FakeSettings())
|
|
|
|
# Modify the points to simulate Tibber updating a price.
|
|
updated_points = [
|
|
PricePoint(
|
|
starts_at=p.starts_at,
|
|
total=p.total + 0.10, # price changed
|
|
energy=p.energy + 0.08,
|
|
tax=p.tax + 0.02,
|
|
currency=p.currency,
|
|
level="EXPENSIVE",
|
|
resolution=p.resolution,
|
|
)
|
|
for p in points
|
|
]
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.tibber_prices.fetch_price_range",
|
|
lambda token, home_id_or_none: updated_points,
|
|
)
|
|
|
|
refresh_prices(session, _FakeSettings())
|
|
|
|
rows = session.execute(select(TibberPrice).order_by(TibberPrice.starts_at)).scalars().all()
|
|
assert len(rows) == 2 # still 2, not 4
|
|
|
|
for row, up in zip(rows, updated_points):
|
|
assert row.total == pytest.approx(up.total)
|
|
assert row.level == "EXPENSIVE"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test: no-op conditions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_refresh_prices_noop_no_active_contract(energy_db, monkeypatch):
|
|
"""No active contract at all → no-op (returns 0, no DB writes)."""
|
|
session = energy_db
|
|
fetch_called = {"called": False}
|
|
|
|
def _fake_fetch(token, home_id_or_none):
|
|
fetch_called["called"] = True
|
|
return _make_price_points(3)
|
|
|
|
monkeypatch.setattr("app.services.tibber_prices.fetch_price_range", _fake_fetch)
|
|
|
|
count = refresh_prices(session, _FakeSettings())
|
|
|
|
assert count == 0
|
|
assert not fetch_called["called"]
|
|
assert _count_tibber_price_rows(session) == 0
|
|
|
|
|
|
def test_refresh_prices_noop_manual_contract(energy_db, monkeypatch):
|
|
"""Active manual contract (not tibber) → no-op."""
|
|
session = energy_db
|
|
_create_active_manual_contract(session)
|
|
fetch_called = {"called": False}
|
|
|
|
def _fake_fetch(token, home_id_or_none):
|
|
fetch_called["called"] = True
|
|
return _make_price_points(3)
|
|
|
|
monkeypatch.setattr("app.services.tibber_prices.fetch_price_range", _fake_fetch)
|
|
|
|
count = refresh_prices(session, _FakeSettings())
|
|
|
|
assert count == 0
|
|
assert not fetch_called["called"]
|
|
assert _count_tibber_price_rows(session) == 0
|
|
|
|
|
|
def test_refresh_prices_noop_empty_token(energy_db, monkeypatch):
|
|
"""Empty token → no-op even if tibber contract is active."""
|
|
session = energy_db
|
|
_create_active_tibber_contract(session)
|
|
fetch_called = {"called": False}
|
|
|
|
def _fake_fetch(token, home_id_or_none):
|
|
fetch_called["called"] = True
|
|
return _make_price_points(3)
|
|
|
|
monkeypatch.setattr("app.services.tibber_prices.fetch_price_range", _fake_fetch)
|
|
|
|
count = refresh_prices(session, _FakeSettings(token=""))
|
|
|
|
assert count == 0
|
|
assert not fetch_called["called"]
|
|
assert _count_tibber_price_rows(session) == 0
|
|
|
|
|
|
def test_refresh_prices_noop_whitespace_token(energy_db, monkeypatch):
|
|
"""Whitespace-only token → no-op (treated as not configured)."""
|
|
session = energy_db
|
|
_create_active_tibber_contract(session)
|
|
fetch_called = {"called": False}
|
|
|
|
def _fake_fetch(token, home_id_or_none):
|
|
fetch_called["called"] = True
|
|
return _make_price_points(2)
|
|
|
|
monkeypatch.setattr("app.services.tibber_prices.fetch_price_range", _fake_fetch)
|
|
|
|
count = refresh_prices(session, _FakeSettings(token=" "))
|
|
|
|
assert count == 0
|
|
assert not fetch_called["called"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test: inactive tibber contract
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_refresh_prices_noop_inactive_tibber_contract(energy_db, monkeypatch):
|
|
"""Inactive tibber contract → no-op (active flag is False)."""
|
|
session = energy_db
|
|
|
|
# Create a tibber contract but do NOT activate it.
|
|
now = datetime.now(UTC)
|
|
contract = EnergyContract(
|
|
name="Inactive Tibber",
|
|
kind="tibber",
|
|
active=False, # <-- inactive
|
|
currency="EUR",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(contract)
|
|
session.commit()
|
|
|
|
fetch_called = {"called": False}
|
|
|
|
def _fake_fetch(token, home_id_or_none):
|
|
fetch_called["called"] = True
|
|
return _make_price_points(2)
|
|
|
|
monkeypatch.setattr("app.services.tibber_prices.fetch_price_range", _fake_fetch)
|
|
|
|
count = refresh_prices(session, _FakeSettings())
|
|
|
|
assert count == 0
|
|
assert not fetch_called["called"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test: client errors propagate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_refresh_prices_propagates_client_error(energy_db, monkeypatch):
|
|
"""TibberError from the client is NOT swallowed by the service."""
|
|
session = energy_db
|
|
_create_active_tibber_contract(session)
|
|
|
|
def _failing_fetch(token, home_id_or_none):
|
|
raise TibberError("simulated network failure")
|
|
|
|
monkeypatch.setattr("app.services.tibber_prices.fetch_price_range", _failing_fetch)
|
|
|
|
with pytest.raises(TibberError):
|
|
refresh_prices(session, _FakeSettings())
|
|
|
|
# No rows should have been written.
|
|
assert _count_tibber_price_rows(session) == 0
|