adminandClaude Opus 5.5
6862dea52a
- test_price_secret.py: the fixed-tariff note on the real /config/validate route, EN and NL, and none for a spaces-only profile (web.py lang wiring). - test_price_service.py: the start line is INFO, not WARNING; a dismiss that raises does not stop the cycle. - test_maint13/test_setup_only/test_watchdog: ignore only the dismiss of the old card, so a create with that id is still counted. - DECISIONS rule 7 quotes the settings note as it is; messages.py comment. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
1747 lines
72 KiB
Python
1747 lines
72 KiB
Python
"""PRICE-01 - the owner's price server as a price source, and its fallbacks.
|
|
`python3 test_price_service.py`
|
|
|
|
WHAT THIS FILE DEFENDS, in the order the card names it:
|
|
|
|
1. `price_service` fetches from ONE fixed HTTPS host with a bearer token, and
|
|
the price it returns is wholesale energy, so markup and taxes are added
|
|
once, exactly as for `generic_dayahead`.
|
|
2. The chain own server -> energy-charts -> stored prices: every way the
|
|
server can fail steps down, every step down and back up is logged once,
|
|
and a day with every source down still plans from what was kept.
|
|
3. No price for now: one status line and one notification, once, gone when a
|
|
price returns. English and Dutch.
|
|
4. The token is in no log, no status row, no notification, no error text.
|
|
5. The maintenance charge gate keeps the Home Assistant rule (current price
|
|
at or below today's average), read from the provider. The same day is fed
|
|
both ways and every 15-minute period must give the same answer.
|
|
6. Modes plan on these prices.
|
|
|
|
THE SERVER IS A MOCK. This session holds no token, so it cannot ask the real
|
|
one for a price. `Server` below answers the layout the design's API section
|
|
gives (`unix_seconds`, `price`, `unit`, `zone`, `covers`, `source`,
|
|
`fetched_at`), over a transport with the same contract as `price.https_get`.
|
|
Section 11 drives the REAL `https_get` against a local aiohttp server, for the
|
|
parts a mock transport cannot show: the header on the wire, the status words,
|
|
the timeout, a refused connection.
|
|
|
|
THE PRICES ARE NOT A FLAT LINE. They are deterministic and uneven, with a
|
|
negative midday, so "at or below the average" separates quarters on both sides
|
|
and a rule that always says yes or always says no cannot pass.
|
|
|
|
⚠️ MUTATING A MODULE THIS FILE GRADES? Purge `__pycache__` first, or use
|
|
`tools/mutation_run.sh`. See MUTATION.md.
|
|
"""
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
from named_absence import nth
|
|
from app import main as _main
|
|
from app import messages, price as P, price_chain as C
|
|
from app.price import (
|
|
PROVIDERS, PriceProduct, SourceUnavailable, brussels, brussels_offset,
|
|
load_prices, scrub, status_words)
|
|
from app.price_history import PriceHistory
|
|
from app.static_tariff import resolve_tariff
|
|
from app.validate import validate_startup_config
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def run(coro):
|
|
return asyncio.run(coro)
|
|
|
|
|
|
UTC = timezone.utc
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
TOKEN = "tk-7Hq2-secret-9xZ4-PRICE01"
|
|
HOST = P.PriceService.HOST
|
|
EC_HOST = "api.energy-charts.info"
|
|
MARKUP, TAXES = 0.031, 0.057
|
|
_TMP = []
|
|
|
|
|
|
def tmpdir(prefix="price01-"):
|
|
d = tempfile.mkdtemp(prefix=prefix)
|
|
_TMP.append(d)
|
|
return d
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# The mock server, and the day it serves
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def local_midnight(d: date) -> datetime:
|
|
"""Brussels midnight of `d` as a UTC instant. Midnight is never inside the
|
|
02:00-03:00 change, so one of the two offsets is always right."""
|
|
for off in (P.CET, P.CEST):
|
|
t = datetime(d.year, d.month, d.day, tzinfo=off)
|
|
if brussels_offset(t) == off:
|
|
return t.astimezone(UTC)
|
|
raise AssertionError(d)
|
|
|
|
|
|
def day_starts(d: date) -> list:
|
|
"""The UTC instants of every 15-minute period of Brussels day `d`: 96, or
|
|
92 and 100 on the two clock-change days."""
|
|
a, b = local_midnight(d), local_midnight(d + timedelta(days=1))
|
|
out, t = [], a
|
|
while t < b:
|
|
out.append(t)
|
|
t += timedelta(seconds=900)
|
|
return out
|
|
|
|
|
|
def ws_price(d: date, i: int) -> float:
|
|
"""EUR/MWh for period `i` of day `d`. Uneven, two decimals, negative at
|
|
midday on some days. Not a function of anything the code under test sees."""
|
|
base = 40 + 0.9 * ((d.toordinal() * 131 + i * i * 7) % 120)
|
|
if 44 <= i <= 58 and d.toordinal() % 2 == 0:
|
|
base -= 80
|
|
return round(base, 2)
|
|
|
|
|
|
def day_rows(d: date) -> list:
|
|
return [(int(t.timestamp()), ws_price(d, i))
|
|
for i, t in enumerate(day_starts(d))]
|
|
|
|
|
|
class Clock:
|
|
def __init__(self, now):
|
|
self.now = now
|
|
|
|
def __call__(self):
|
|
return self.now
|
|
|
|
|
|
class Server:
|
|
"""The price server, as the design's API section describes it.
|
|
|
|
`published(now)` is the days it holds: today always, tomorrow from 13:00
|
|
Brussels time if `tomorrow` is on. `mode` picks what it does to a request;
|
|
every mode that is not "up" is a way a real server or the road to it can
|
|
fail.
|
|
"""
|
|
|
|
def __init__(self, clock, token=TOKEN):
|
|
self.clock, self.token = clock, token
|
|
self.mode = "up"
|
|
self.tomorrow = True
|
|
self.requests = []
|
|
|
|
def days(self):
|
|
today = brussels(self.clock()).date()
|
|
return [today] + ([today + timedelta(days=1)] if self.tomorrow else [])
|
|
|
|
def doc(self, days=None):
|
|
days = self.days() if days is None else days
|
|
rows = [r for d in days for r in day_rows(d)]
|
|
return {"unix_seconds": [r[0] for r in rows],
|
|
"price": [r[1] for r in rows],
|
|
"unit": "EUR / MWh", "zone": "BE",
|
|
"covers": [d.isoformat() for d in days],
|
|
"source": {d.isoformat(): "entsoe" for d in days},
|
|
"fetched_at": self.clock().strftime("%Y-%m-%dT%H:%M:%SZ")}
|
|
|
|
def answer(self, headers):
|
|
self.requests.append(dict(headers or {}))
|
|
m = self.mode
|
|
if (headers or {}).get("Authorization") != "Bearer " + self.token \
|
|
and m == "up":
|
|
raise SourceUnavailable(status_words(401))
|
|
if m == "refused":
|
|
# A hostile library text: it carries the header it was sent.
|
|
raise ConnectionRefusedError(
|
|
f"Cannot connect to host {HOST}:443 ssl:default "
|
|
f"[Connect call failed] Authorization: Bearer {self.token}")
|
|
if m == "timeout":
|
|
raise asyncio.TimeoutError()
|
|
if m in ("401", "403", "429", "500", "503"):
|
|
raise SourceUnavailable(status_words(int(m)))
|
|
if m == "bad_json":
|
|
return b"<html>Bad gateway</html>"
|
|
d = self.doc()
|
|
if m == "no_arrays":
|
|
del d["price"]
|
|
elif m == "unequal":
|
|
d["price"] = d["price"][:-3]
|
|
elif m == "wrong_unit":
|
|
d["unit"] = "EUR / kWh"
|
|
elif m == "bad_covers":
|
|
d["covers"] = "2026-10-14"
|
|
elif m == "empty":
|
|
d["unix_seconds"], d["price"] = [], []
|
|
elif m == "stale":
|
|
d = self.doc([brussels(self.clock()).date() - timedelta(days=3)])
|
|
elif m == "uncovered":
|
|
d["covers"] = []
|
|
elif m == "partial_tomorrow":
|
|
# Tomorrow's rows are there but the server does not call the day
|
|
# complete, so it must not be used.
|
|
d["covers"] = d["covers"][:1]
|
|
return json.dumps(d).encode()
|
|
|
|
|
|
class EnergyCharts:
|
|
"""api.energy-charts.info: the columnar layout, for the dates asked for."""
|
|
|
|
def __init__(self, clock):
|
|
self.clock = clock
|
|
self.mode = "up"
|
|
self.tomorrow = True
|
|
self.requests = []
|
|
|
|
def answer(self, url):
|
|
self.requests.append(url)
|
|
if self.mode == "down":
|
|
raise SourceUnavailable("energy-charts is not answering")
|
|
if self.mode == "503":
|
|
raise ConnectionError("503 Service Unavailable")
|
|
q = dict(p.split("=") for p in url.split("?", 1)[1].split("&"))
|
|
a, b = date.fromisoformat(q["start"]), date.fromisoformat(q["end"])
|
|
today = brussels(self.clock()).date()
|
|
out, d = [], a
|
|
while d <= b:
|
|
if d <= today or (d == today + timedelta(days=1)
|
|
and self.tomorrow):
|
|
out += day_rows(d)
|
|
d += timedelta(days=1)
|
|
return json.dumps({"unix_seconds": [r[0] for r in out],
|
|
"price": [r[1] for r in out],
|
|
"unit": "EUR / MWh", "license_info": "x",
|
|
"deprecated": False}).encode()
|
|
|
|
|
|
class Net:
|
|
"""The transport. Same contract as `price.https_get`."""
|
|
|
|
def __init__(self, clock):
|
|
self.clock = clock
|
|
self.server = Server(clock)
|
|
self.ec = EnergyCharts(clock)
|
|
self.calls = []
|
|
|
|
async def __call__(self, url, timeout_s, headers=None):
|
|
self.calls.append((url, dict(headers or {})))
|
|
host = url.split("/")[2]
|
|
if host == HOST:
|
|
return self.server.answer(headers)
|
|
if host == EC_HOST:
|
|
return self.ec.answer(url)
|
|
raise AssertionError(f"a request left for an undeclared host: {url}")
|
|
|
|
def down(self):
|
|
self.server.mode, self.ec.mode = "refused", "down"
|
|
|
|
def up(self):
|
|
self.server.mode, self.ec.mode = "up", "up"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# A place to read logs, notifications and statuses from
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class Tap(logging.Handler):
|
|
def __init__(self):
|
|
super().__init__(logging.DEBUG)
|
|
self.records = []
|
|
|
|
def emit(self, record):
|
|
self.records.append(record)
|
|
|
|
def lines(self, minimum=logging.INFO):
|
|
return [r.getMessage() for r in self.records
|
|
if r.levelno >= minimum]
|
|
|
|
def text(self):
|
|
return "\n".join(f"{r.levelname} {r.name} {r.getMessage()}"
|
|
for r in self.records)
|
|
|
|
|
|
TAP = Tap()
|
|
_root = logging.getLogger()
|
|
_root.addHandler(TAP)
|
|
_root.setLevel(logging.DEBUG)
|
|
|
|
|
|
def mark():
|
|
return len(TAP.records)
|
|
|
|
|
|
def since_mark(n, minimum=logging.INFO):
|
|
return [r.getMessage() for r in TAP.records[n:] if r.levelno >= minimum]
|
|
|
|
|
|
def opts(**over):
|
|
o = {"price_provider": "price_service", "price_service_token": TOKEN,
|
|
"price_markup_eur_kwh": MARKUP, "price_taxes_eur_kwh": TAXES,
|
|
"price_timeout_s": 5}
|
|
o.update(over)
|
|
return o
|
|
|
|
|
|
NOW = datetime(2026, 10, 14, 15, 7, tzinfo=UTC) # 17:07 Brussels (CEST)
|
|
|
|
|
|
def make_ctx(now=NOW, **over):
|
|
clock = Clock(now)
|
|
net = Net(clock)
|
|
hist = PriceHistory(os.path.join(tmpdir(), "price_history.json"))
|
|
return type("Ctx", (), {"clock": clock, "net": net, "hist": hist,
|
|
"watch": C.PriceWatch(), "o": opts(**over)})()
|
|
|
|
|
|
def refresh(ctx, now=None):
|
|
"""One background refresh, exactly as `run_tariff_refresh` calls it."""
|
|
if now is not None:
|
|
ctx.clock.now = now
|
|
now = ctx.clock.now
|
|
return run(resolve_tariff(ctx.o, now, now + timedelta(hours=48),
|
|
transport=ctx.net, now=now, history=ctx.hist,
|
|
watch=ctx.watch))
|
|
|
|
|
|
def covers_now(plan, now):
|
|
return plan.published is not None and plan.published.covers(now)
|
|
|
|
|
|
def all_in(ws, markup=MARKUP, taxes=TAXES):
|
|
return round(ws / 1000 + markup + taxes, 6)
|
|
|
|
|
|
# =========================================================================== #
|
|
print("0. the mock is the server the design describes, and the day is uneven")
|
|
_ctx = make_ctx()
|
|
_d = _ctx.net.server.doc()
|
|
check("the answer has the design's keys",
|
|
{"unix_seconds", "price", "unit", "zone", "covers", "source",
|
|
"fetched_at"} <= set(_d))
|
|
check("...with one price per start, 900 s apart, today and tomorrow",
|
|
len(_d["unix_seconds"]) == len(_d["price"]) == 192
|
|
and {b - a for a, b in zip(_d["unix_seconds"],
|
|
_d["unix_seconds"][1:])} == {900}
|
|
and _d["covers"] == ["2026-10-14", "2026-10-15"])
|
|
check("the fixture is uneven and has a negative price, so 'at or below the "
|
|
"average' separates periods on both sides",
|
|
min(_d["price"]) < 0 < max(_d["price"])
|
|
and 20 < sum(1 for p in _d["price"][:96]
|
|
if p <= sum(_d["price"][:96]) / 96) < 76)
|
|
check("clock-change days have 92 and 100 periods, not 96",
|
|
(len(day_starts(date(2026, 3, 29))), len(day_starts(date(2026, 10, 25))))
|
|
== (92, 100))
|
|
|
|
# =========================================================================== #
|
|
print("\n1. the provider: one fixed host, a bearer token, wholesale energy")
|
|
_ctx = make_ctx()
|
|
_n = mark()
|
|
_plan = refresh(_ctx)
|
|
check("the first step answers: the plan came from the price service",
|
|
_plan.step == C.STEP_OWN and _plan.published.provider == "price_service"
|
|
and _plan.from_feed)
|
|
_url, _hdr = _ctx.net.calls[0]
|
|
check("the request went to the one fixed HTTPS URL, and only to it",
|
|
_url == "https://prices.kammenstraatha.duckdns.org/v1/dayahead/BE"
|
|
and [u for u, _h in _ctx.net.calls if HOST not in u] == [])
|
|
check("...with the token as a bearer header", _hdr == {"Authorization":
|
|
"Bearer " + TOKEN})
|
|
check("...and the token is not in the URL", TOKEN not in _url)
|
|
check("the host is declared as the plugin's one egress host, so the framework "
|
|
"refuses any other",
|
|
P.PriceService.capabilities().egress_hosts == (HOST,))
|
|
def _refused(fn):
|
|
try:
|
|
fn()
|
|
except SourceUnavailable:
|
|
return True
|
|
return False
|
|
|
|
|
|
check("...an http:// URL on the right host is refused",
|
|
_refused(lambda: P.check_egress("http://" + HOST + "/v1/dayahead/BE",
|
|
P.PriceService.capabilities())))
|
|
check("...and so is another host over https",
|
|
_refused(lambda: P.check_egress("https://example.org/v1/dayahead/BE",
|
|
P.PriceService.capabilities())))
|
|
_ctx2 = make_ctx(price_feed_url="http://evil.example/x")
|
|
refresh(_ctx2)
|
|
check("the operator's feed override is ignored by this provider: the request "
|
|
"still went to the fixed URL",
|
|
_ctx2.net.calls[0][0] == P.PriceService.URL)
|
|
|
|
# markup and taxes are added once, as for generic_dayahead
|
|
_ps = _plan.published
|
|
_gd = run(load_prices(opts(price_provider="generic_dayahead"), NOW,
|
|
NOW + timedelta(hours=48), transport=_ctx.net))
|
|
_ps_prices = [p.eur_per_kwh for p in _ps.points]
|
|
_want = [all_in(r[1]) for d in _ctx.net.server.days() for r in day_rows(d)]
|
|
check("the price is the wholesale figure plus markup and taxes, once: every "
|
|
"point equals ws/1000 + markup + taxes",
|
|
_ps_prices == _want)
|
|
check("...and a price_service series equals the generic_dayahead one for the "
|
|
"same days, point for point (start and price)",
|
|
[(p.start, p.eur_per_kwh) for p in _ps.points]
|
|
== [(p.start, p.eur_per_kwh) for p in _gd.points])
|
|
check("the provider declares energy only, so the framework adds markup and "
|
|
"taxes, and the product is day-ahead",
|
|
P.PriceService.capabilities().adds
|
|
== P.GenericDayAhead.capabilities().adds
|
|
and P.PriceService.capabilities().product is PriceProduct.DAY_AHEAD)
|
|
check("the plugin cannot reach a framework key (register() would refuse it)",
|
|
P.code_leaks(P.PriceService) == [])
|
|
|
|
_ctx3 = make_ctx(price_service_token="")
|
|
_p3 = refresh(_ctx3)
|
|
check("an empty token sends NO request to the server and falls to energy-charts",
|
|
not any(HOST in u for u, _h in _ctx3.net.calls)
|
|
and _p3.step == C.STEP_PUBLIC)
|
|
check("a token with a line break in it is never sent",
|
|
not any(HOST in u for u, _h in
|
|
(lambda c: (refresh(c), c.net.calls)[1])(
|
|
make_ctx(price_service_token="abc\r\nX-Evil: 1"))))
|
|
|
|
# =========================================================================== #
|
|
print("\n2. every way the server can fail steps down to energy-charts")
|
|
_FAILURES = {
|
|
"refused": "does not answer",
|
|
"timeout": "does not answer",
|
|
"401": "refused the token",
|
|
"403": "refused the token",
|
|
"429": "does not answer",
|
|
"500": "does not answer",
|
|
"503": "does not answer",
|
|
"bad_json": "does not answer",
|
|
"no_arrays": "does not answer",
|
|
"unequal": "does not answer",
|
|
"wrong_unit": "does not answer",
|
|
"bad_covers": "does not answer",
|
|
"empty": "does not answer",
|
|
"stale": "does not answer",
|
|
"uncovered": "does not answer",
|
|
}
|
|
for _mode, _words in _FAILURES.items():
|
|
_c = make_ctx()
|
|
_c.net.server.mode = _mode
|
|
_n = mark()
|
|
_p = refresh(_c)
|
|
_warn = since_mark(_n, logging.WARNING)
|
|
check(f"server {_mode}: prices come from energy-charts, firm, covering now",
|
|
_p.step == C.STEP_PUBLIC and _p.from_feed
|
|
and _p.published.provider == "generic_dayahead"
|
|
and covers_now(_p, _c.clock.now))
|
|
check(f"...one WARNING says why, plainly ({_warn})",
|
|
len(_warn) == 1 and _words in _warn[0]
|
|
and "energy-charts" in _warn[0])
|
|
check(f"...and the energy-charts request never carried the token",
|
|
all(TOKEN not in json.dumps(h)
|
|
for u, h in _c.net.calls if EC_HOST in u))
|
|
_n2 = mark()
|
|
refresh(_c, _c.clock.now + timedelta(minutes=30))
|
|
check(f"...and the next refresh of the same failure logs nothing again",
|
|
since_mark(_n2, logging.INFO) == [])
|
|
|
|
_c = make_ctx(price_feed_url="https://api.energy-charts.info/price?bzn=BE")
|
|
_c.net.server.mode = "refused"
|
|
_p = refresh(_c)
|
|
_ec_urls = [u for u, _h in _c.net.calls if EC_HOST in u]
|
|
check("the operator's feed override is not used for the energy-charts step: it "
|
|
"is asked for its own feed with the date range, and answers",
|
|
_p.step == C.STEP_PUBLIC and len(_ec_urls) == 1
|
|
and "start=" in _ec_urls[0] and "end=" in _ec_urls[0])
|
|
|
|
# partial day: tomorrow not published yet (before ~13:00 the server holds today)
|
|
_c = make_ctx(now=datetime(2026, 10, 14, 6, 0, tzinfo=UTC))
|
|
_c.net.server.tomorrow = False
|
|
_n = mark()
|
|
_p = refresh(_c)
|
|
check("tomorrow missing: the server is still the source, no warning",
|
|
_p.step == C.STEP_OWN and since_mark(_n, logging.WARNING) == [])
|
|
_days = {brussels(x.start).date() for x in _p.published.points}
|
|
check("...the series holds today only, and the plan carries the rest from "
|
|
"yesterday's saved day as estimated, not firm",
|
|
_days == {date(2026, 10, 14)})
|
|
_c = make_ctx()
|
|
_c.net.server.mode = "partial_tomorrow"
|
|
_p = refresh(_c)
|
|
check("a day the server does not list as complete is not used, even if its "
|
|
"rows are there",
|
|
_p.step == C.STEP_OWN
|
|
and {brussels(x.start).date() for x in _p.published.points}
|
|
== {date(2026, 10, 14)})
|
|
|
|
# =========================================================================== #
|
|
print("\n3. both sources down: the prices the history kept")
|
|
_c = make_ctx()
|
|
refresh(_c) # a fetch that worked
|
|
_snap = open(_c.hist.path, "rb").read()
|
|
_st0 = os.stat(_c.hist.path)
|
|
_c.net.down()
|
|
_n = mark()
|
|
_p = refresh(_c, NOW + timedelta(minutes=30))
|
|
_warn = since_mark(_n, logging.WARNING)
|
|
check("the plan comes from the saved prices, is firm, and covers now",
|
|
_p.step == C.STEP_STORED and _p.from_feed
|
|
and _p.published.provider == "price_history"
|
|
and covers_now(_p, _c.clock.now))
|
|
check(f"...and one WARNING says so plainly ({_warn})",
|
|
_warn == ["No price source answers, so the add-on uses the prices it "
|
|
"saved."])
|
|
check("...with the reasons of each step on a DEBUG line, token-free",
|
|
any("Price sources said" in r.getMessage() and r.levelno == logging.DEBUG
|
|
for r in TAP.records[_n:])
|
|
and all(TOKEN not in r.getMessage() for r in TAP.records[_n:]))
|
|
_st1 = os.stat(_c.hist.path)
|
|
check("...and the saved file was not written again from a read of itself (the "
|
|
"file is replaced on every save, so the same inode and time mean no save)",
|
|
open(_c.hist.path, "rb").read() == _snap
|
|
and (_st1.st_ino, _st1.st_mtime_ns) == (_st0.st_ino, _st0.st_mtime_ns))
|
|
_stored_prices = [p.eur_per_kwh for p in _p.published.points]
|
|
check("the saved prices are the ones fetched, not carried guesses",
|
|
all(not p.provisional for p in _p.published.points)
|
|
and _stored_prices == _want)
|
|
_n = mark()
|
|
refresh(_c, NOW + timedelta(minutes=60))
|
|
check("a second refresh in the same state logs nothing",
|
|
since_mark(_n, logging.INFO) == [])
|
|
|
|
# nothing stored, everything down: no price
|
|
_c = make_ctx()
|
|
_c.net.down()
|
|
_n = mark()
|
|
_p = refresh(_c)
|
|
check("with nothing saved and every source down there is no plan from a feed",
|
|
_p.step == C.STEP_NONE and not _p.from_feed and _p.published is None)
|
|
check("...the chain says so in one WARNING",
|
|
[r.getMessage() for r in TAP.records[_n:]
|
|
if r.name == "goodwe.price.chain" and r.levelno >= logging.WARNING]
|
|
== ["No price source answers, and no saved price covers now."])
|
|
check("...and what each step said is kept for DEBUG, without the token",
|
|
len(_p.reasons) == 3 and all(TOKEN not in r for r in _p.reasons)
|
|
and "price_service" in _p.reasons[0]
|
|
and "generic_dayahead" in _p.reasons[1]
|
|
and "price_history" in _p.reasons[2])
|
|
|
|
# =========================================================================== #
|
|
print("\n4. back up the chain, one line per step")
|
|
_c = make_ctx()
|
|
_c.net.down()
|
|
refresh(_c)
|
|
_n = mark()
|
|
_c.net.ec.mode = "up"
|
|
refresh(_c, NOW + timedelta(minutes=30))
|
|
check("energy-charts back: INFO says so once",
|
|
since_mark(_n, logging.INFO)
|
|
== ["Prices come from energy-charts again."])
|
|
_n = mark()
|
|
_c.net.server.mode = "up"
|
|
refresh(_c, NOW + timedelta(minutes=60))
|
|
check("the server back: INFO says so once",
|
|
since_mark(_n, logging.INFO) == ["The price service answers again."])
|
|
_n = mark()
|
|
refresh(_c, NOW + timedelta(minutes=90))
|
|
check("and nothing more while it stays up", since_mark(_n, logging.INFO) == [])
|
|
_c = make_ctx()
|
|
refresh(_c)
|
|
_c.net.down()
|
|
refresh(_c, NOW + timedelta(minutes=30))
|
|
_n = mark()
|
|
_c.net.server.mode = "up"
|
|
refresh(_c, NOW + timedelta(minutes=60))
|
|
check("stored -> server directly is one 'answers again' line",
|
|
since_mark(_n, logging.INFO) == ["The price service answers again."])
|
|
_c = make_ctx()
|
|
_n = mark()
|
|
_c.net.server.mode = "401"
|
|
refresh(_c)
|
|
check("a start-up with the server refusing is one WARNING, not an INFO then a "
|
|
"WARNING",
|
|
since_mark(_n, logging.INFO)
|
|
== ["The price service refused the token, so prices come from "
|
|
"energy-charts."])
|
|
|
|
_w = C.PriceWatch()
|
|
_w.note(C.STEP_NONE)
|
|
_n = mark()
|
|
_w.note(C.STEP_STORED)
|
|
check("coming back from no price to the saved prices is one INFO, not a "
|
|
"second alarm",
|
|
since_mark(_n, logging.INFO)
|
|
== ["The add-on uses the prices it saved again."])
|
|
|
|
# =========================================================================== #
|
|
print("\n5. a full day with every source down")
|
|
# A good fetch at 13:30 Brussels with tomorrow in it; then nothing answers for
|
|
# 24 hours. Every 30-minute refresh must still produce a firm plan that covers
|
|
# now, until the saved day itself runs out at midnight after tomorrow.
|
|
_start = datetime(2026, 10, 14, 11, 30, tzinfo=UTC) # 13:30 CEST
|
|
_c = make_ctx(now=_start)
|
|
refresh(_c)
|
|
_c.net.down()
|
|
_bad = []
|
|
_t = _start
|
|
_gate_nones = 0
|
|
while _t < _start + timedelta(hours=24):
|
|
_t += timedelta(minutes=30)
|
|
_p = refresh(_c, _t)
|
|
if not (_p.from_feed and _p.step == C.STEP_STORED
|
|
and covers_now(_p, _t)):
|
|
_bad.append(_t.isoformat())
|
|
if not _p.firm_points():
|
|
_bad.append("no firm point at " + _t.isoformat())
|
|
check(f"48 refreshes over 24 h, all from the saved day, all firm, all covering "
|
|
f"now ({_bad[:3]})", not _bad)
|
|
_g = C.gate_for(_p.published, _t)
|
|
check("...and the charge gate still gives an answer 24 h later",
|
|
_g.covered and _g.words.startswith("Prices from saved prices"))
|
|
_end = local = local_midnight(date(2026, 10, 16))
|
|
_p = refresh(_c, _end - timedelta(minutes=15))
|
|
check("the last saved period still covers", covers_now(_p, _end
|
|
- timedelta(minutes=15)))
|
|
_p = refresh(_c, _end)
|
|
check("and at the first moment after the saved days there is no price, and "
|
|
"the series says since when",
|
|
_p.step == C.STEP_NONE and C.gate_for(
|
|
_c.hist.stored(_end).match(series=lambda r: r, no_data=None), _end
|
|
).since == _end)
|
|
|
|
# =========================================================================== #
|
|
print("\n6. no price for now: one status line, one notification, cleared")
|
|
OPTS_C = {"setpoint_entity": "number.sp", "max_w": 5000,
|
|
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
|
|
"meter_source": "off", "meter_entity": "sensor.grid",
|
|
"maintenance_enabled": True,
|
|
"maintenance_charge_w": 2000, "maintenance_discharge_w": 2000}
|
|
|
|
|
|
class FakeHass:
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.values = {"sensor.grid": 0.0, "sensor.soc": 55.0,
|
|
"sensor.batt": 0.0}
|
|
|
|
async def limits(self, entity):
|
|
return (-5000.0, 5000.0, 1.0)
|
|
|
|
async def number(self, entity, invert=False):
|
|
return self.values.get(entity)
|
|
|
|
async def set_number(self, entity, value):
|
|
return True
|
|
|
|
async def call(self, domain, service, data):
|
|
self.calls.append((domain, service, data))
|
|
return True
|
|
|
|
|
|
class FakeStore:
|
|
data = {}
|
|
|
|
def set(self, *a):
|
|
pass
|
|
|
|
def get_time(self, key):
|
|
return None
|
|
|
|
|
|
class FakePub:
|
|
enabled = False
|
|
|
|
def publish(self, *a, **k):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class FakeDT(datetime):
|
|
"""`datetime` with a clock the test moves."""
|
|
_clock = None
|
|
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
t = cls._clock.now
|
|
return t.astimezone(tz) if tz else t.replace(tzinfo=None)
|
|
|
|
|
|
def controller(clock, o, tmp=None):
|
|
tmp = tmp or tmpdir()
|
|
stop = asyncio.Event()
|
|
ctl = _main.Controller(dict(OPTS_C, **o), FakeHass(), FakeStore(),
|
|
FakePub(), stop=stop,
|
|
peak_db_path=os.path.join(tmp, "tariff.db"),
|
|
price_history_path=os.path.join(tmp, "ph.json"))
|
|
return ctl
|
|
|
|
|
|
_REAL_DT, _REAL_GET = _main.datetime, P.https_get
|
|
_real_resolve = _main.resolve_tariff
|
|
|
|
|
|
class Wired:
|
|
"""A controller whose clock and network are the test's."""
|
|
|
|
def __init__(self, now, **over):
|
|
self.clock = Clock(now)
|
|
FakeDT._clock = self.clock
|
|
self.net = Net(self.clock)
|
|
self.ctl = controller(self.clock, opts(**over))
|
|
self.refreshes = 0
|
|
|
|
def __enter__(self):
|
|
_main.datetime = FakeDT
|
|
P.https_get = self.net
|
|
W = self
|
|
|
|
async def counting(*a, **k):
|
|
r = await _real_resolve(*a, **k)
|
|
W.refreshes += 1
|
|
return r
|
|
_main.resolve_tariff = counting
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
_main.datetime, P.https_get = _REAL_DT, _REAL_GET
|
|
_main.resolve_tariff = _real_resolve
|
|
|
|
async def refresh(self, now=None):
|
|
if now is not None:
|
|
self.clock.now = now
|
|
c = self.ctl
|
|
n = self.refreshes
|
|
c.stopping = False
|
|
c.stop_event.clear()
|
|
task = asyncio.create_task(c.run_tariff_refresh())
|
|
for _ in range(500):
|
|
if self.refreshes > n:
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
await asyncio.sleep(0.02)
|
|
c.stopping = True
|
|
c.stop_event.set()
|
|
await asyncio.wait_for(task, timeout=5)
|
|
|
|
async def tick(self, now=None):
|
|
"""One control tick as `run_control` makes it: read inputs, then the
|
|
cycle, which is where the price-gap check is called from."""
|
|
if now is not None:
|
|
self.clock.now = now
|
|
self.ctl.auto = True
|
|
await self.ctl.read_inputs()
|
|
await self.ctl.cycle()
|
|
|
|
def all_notes(self):
|
|
return [(s, d) for dom, s, d in self.ctl.hass.calls
|
|
if dom == "persistent_notification"]
|
|
|
|
def notes(self):
|
|
"""Without the dismiss of the old fixed-tariff card that every start
|
|
makes (section 14 looks at it through `all_notes`)."""
|
|
return [x for x in self.all_notes()
|
|
if x != ("dismiss", {"notification_id": "goodwe_fixed_tariff_unused"})]
|
|
|
|
|
|
async def _scenario_gap():
|
|
out = {}
|
|
with Wired(NOW) as W:
|
|
c = W.ctl
|
|
out["rows_before"] = dict(c.status_payload()["rows"])
|
|
await W.tick()
|
|
out["before_first_refresh"] = (c.price_gap, list(W.notes()))
|
|
W.net.down()
|
|
await W.refresh()
|
|
for k in range(6): # six ticks inside the gap
|
|
await W.tick(NOW + timedelta(seconds=30 * k))
|
|
out["gap"] = (c.price_gap, c.price_gap_since, list(W.notes()))
|
|
out["rows_gap"] = dict(c.status_payload()["rows"])
|
|
out["payload_gap"] = json.dumps(c.status_payload(), default=str)
|
|
out["cheap_gap"] = (c.cheap, c._price_gated,
|
|
c._charge_gate_open(c.inverters[0]))
|
|
c.inverters[0].batt = -2000.0 # a pack really taking charge
|
|
out["open_while_charging"] = c._charge_gate_open(c.inverters[0])
|
|
c.inverters[0].batt = None
|
|
# the prices come back
|
|
W.net.up()
|
|
await W.refresh(NOW + timedelta(minutes=30))
|
|
for k in range(4):
|
|
await W.tick(NOW + timedelta(minutes=30, seconds=30 * k))
|
|
out["back"] = (c.price_gap, list(W.notes()))
|
|
out["rows_back"] = dict(c.status_payload()["rows"])
|
|
out["payload_back"] = json.dumps(c.status_payload(), default=str)
|
|
out["words_back"] = c.tariff_words
|
|
return out
|
|
|
|
|
|
_S = run(_scenario_gap())
|
|
check("before the first refresh has finished there is no gap and no "
|
|
"notification, so a start-up is not an alarm",
|
|
_S["before_first_refresh"] == (False, []) and
|
|
"Prices" not in _S["rows_before"])
|
|
_gap, _since, _notes = _S["gap"]
|
|
check("every source down with nothing saved: a gap, since nothing ever "
|
|
"published", _gap is True and _since is None)
|
|
check("...exactly ONE notification over six ticks, with the id the docs name",
|
|
len(_notes) == 1 and _notes[0][0] == "create"
|
|
and _notes[0][1]["notification_id"] == "goodwe_no_prices")
|
|
check("...in English and Dutch, plain and short",
|
|
"No prices have come in yet" in _notes[0][1]["message"]
|
|
and "Prijzen ontbreken. Er zijn nog geen prijzen binnen"
|
|
in _notes[0][1]["message"]
|
|
and _notes[0][1]["title"] == "No prices")
|
|
check("...the status row says it once, in the card's words",
|
|
_S["rows_gap"].get("Prices") == "No prices yet, maintenance waits.")
|
|
check("...Cheap window is no, and the charge gate is shut",
|
|
_S["cheap_gap"] == (False, True, False))
|
|
check("...but a pack that is really taking charge is not stopped by it",
|
|
_S["open_while_charging"] is True)
|
|
_gap2, _notes2 = _S["back"]
|
|
check("a price returns: the gap ends, the same notification is taken down "
|
|
"once, and no second one was raised",
|
|
_gap2 is False and len(_notes2) == 2 and _notes2[1][0] == "dismiss"
|
|
and _notes2[1][1] == {"notification_id": "goodwe_no_prices"})
|
|
check("...and the row goes with it", "Prices" not in _S["rows_back"])
|
|
check("...and the Tariff row names the source and the rule",
|
|
_S["words_back"].startswith("Prices from the price service,"))
|
|
|
|
# the refresh task stops: after 90 minutes the plan is retired and the gate
|
|
# shuts, with the moment it was retired as 'since'
|
|
async def _scenario_dead_task():
|
|
start = datetime(2026, 10, 14, 11, 30, tzinfo=UTC)
|
|
with Wired(start) as W:
|
|
c = W.ctl
|
|
await W.refresh()
|
|
await W.tick(start + timedelta(minutes=80))
|
|
before = (c.price_gap, c.cheap is not None, len(W.notes()))
|
|
await W.tick(start + timedelta(minutes=95))
|
|
await W.tick(start + timedelta(minutes=96))
|
|
mid = (c.price_gap, c.price_gap_since, list(W.notes()),
|
|
c._charge_gate_open(c.inverters[0]))
|
|
await W.refresh(start + timedelta(minutes=100))
|
|
await W.tick(start + timedelta(minutes=101))
|
|
return before, mid, (c.price_gap, list(W.notes()))
|
|
_b, _m, _a = run(_scenario_dead_task())
|
|
check("a plan 80 minutes old is still current: no gap, no notification",
|
|
_b == (False, True, 0))
|
|
check("at 95 minutes with no refresh the plan is retired: a gap, since the "
|
|
"moment it was retired (90 minutes after the fetch), one notification, "
|
|
"gate shut",
|
|
_m[0] is True and _m[1] == datetime(2026, 10, 14, 11, 30, tzinfo=UTC)
|
|
+ timedelta(seconds=_main.TARIFF_MAX_AGE_S)
|
|
and len(_m[2]) == 1 and _m[3] is False)
|
|
check("...and the next refresh that works takes it down again",
|
|
_a[0] is False and [n[0] for n in _a[1]] == ["create", "dismiss"])
|
|
|
|
# a gap that begins mid-day, with a price history that ends: 'since' is shown
|
|
async def _scenario_since():
|
|
start = datetime(2026, 10, 14, 11, 30, tzinfo=UTC)
|
|
with Wired(start) as W:
|
|
c = W.ctl
|
|
await W.refresh() # a good fetch
|
|
W.net.down()
|
|
end = local_midnight(date(2026, 10, 16)) # the saved days run out
|
|
for k in range(3):
|
|
await W.refresh(end - timedelta(minutes=30 * (2 - k)))
|
|
await W.tick(end + timedelta(minutes=5))
|
|
gap = (c.price_gap, c.price_gap_since)
|
|
row = dict(c.status_payload()["rows"])
|
|
notes = list(W.notes())
|
|
# the Home screen
|
|
from app import homeview
|
|
plan = homeview.build_plan(c, W.clock.now, False, True)
|
|
return gap, row, notes, plan["lines"], end
|
|
_gap, _row, _notes, _lines, _end = run(_scenario_since())
|
|
check("when the saved days run out, the gap begins at that moment",
|
|
_gap == (True, _end))
|
|
check("...the row reads 'No prices since 00:00, maintenance waits.' (Brussels "
|
|
"time, the card's words)",
|
|
_row.get("Prices") == "No prices since 00:00, maintenance waits.")
|
|
check("...the one notification carries the same moment, in English and Dutch",
|
|
len(_notes) == 1
|
|
and "No prices since 00:00, so the monthly battery cycle waits"
|
|
in _notes[0][1]["message"]
|
|
and "Prijzen ontbreken. Sinds 00:00 zijn er geen prijzen, dus de "
|
|
"maandelijkse batterijcyclus wacht" in _notes[0][1]["message"])
|
|
check("...and the Home screen says it once (English)",
|
|
_lines.count("No prices since 00:00, maintenance waits.") == 1)
|
|
from app import uitext as _uitext # noqa: E402
|
|
_nl = _uitext.words("nl")["home_words"]
|
|
check("...with the Dutch words 'Geen prijzen sinds {at}, onderhoud wacht.'",
|
|
_nl["no_prices_since"] == "Geen prijzen sinds {at}, onderhoud wacht."
|
|
and _nl["no_prices_yet"] == "Nog geen prijzen, onderhoud wacht."
|
|
and _nl["no_prices_since_day"]
|
|
== "Geen prijzen sinds {day} {month} om {at}, onderhoud wacht.")
|
|
_a = brussels(datetime(2026, 10, 13, 19, 0, tzinfo=UTC))
|
|
_b = brussels(datetime(2026, 10, 14, 9, 0, tzinfo=UTC))
|
|
check("an earlier day is spelled with its date, not only a time, in each "
|
|
"language's own words",
|
|
messages.since_text(_a, _b) == "13 Oct at 21:00"
|
|
and messages.since_text(_a, _b, "nl") == "13 okt om 21:00"
|
|
and messages.since_text(_b, _b) == "11:00")
|
|
from app import homeview as _hv # noqa: E402
|
|
_prev = _hv._LANG[0]
|
|
try:
|
|
_hv._LANG[0] = "nl"
|
|
_nl_line = _hv.line_no_prices(datetime(2026, 10, 13, 19, 0, tzinfo=UTC),
|
|
datetime(2026, 10, 14, 9, 0, tzinfo=UTC))
|
|
_hv._LANG[0] = "en"
|
|
_en_line = _hv.line_no_prices(datetime(2026, 10, 13, 19, 0, tzinfo=UTC),
|
|
datetime(2026, 10, 14, 9, 0, tzinfo=UTC))
|
|
finally:
|
|
_hv._LANG[0] = _prev
|
|
check("...and the Home line for an earlier day is in the page's language "
|
|
f"({_en_line!r}, {_nl_line!r})",
|
|
_en_line == "No prices since 13 October at 21:00, maintenance waits."
|
|
and _nl_line == "Geen prijzen sinds 13 oktober om 21:00, onderhoud wacht.")
|
|
|
|
# =========================================================================== #
|
|
print("\n7. the token is in no log, status row, notification or error text")
|
|
_everything = [TAP.text(),
|
|
json.dumps(_S, default=str), json.dumps(
|
|
[_row, _notes, _lines], default=str)]
|
|
check("across every scenario above the token never appeared in any log record "
|
|
"at any level",
|
|
all(TOKEN not in r.getMessage() and TOKEN not in str(r.args)
|
|
and TOKEN not in (r.exc_text or "") for r in TAP.records))
|
|
check("...nor in a status row, a notification or a home line",
|
|
all(TOKEN not in blob for blob in _everything[1:]))
|
|
check("...nor anywhere in the whole status payload, during the gap or after it",
|
|
len(_S["payload_gap"]) > 1000 and len(_S["payload_back"]) > 1000
|
|
and TOKEN not in _S["payload_gap"] and TOKEN not in _S["payload_back"])
|
|
check("...nor in the reason a failed read carries, even when the library's own "
|
|
"error text holds the header it was sent",
|
|
TOKEN not in (lambda c: (c.net.server.__setattr__("mode", "refused"),
|
|
c.net.ec.__setattr__("mode", "down"),
|
|
refresh(c).read.reason)[2])(make_ctx()))
|
|
_o = opts()
|
|
check("scrub() takes the value out of any text, and leaves other text alone",
|
|
scrub(f"x {TOKEN} y", _o) == "x *** y"
|
|
and scrub("no secret here", _o) == "no secret here"
|
|
and scrub(f"x {TOKEN}", {"price_service_token": ""}) == f"x {TOKEN}")
|
|
|
|
|
|
async def _hostile(url, timeout_s, headers=None):
|
|
# A transport that fails with the full request in its text, by any means.
|
|
raise RuntimeError(f"boom {headers} {url} {TOKEN}")
|
|
|
|
|
|
_r = run(load_prices(opts(), NOW, NOW + timedelta(hours=48),
|
|
transport=_hostile))
|
|
check("an exception from the transport that carries the token reaches no "
|
|
"reason text", _r.kind == "no_data" and TOKEN not in _r.reason)
|
|
|
|
|
|
def _fetch_error(transport):
|
|
"""What the PLUGIN itself raises, before the framework's scrub sees it."""
|
|
try:
|
|
run(P.PriceService().fetch(P.PluginOptions(opts()), NOW,
|
|
NOW + timedelta(hours=48), transport))
|
|
except SourceUnavailable as e:
|
|
return str(e)
|
|
return ""
|
|
|
|
|
|
_own = _fetch_error(_hostile)
|
|
check("the plugin's own error text has no token in it even before the "
|
|
"framework scrubs (defence in depth: two layers, each graded alone)",
|
|
_own.startswith("the price service did not answer (RuntimeError)")
|
|
and TOKEN not in _own)
|
|
|
|
|
|
async def _hostile2(url, timeout_s, headers=None):
|
|
raise SourceUnavailable(f"bad: {headers}")
|
|
|
|
|
|
check("a SourceUnavailable built from the headers is scrubbed too",
|
|
TOKEN not in run(load_prices(opts(), NOW, NOW + timedelta(hours=48),
|
|
transport=_hostile2)).reason)
|
|
from app import uilayout as _L, uipage as _UP # noqa: E402
|
|
check("the settings page never draws the token (it is a hidden, secret key), "
|
|
"and it can keep the provider but not choose it",
|
|
"price_service_token" in _L.HIDDEN
|
|
and _L.field("price_service_token") is None
|
|
and "price_service" in _UP.ONLY_IF_HELD)
|
|
|
|
# =========================================================================== #
|
|
print("\n8. the maintenance charge gate: the Home Assistant rule, from the "
|
|
"provider's prices")
|
|
|
|
|
|
def ha_state_now(rows, t):
|
|
"""sensor.price_service_now, as price_service.yaml computes it."""
|
|
u = [r[0] for r in rows]
|
|
p = [r[1] for r in rows]
|
|
step = (u[1] - u[0]) if len(u) > 1 else 900
|
|
v = None
|
|
for i in range(len(u)):
|
|
if u[i] <= t < u[i] + step:
|
|
v = p[i]
|
|
if not (len(u) > 0 and u[0] <= t < u[-1] + 900):
|
|
return None
|
|
return None if v is None else round(v / 1000, 5)
|
|
|
|
|
|
def ha_state_avg(rows, day_start, day_end):
|
|
"""sensor.price_service_avg, as price_service.yaml computes it."""
|
|
s = n = 0
|
|
for ts, pr in rows:
|
|
if day_start <= ts < day_end:
|
|
s += pr
|
|
n += 1
|
|
return round(s / n / 1000, 5) if n else None
|
|
|
|
|
|
def ha_cheap(rows, t, today):
|
|
"""The add-on's old rule: `now_p <= avg_p`, None means not cheap."""
|
|
now_p = ha_state_now(rows, t)
|
|
a = int(local_midnight(today).timestamp())
|
|
b = int(local_midnight(today + timedelta(days=1)).timestamp())
|
|
avg_p = ha_state_avg(rows, a, b)
|
|
return now_p is not None and avg_p is not None and now_p <= avg_p
|
|
|
|
|
|
def published_series(day_list, markup=MARKUP, taxes=TAXES):
|
|
"""The all-in series the provider would publish for these days, built by
|
|
the real framework from the mock's rows."""
|
|
c = make_ctx(now=local_midnight(day_list[0]) + timedelta(hours=10),
|
|
price_markup_eur_kwh=markup, price_taxes_eur_kwh=taxes)
|
|
c.net.server.days = lambda: day_list
|
|
p = refresh(c)
|
|
assert p.step == C.STEP_OWN, p.step
|
|
return p.published, [r for d in day_list for r in day_rows(d)]
|
|
|
|
|
|
_days = [date(2026, 10, 14), date(2026, 10, 15), date(2026, 3, 29),
|
|
date(2026, 10, 25), date(2026, 7, 1), date(2026, 1, 2)]
|
|
_diffs, _checked, _cheap_yes = [], 0, 0
|
|
for _d in _days:
|
|
_ser, _rows = published_series([_d, _d + timedelta(days=1)])
|
|
for _t in day_starts(_d):
|
|
_want = ha_cheap(_rows, int(_t.timestamp()), _d)
|
|
for dt in (0, 450, 899): # first, middle and last second
|
|
_tt = _t + timedelta(seconds=dt)
|
|
_got = C.gate_for(_ser, _tt).cheap
|
|
_checked += 1
|
|
if _got != _want:
|
|
_diffs.append((_d.isoformat(), _tt.isoformat(), _want, _got))
|
|
_cheap_yes += _want
|
|
check(f"{_checked} instants over {len(_days)} days (96-period days, the two "
|
|
f"clock-change days, winter, summer): the provider's gate equals "
|
|
f"now_p <= avg_p on the same prices, every time ({_diffs[:2]})",
|
|
not _diffs and _checked > 1000)
|
|
check("...and the days really separate: a fifth to four fifths of periods are "
|
|
"cheap, so neither 'always' nor 'never' passes",
|
|
0.2 < _cheap_yes / (_checked / 3) < 0.8)
|
|
_diffs = []
|
|
for _mk, _tx in ((0.0, 0.0), (0.03, 0.11736), (0.04321, 0.09876),
|
|
(0.0, 0.2)):
|
|
_ser, _rows = published_series([date(2026, 10, 14), date(2026, 10, 15)],
|
|
_mk, _tx)
|
|
for _t in day_starts(date(2026, 10, 14)):
|
|
if C.gate_for(_ser, _t).cheap != ha_cheap(_rows, int(_t.timestamp()),
|
|
date(2026, 10, 14)):
|
|
_diffs.append((_mk, _tx, _t.isoformat()))
|
|
check(f"...for markup and taxes of up to five decimals, with the wholesale "
|
|
f"sensor unchanged: the same answer ({_diffs[:2]})", not _diffs)
|
|
|
|
# the same day through the REAL controller, both ways
|
|
async def _both_ways(day):
|
|
rows = [r for d in (day, day + timedelta(days=1)) for r in day_rows(d)]
|
|
out = []
|
|
t0 = local_midnight(day)
|
|
with Wired(t0 + timedelta(hours=10), price_markup_eur_kwh=MARKUP,
|
|
price_taxes_eur_kwh=TAXES) as W:
|
|
W.net.server.days = lambda: [day, day + timedelta(days=1)]
|
|
W.net.ec.tomorrow = True
|
|
W.clock.now = t0 + timedelta(hours=10)
|
|
await W.refresh()
|
|
ps = W.ctl
|
|
ha_hass = FakeHass()
|
|
FakeDT._clock = W.clock
|
|
ha = controller(W.clock, {"price_provider": "off",
|
|
"price_now_entity": "sensor.now",
|
|
"price_avg_entity": "sensor.avg"})
|
|
ha.hass = ha_hass
|
|
a, b = (int(t0.timestamp()),
|
|
int(local_midnight(day + timedelta(days=1)).timestamp()))
|
|
last = None
|
|
for t in day_starts(day):
|
|
W.clock.now = t + timedelta(seconds=100)
|
|
# The background refresh runs every 30 minutes; a plan older than
|
|
# 90 minutes is retired, which is what keeps this gate shut when
|
|
# the refresh task has died.
|
|
if last is None or (t - last) >= timedelta(minutes=30):
|
|
await W.refresh(W.clock.now)
|
|
last = t
|
|
ha_hass.values["sensor.now"] = ha_state_now(
|
|
rows, int(W.clock.now.timestamp()))
|
|
ha_hass.values["sensor.avg"] = ha_state_avg(rows, a, b)
|
|
await ps.read_inputs()
|
|
await ha.read_inputs()
|
|
out.append((ha.cheap, ps.cheap,
|
|
ha._charge_gate_open(ha.inverters[0]),
|
|
ps._charge_gate_open(ps.inverters[0]),
|
|
ha._price_gated, ps._price_gated))
|
|
return out
|
|
|
|
|
|
_rows_both = run(_both_ways(date(2026, 10, 14)))
|
|
check(f"the same day fed through the Home Assistant pair and through "
|
|
f"price_service: {len(_rows_both)} periods, the cheap flag and the "
|
|
f"charge gate agree in every one",
|
|
len(_rows_both) == 96
|
|
and all(a == b and c == d and e == f
|
|
for a, b, c, d, e, f in _rows_both))
|
|
check("...with both kinds of period present (cheap and not cheap)",
|
|
{r[0] for r in _rows_both} == {True, False})
|
|
check("...and an open gate whenever it is cheap, a shut one when it is not "
|
|
"(no pack is charging)",
|
|
all(r[3] == r[1] for r in _rows_both))
|
|
|
|
# what the old rule did when the price was unavailable, and what this does
|
|
_ser, _rows = published_series([date(2026, 10, 14)])
|
|
_late = local_midnight(date(2026, 10, 15)) + timedelta(minutes=1)
|
|
check("with no price for now both rules say not cheap (the HA sensor was "
|
|
"'unavailable'), and the gate stays shut",
|
|
C.gate_for(_ser, _late).cheap is False
|
|
and ha_cheap(_rows, int(_late.timestamp()), date(2026, 10, 15)) is False)
|
|
check("it is NOT TariffPlan.cheap_now(): on a price exactly at the average "
|
|
"this gate says cheap where cheap_now (strictly below) says no",
|
|
C.gate_for(
|
|
P.PriceRead.series(
|
|
[P.PricePoint(local_midnight(date(2026, 10, 14))
|
|
+ timedelta(minutes=15 * i), v, False)
|
|
for i, v in enumerate((0.10, 0.20, 0.30))],
|
|
PriceProduct.DAY_AHEAD, "price_service", NOW),
|
|
local_midnight(date(2026, 10, 14)) + timedelta(minutes=16)).cheap
|
|
is True)
|
|
|
|
# The Home Assistant average was rounded to 5 decimals before it was compared,
|
|
# so a quarter that sits ABOVE the exact average but AT the rounded one was
|
|
# cheap. Real days almost never land on that, so it is built here: three
|
|
# periods at 0.20000 and five at 0.20001 average 0.20000625, which rounds to
|
|
# 0.20001, and a period at 0.20001 is therefore cheap by the old rule.
|
|
_t0 = local_midnight(date(2026, 10, 14))
|
|
_vals = [0.20000] * 3 + [0.20001] * 5
|
|
_toy = P.PriceRead.series(
|
|
[P.PricePoint(_t0 + timedelta(minutes=15 * i), v, False)
|
|
for i, v in enumerate(_vals)], PriceProduct.DAY_AHEAD, "price_service",
|
|
NOW)
|
|
_toy_rows = [(int((_t0 + timedelta(minutes=15 * i)).timestamp()),
|
|
round(v * 1000, 2)) for i, v in enumerate(_vals)]
|
|
_at = lambda i: _t0 + timedelta(minutes=15 * i + 1) # noqa: E731
|
|
check("a period above the exact average but at the 5-decimal rounded one is "
|
|
"cheap, as the Home Assistant pair said: the gate and the pair agree on "
|
|
"all eight periods of a day built to sit on that edge, and the edge "
|
|
"period is cheap",
|
|
[C.gate_for(_toy, _at(i)).cheap for i in range(8)]
|
|
== [ha_cheap(_toy_rows, int(_at(i).timestamp()), date(2026, 10, 14))
|
|
for i in range(8)]
|
|
and C.gate_for(_toy, _at(7)).cheap is True)
|
|
|
|
# =========================================================================== #
|
|
print("\n9. battery modes plan on the provider's prices")
|
|
async def _modes():
|
|
day = date(2026, 10, 14)
|
|
t = local_midnight(day) + timedelta(hours=8)
|
|
with Wired(t, battery_modes_enabled=True) as W:
|
|
await W.refresh()
|
|
c = W.ctl
|
|
plan = c.fresh_tariff_plan(t)
|
|
return (plan is not None and plan.has_prices, plan.origin,
|
|
plan.n_firm > 150,
|
|
[p.provider for p in [plan.read]][0], c.modes_on)
|
|
|
|
|
|
_m = run(_modes())
|
|
check("the standing plan is a feed plan of firm price_service prices covering "
|
|
"the two days, and battery modes are on",
|
|
_m[0] and _m[1] == "provider" and _m[2] and _m[3] == "price_service"
|
|
and _m[4] is True)
|
|
|
|
async def _modes_plan():
|
|
day = date(2026, 10, 14)
|
|
t = local_midnight(day) + timedelta(hours=8)
|
|
with Wired(t, battery_modes_enabled=True, arbitrage_enabled=True) as W:
|
|
await W.refresh()
|
|
c = W.ctl
|
|
plan = c.fresh_tariff_plan(t)
|
|
verdict = plan.arbitrage(
|
|
peak_guard_active=False, min_spread_eur_kwh=0.05,
|
|
uncertainty=frozenset())
|
|
return verdict.match(allowed=lambda v: ("allowed", v.spread_eur_kwh),
|
|
refused=lambda v: ("refused", v.reason))
|
|
|
|
|
|
_v = run(_modes_plan())
|
|
check(f"...and the arbitrage verdict on them is a real one over firm prices "
|
|
f"{_v}", _v[0] == "allowed" and _v[1] > 0.1)
|
|
|
|
async def _runs_for(provider):
|
|
"""The kept price-run plan of a battery in cheap-hours mode, built by the
|
|
shipped controller from a plan that came through `provider`."""
|
|
from app import packmode
|
|
t = local_midnight(date(2026, 10, 14)) + timedelta(hours=8)
|
|
over = {"battery_modes_enabled": True, "capacity_kwh": 10}
|
|
if provider != "price_service":
|
|
over["price_provider"] = provider
|
|
with Wired(t, **over) as W:
|
|
await W.refresh()
|
|
c = W.ctl
|
|
c.store = FakeStore()
|
|
c.store.data = {packmode.unit_key("inverter-a", packmode.MODE_KEY):
|
|
packmode.CHEAP_HOURS}
|
|
c.inverters[0].soc, c.inverters[0].capacity_kwh = 40.0, 10.0
|
|
return c._price_runs(t, c._mode_packs(), 20, 95)
|
|
|
|
|
|
_ra, _rb = run(_runs_for("price_service")), run(_runs_for("generic_dayahead"))
|
|
check("a battery in cheap-hours mode gets a real price-run plan from "
|
|
"price_service prices (a cheap run is chosen)",
|
|
_ra is not None and _ra.cheap is not None)
|
|
check("...the same plan, run for run, that generic_dayahead gives on the same "
|
|
"prices: the modes read these prices as they read those",
|
|
_ra == _rb)
|
|
|
|
# =========================================================================== #
|
|
print("\n10. the options: validation, defaults, the old pair")
|
|
def problems(**over):
|
|
o = {"price_provider": "price_service", "price_service_token": TOKEN}
|
|
o.update(over)
|
|
base = dict(OPTS_C)
|
|
base.update(o)
|
|
return validate_startup_config(base, None, None)
|
|
|
|
|
|
check("price_service with a token starts", problems() == [])
|
|
_e = problems(price_service_token="")
|
|
check("...without a token it is refused, naming both settings by label and key "
|
|
f"({_e})", len(_e) == 1 and "“Dynamic price source” (price_provider)"
|
|
in _e[0] and "“Price service token” (price_service_token)" in _e[0]
|
|
and TOKEN not in _e[0])
|
|
# Long enough to pass the length rule, so only the space can refuse it.
|
|
_SPACED = "tk-0123456789 abcdefgh"
|
|
_e = problems(price_service_token=_SPACED)
|
|
check("...a token with a space in it is refused for the space, and the token "
|
|
"is not echoed",
|
|
len(_e) == 1 and "space" in _e[0] and _SPACED not in _e[0])
|
|
_e = problems(price_now_entity="sensor.a", price_avg_entity="sensor.b")
|
|
check(f"the existing conflict refusal stands: the Home Assistant pair beside "
|
|
f"this provider is refused ({len(_e)} problem)",
|
|
any("would win over it" in x for x in _e))
|
|
check("a static_profile beside it is accepted (it is the fallback for "
|
|
"trading)", problems(static_profile=json.dumps({"periods": [
|
|
{"name": "all", "days": "all", "from": "00:00", "to": "24:00",
|
|
"eur_per_kwh": 0.25}]})) == [])
|
|
check("the shipped default token is empty and the option is a password field",
|
|
P.PRICE_OPTION_DEFAULTS["price_service_token"] == ""
|
|
and "price_service_token: password?" in open(
|
|
os.path.join(HERE, "config.yaml")).read())
|
|
check("config.yaml offers the new provider id",
|
|
"price_service" in open(os.path.join(HERE, "config.yaml")).read()
|
|
.split("price_provider: list(")[1].split(")")[0])
|
|
check("only this provider asks for the chain; the others take the road they "
|
|
"always took",
|
|
[pid for pid in sorted(PROVIDERS)
|
|
if C.uses_chain({"price_provider": pid})] == ["price_service"]
|
|
and not C.uses_chain({"price_provider": "off"}))
|
|
|
|
# =========================================================================== #
|
|
print("\n11. the real transport, against a local server")
|
|
async def _wire():
|
|
from aiohttp import web
|
|
seen = []
|
|
hold = {"mode": "ok"}
|
|
|
|
async def handler(request):
|
|
seen.append(dict(request.headers))
|
|
if hold["mode"] == "slow":
|
|
await asyncio.sleep(3)
|
|
if hold["mode"] in ("401", "403", "429", "503", "418"):
|
|
return web.Response(status=int(hold["mode"]), text="nope " + TOKEN)
|
|
if request.headers.get("Authorization") != "Bearer " + TOKEN:
|
|
return web.Response(status=401, text="bad token")
|
|
return web.Response(body=b'{"ok": 1}')
|
|
elsewhere = []
|
|
|
|
async def other(request):
|
|
elsewhere.append(dict(request.headers))
|
|
return web.Response(body=b'{"ok": 2}')
|
|
|
|
async def moved(request):
|
|
raise web.HTTPFound("/elsewhere")
|
|
app = web.Application()
|
|
app.router.add_get("/v1/dayahead/BE", handler)
|
|
app.router.add_get("/moved", moved)
|
|
app.router.add_get("/elsewhere", other)
|
|
runner = web.AppRunner(app)
|
|
await runner.setup()
|
|
site = web.TCPSite(runner, "127.0.0.1", 0)
|
|
await site.start()
|
|
port = site._server.sockets[0].getsockname()[1]
|
|
url = f"http://127.0.0.1:{port}/v1/dayahead/BE"
|
|
res = {}
|
|
hdr = {"Authorization": "Bearer " + TOKEN}
|
|
res["ok"] = await P.https_get(url, 3, headers=hdr)
|
|
res["seen"] = nth(seen, -1)
|
|
for st in ("401", "403", "429", "503", "418"):
|
|
hold["mode"] = st
|
|
try:
|
|
await P.https_get(url, 3, headers=hdr)
|
|
res[st] = "no error"
|
|
except SourceUnavailable as e:
|
|
res[st] = str(e)
|
|
except Exception as e: # noqa: BLE001
|
|
res[st] = "LEAK " + repr(e)
|
|
try:
|
|
await P.https_get(url.replace("/v1/dayahead/BE", "/moved"), 3,
|
|
headers=hdr)
|
|
res["302"] = "no error"
|
|
except SourceUnavailable as e:
|
|
res["302"] = str(e)
|
|
res["followed"] = len(elsewhere)
|
|
hold["mode"] = "slow"
|
|
try:
|
|
await P.https_get(url, 0.4, headers=hdr)
|
|
res["slow"] = "no error"
|
|
except SourceUnavailable as e:
|
|
res["slow"] = str(e)
|
|
except Exception as e: # noqa: BLE001
|
|
res["slow"] = "LEAK " + repr(e)
|
|
await runner.cleanup()
|
|
try:
|
|
await P.https_get(url, 1, headers=hdr)
|
|
res["closed"] = "no error"
|
|
except SourceUnavailable as e:
|
|
res["closed"] = str(e)
|
|
except Exception as e: # noqa: BLE001
|
|
res["closed"] = "LEAK " + repr(e)
|
|
# without headers it is the old transport: a status raises aiohttp's error
|
|
return res
|
|
|
|
|
|
_w = run(_wire())
|
|
check("the token went on the wire as a bearer header, and the body came back",
|
|
_w["seen"].get("Authorization") == "Bearer " + TOKEN
|
|
and _w["ok"] == b'{"ok": 1}')
|
|
check(f"401 and 403 are 'refused the token', with the status and no body "
|
|
f"({_w['401']!r})",
|
|
_w["401"] == "the price service refused the token (401)"
|
|
and _w["403"] == "the price service refused the token (403)")
|
|
check(f"429, 5xx and other statuses have their own plain words "
|
|
f"({_w['429']!r}, {_w['503']!r}, {_w['418']!r})",
|
|
_w["429"] == "the price service says too many requests (429)"
|
|
and _w["503"] == "the price service has an error (503)"
|
|
and _w["418"] == "the price service answered 418")
|
|
check(f"a redirect is not followed, so the token cannot be taken to another "
|
|
f"place: it is reported as a status and the target was never asked "
|
|
f"({_w['302']!r}, requests to the target: {_w['followed']})",
|
|
_w["302"] == "the price service answered 302" and _w["followed"] == 0)
|
|
check(f"a timeout and a refused connection are SourceUnavailable with the "
|
|
f"exception's NAME only ({_w['slow']!r}, {_w['closed']!r})",
|
|
_w["slow"].startswith("the price service did not answer (")
|
|
and _w["closed"].startswith("the price service did not answer ("))
|
|
check("...and nothing the server sent back, token echo included, reached any "
|
|
"message", all(TOKEN not in v for k, v in _w.items()
|
|
if isinstance(v, str)))
|
|
|
|
# =========================================================================== #
|
|
print("\n12. restart in the middle of an outage, and the shipped logger names")
|
|
_c = make_ctx()
|
|
refresh(_c)
|
|
_c.net.down()
|
|
_hist2 = PriceHistory(_c.hist.path) # a restart: new objects
|
|
_c2 = type("C", (), {"clock": _c.clock, "net": _c.net, "hist": _hist2,
|
|
"watch": C.PriceWatch(), "o": _c.o})()
|
|
_n = mark()
|
|
_p = refresh(_c2, NOW + timedelta(hours=1))
|
|
check("a restart with every source down plans from what the file kept",
|
|
_p.step == C.STEP_STORED and covers_now(_p, _c2.clock.now))
|
|
check("...and says so once, at WARNING",
|
|
since_mark(_n, logging.WARNING)
|
|
== ["No price source answers, so the add-on uses the prices it saved."])
|
|
check("the chain logs under goodwe.price.chain",
|
|
all(r.name == "goodwe.price.chain" for r in TAP.records[_n:]
|
|
if "price source" in r.getMessage() or "price service"
|
|
in r.getMessage()))
|
|
|
|
# =========================================================================== #
|
|
print("\n13. round 2: the no-price words only with the maintenance charge on (F3)")
|
|
async def _scenario_maint(on):
|
|
"""Every source down, six ticks, then the prices come back. `on` is the
|
|
maintenance switch."""
|
|
from app import homeview
|
|
with Wired(NOW, maintenance_enabled=on) as W:
|
|
c = W.ctl
|
|
W.net.down()
|
|
await W.refresh()
|
|
for k in range(6):
|
|
await W.tick(NOW + timedelta(seconds=30 * k))
|
|
mid = {"gap": c.price_gap, "notes": list(W.notes()),
|
|
"row": dict(c.status_payload()["rows"]).get("Prices"),
|
|
"home": [x for x in homeview.build_plan(
|
|
c, W.clock.now, False, True)["lines"]
|
|
if "No prices" in x],
|
|
"gate": c._charge_gate_open(c.inverters[0]),
|
|
"cheap": c.cheap}
|
|
W.net.up()
|
|
await W.refresh(NOW + timedelta(minutes=30))
|
|
await W.tick(NOW + timedelta(minutes=30))
|
|
return mid, list(W.notes())
|
|
|
|
|
|
_on_mid, _on_all = run(_scenario_maint(True))
|
|
_off_mid, _off_all = run(_scenario_maint(False))
|
|
check("maintenance on: the row, the Home line and ONE notification appear",
|
|
_on_mid["gap"] is True and len(_on_mid["notes"]) == 1
|
|
and _on_mid["row"] == "No prices yet, maintenance waits."
|
|
and _on_mid["home"] == ["No prices yet, maintenance waits."])
|
|
check("...and the notification is taken down once when prices return",
|
|
[n[0] for n in _on_all] == ["create", "dismiss"])
|
|
check("maintenance off: no row, no Home line, no notification, at any tick",
|
|
_off_mid["gap"] is False and _off_mid["notes"] == []
|
|
and _off_mid["row"] is None and _off_mid["home"] == []
|
|
and _off_all == [])
|
|
check("...and the gate itself does not change with the switch: still shut "
|
|
"with no price",
|
|
_off_mid["gate"] is False and _on_mid["gate"] is False
|
|
and _off_mid["cheap"] is False)
|
|
|
|
|
|
async def _scenario_switched_off():
|
|
"""A notification raised with the cycle on is cleared when it is off."""
|
|
with Wired(NOW) as W:
|
|
c = W.ctl
|
|
W.net.down()
|
|
await W.refresh()
|
|
await W.tick(NOW)
|
|
raised = list(W.notes())
|
|
c.o["maintenance_enabled"] = False
|
|
await W.tick(NOW + timedelta(seconds=30))
|
|
await W.tick(NOW + timedelta(seconds=60))
|
|
return raised, list(W.notes()), c.price_gap
|
|
|
|
|
|
_r, _n, _g = run(_scenario_switched_off())
|
|
check("a notification raised while the cycle was on is cleared, once, when "
|
|
"it is off",
|
|
len(_r) == 1 and [x[0] for x in _n] == ["create", "dismiss"]
|
|
and _g is False)
|
|
|
|
|
|
async def _scenario_key_missing():
|
|
"""No `maintenance_enabled` key at all: its shipped default (off) holds."""
|
|
with Wired(NOW) as W:
|
|
c = W.ctl
|
|
c.o.pop("maintenance_enabled", None)
|
|
W.net.down()
|
|
await W.refresh()
|
|
for k in range(3):
|
|
await W.tick(NOW + timedelta(seconds=30 * k))
|
|
return c.price_gap, list(W.notes())
|
|
|
|
|
|
_g2, _n2 = run(_scenario_key_missing())
|
|
check("with no maintenance_enabled key its default (off) holds: no no-price "
|
|
"words, no notification", _g2 is False and _n2 == [])
|
|
|
|
async def _modes_no_price():
|
|
"""Battery modes on, maintenance off, no price anywhere: what happens."""
|
|
from app import packmode, homeview
|
|
with Wired(NOW, battery_modes_enabled=True, capacity_kwh=10,
|
|
maintenance_enabled=False) as W:
|
|
c = W.ctl
|
|
W.net.down()
|
|
await W.refresh()
|
|
c.store = FakeStore()
|
|
c.store.data = {packmode.unit_key("inverter-a", packmode.MODE_KEY):
|
|
packmode.CHEAP_HOURS}
|
|
c.inverters[0].soc, c.inverters[0].capacity_kwh = 40.0, 10.0
|
|
await W.tick(NOW)
|
|
runs = c._price_runs(NOW, c._mode_packs(), 20, 95)
|
|
plan = homeview.build_plan(c, NOW, False, True)["lines"]
|
|
return runs, c.price_gap, list(W.notes()), plan
|
|
|
|
|
|
_mr, _mg, _mn, _ml = run(_modes_no_price())
|
|
check("battery modes with no price and maintenance off: no price run is "
|
|
"made (so no battery holds or trades on price), and nothing about "
|
|
"'maintenance waits' is shown or notified",
|
|
_mr is None and _mg is False and _mn == []
|
|
and not [x for x in _ml if "maintenance waits" in x])
|
|
print(" Home plan lines in that state:", _ml)
|
|
|
|
# =========================================================================== #
|
|
print("\n14. round 3: a fixed tariff beside price_service is a log line and a "
|
|
"settings note, no notification (G1, S05)")
|
|
_PROFILE = json.dumps({"periods": [
|
|
{"name": "all", "days": "all", "from": "00:00", "to": "24:00",
|
|
"eur_per_kwh": 0.25}]})
|
|
_LINE = "Fixed tariff is only used if no prices are available."
|
|
_NOTE = ("With the price service, the fixed tariff is only used if no prices "
|
|
"are available.")
|
|
_OLD_ID = "goodwe_fixed_tariff_unused"
|
|
|
|
|
|
async def _scenario_fixed(**over):
|
|
with Wired(NOW, **over) as W:
|
|
before = mark()
|
|
await W.refresh()
|
|
for k in range(4):
|
|
await W.tick(NOW + timedelta(seconds=30 * k))
|
|
said = [m for m in since_mark(before, logging.INFO) if "only used if no prices" in m]
|
|
return said, W.all_notes(), W.ctl
|
|
|
|
|
|
def _settings_notes(**over):
|
|
base = dict(OPTS_C)
|
|
base.update(price_provider="price_service", price_service_token=TOKEN)
|
|
base.update(over)
|
|
out = []
|
|
validate_startup_config(base, None, None, notes=out)
|
|
return [n.text for n in out if n.keys == ("static_profile",)
|
|
and "fixed tariff" in n.text.lower()]
|
|
|
|
|
|
_said, _fn, _fc = run(_scenario_fixed(static_profile=_PROFILE))
|
|
check("price_service + a set profile: the INFO line appears once over four "
|
|
"ticks, and the setting is not refused",
|
|
_said == [_LINE] and problems(static_profile=_PROFILE) == [])
|
|
check("...no notification is raised, only the old id is dismissed",
|
|
[x for x in _fn if x[1].get("notification_id") == _OLD_ID]
|
|
== [("dismiss", {"notification_id": _OLD_ID})]
|
|
and not [x for x in _fn if x[0] == "create"])
|
|
for _label, _kw in (("no profile", {}),
|
|
("an empty profile", {"static_profile": ""}),
|
|
("a spaces-only profile", {"static_profile": " "}),
|
|
("a line-break-only profile", {"static_profile": "\n \t"}),
|
|
("another provider", {"price_provider": "generic_dayahead",
|
|
"static_profile": _PROFILE})):
|
|
_s, _n, _c = run(_scenario_fixed(**_kw))
|
|
check(f"{_label}: no log line", _s == [])
|
|
check(f"{_label}: the old id is still dismissed at start (harmless)",
|
|
[x for x in _n if x[1].get("notification_id") == _OLD_ID]
|
|
== [("dismiss", {"notification_id": _OLD_ID})])
|
|
_off_s, _, _ = run(_scenario_fixed(price_provider="off",
|
|
static_profile=_PROFILE))
|
|
check("provider off with a profile: no log line", _off_s == [])
|
|
|
|
|
|
async def _scenario_fixed_level_and_raise():
|
|
"""The line's level, and a dismiss that raises: the cycle goes on."""
|
|
with Wired(NOW, static_profile=_PROFILE) as W:
|
|
real = W.ctl.hass.call
|
|
|
|
async def call(domain, service, data):
|
|
if service == "dismiss" and data.get("notification_id") == _OLD_ID:
|
|
raise RuntimeError("Home Assistant said no")
|
|
return await real(domain, service, data)
|
|
W.ctl.hass.call = call
|
|
before = mark()
|
|
await W.refresh()
|
|
try:
|
|
for k in range(2):
|
|
await W.tick(NOW + timedelta(seconds=30 * k))
|
|
done = True
|
|
except Exception as e: # noqa: BLE001 - the check reports it
|
|
done = repr(e)
|
|
mine = [(r.levelno, r.getMessage()) for r in TAP.records[before:]
|
|
if "only used if no prices" in r.getMessage()]
|
|
return done, mine
|
|
|
|
|
|
_done, _lvl = run(_scenario_fixed_level_and_raise())
|
|
check(f"the start line is INFO, not a warning ({_lvl})",
|
|
_lvl == [(logging.INFO, _LINE)])
|
|
check(f"a dismiss that raises does not stop the cycle ({_done})", _done is True)
|
|
check("the log line is the same words as the note's second half",
|
|
_NOTE.endswith(_LINE[0].lower() + _LINE[1:]))
|
|
|
|
check("settings page: price_service + a set profile shows the one note, "
|
|
"beside static_profile", _settings_notes(static_profile=_PROFILE) == [_NOTE])
|
|
check("...not with no, empty, spaces-only or line-break-only profile",
|
|
all(_settings_notes(**kw) == [] for kw in (
|
|
{}, {"static_profile": ""}, {"static_profile": " "},
|
|
{"static_profile": "\n \t"})))
|
|
check("...not with another provider",
|
|
_settings_notes(price_provider="generic_dayahead",
|
|
static_profile=_PROFILE) == []
|
|
and _settings_notes(price_provider="off",
|
|
static_profile=_PROFILE) == [])
|
|
_nn = []
|
|
validate_startup_config(dict(OPTS_C, price_provider="price_service",
|
|
price_service_token=TOKEN,
|
|
static_profile=_PROFILE), None, None, notes=_nn)
|
|
_NOTE_NL = ("Met de prijsdienst wordt het vaste tarief alleen gebruikt als er "
|
|
"geen prijzen zijn.")
|
|
_en = _UP.anchors([], _nn, {"price_provider": "price_service"},
|
|
["price_provider", "static_profile"], "en")
|
|
_nl = _UP.anchors([], _nn, {"price_provider": "price_service"},
|
|
["price_provider", "static_profile"], "nl")
|
|
check("the page puts the note beside the fixed-tariff field, in English and "
|
|
f"Dutch ({_en})",
|
|
[m["text"] for m in _en] == [_NOTE] and [m["text"] for m in _nl]
|
|
== [_NOTE_NL] and _en[0]["tier"] == "info" and _en[0]["field"]
|
|
and _en[0]["field"] == _nl[0]["field"])
|
|
check("...and the note names no refusal: the options are still accepted",
|
|
problems(static_profile=_PROFILE) == [])
|
|
|
|
|
|
async def _scenario_profile_roles(profile):
|
|
"""Prices up: what the cheap window and the gate say. Then every source
|
|
down: what the plan falls back to."""
|
|
out = {}
|
|
over = {"static_profile": profile} if profile else {}
|
|
with Wired(NOW, **over) as W:
|
|
c = W.ctl
|
|
await W.refresh()
|
|
await W.tick()
|
|
out["cheap"], out["gate"] = c.cheap, c._charge_gate_open(c.inverters[0])
|
|
out["words"] = c.tariff_words
|
|
out["origin_up"] = c.tariff_plan.origin
|
|
W.net.down()
|
|
later = NOW + timedelta(days=5) # saved prices run out
|
|
await W.refresh(later)
|
|
await W.tick(later)
|
|
out["origin_down"] = c.tariff_plan.origin
|
|
out["gate_down"] = c._charge_gate_open(c.inverters[0])
|
|
out["points"] = c.tariff_plan.read.match(
|
|
series=lambda r: [pt.eur_per_kwh for pt in r.points][:2],
|
|
no_data=lambda r: [])
|
|
return out
|
|
|
|
|
|
_with = run(_scenario_profile_roles(_PROFILE))
|
|
_without = run(_scenario_profile_roles(None))
|
|
check("with prices up, the cheap window and the gate are the same with or "
|
|
"without a fixed tariff set",
|
|
(_with["cheap"], _with["gate"], _with["words"])
|
|
== (_without["cheap"], _without["gate"], _without["words"])
|
|
and _with["origin_up"] == "provider")
|
|
check("with every source down, the plan falls back to the fixed tariff "
|
|
f"({_with['origin_down']}, {_with['points']}) and without it there is none",
|
|
_with["origin_down"] == "fallback" and _with["points"]
|
|
and set(_with["points"]) == {0.25}
|
|
and _without["origin_down"] == "none")
|
|
check("...and the gate stays shut on that fallback (it is not a firm price)",
|
|
_with["gate_down"] is False)
|
|
|
|
# =========================================================================== #
|
|
print("\n15. round 2: a token shorter than 16 characters is refused (F5)")
|
|
_short = problems(price_service_token="a" * 15)
|
|
check(f"15 characters: refused with one short plain sentence ({_short})",
|
|
_short == ["“Price service token” (price_service_token) is too short. "
|
|
"Paste the whole token."])
|
|
check("...16 characters start", problems(price_service_token="a" * 16) == [])
|
|
check("...the refusal does not echo the value",
|
|
"a" * 15 not in "".join(_short))
|
|
|
|
# =========================================================================== #
|
|
print("\n16. round 2: the guards the reviewer's mutations R01 and R10 got past")
|
|
_t0 = local_midnight(date(2026, 10, 14))
|
|
_now0 = _t0 + timedelta(minutes=31) # inside period 2
|
|
_pts = [P.PricePoint(_t0 + timedelta(minutes=15 * i), v, i == 2)
|
|
for i, v in enumerate((0.10, 0.20, 0.05, 0.40))]
|
|
_prov = P.PriceRead.series(_pts, PriceProduct.DAY_AHEAD, "price_service", NOW)
|
|
_g = C.gate_for(_prov, _now0)
|
|
check("R01: if the quarter holding now is an estimate it is not priced as "
|
|
"firm, even though it is cheap: the gate is shut and says 'no price'",
|
|
_g.cheap is False and _g.covered is False
|
|
and _g.words == "No price for now")
|
|
_firm = P.PriceRead.series(
|
|
[P.PricePoint(p.start, p.eur_per_kwh, False) for p in _pts],
|
|
PriceProduct.DAY_AHEAD, "price_service", NOW)
|
|
check("...the same series with every quarter firm opens (it is cheap), so "
|
|
"only the estimate flag shut it",
|
|
C.gate_for(_firm, _now0).cheap is True
|
|
and C.gate_for(_firm, _now0).covered is True)
|
|
_prov2 = P.PriceRead.series(
|
|
[P.PricePoint(_t0 + timedelta(minutes=15 * i), v, i == 3)
|
|
for i, v in enumerate((0.10, 0.15, 0.05, 0.40))],
|
|
PriceProduct.DAY_AHEAD, "price_service", NOW)
|
|
check("an estimated quarter elsewhere in the day stays out of today's "
|
|
"average: 0.15 against 0.10, 0.15, 0.05 is above their average (0.10), "
|
|
"where with the estimated 0.40 counted it would be cheap (0.175)",
|
|
C.gate_for(_prov2, _t0 + timedelta(minutes=16)).cheap is False)
|
|
|
|
|
|
async def _plain_wire():
|
|
from aiohttp import web, ClientResponseError
|
|
seen, elsewhere = [], []
|
|
|
|
async def ok(request):
|
|
seen.append(dict(request.headers))
|
|
return web.Response(body=b'{"ok": 1}')
|
|
|
|
async def moved(request):
|
|
raise web.HTTPFound("/elsewhere")
|
|
|
|
async def other(request):
|
|
elsewhere.append(1)
|
|
return web.Response(body=b'{"ok": 2}')
|
|
|
|
async def missing(request):
|
|
return web.Response(status=404, text="no")
|
|
|
|
async def broken(request):
|
|
return web.Response(status=503, text="no")
|
|
app = web.Application()
|
|
for path, fn in (("/ok", ok), ("/moved", moved), ("/elsewhere", other),
|
|
("/missing", missing), ("/broken", broken)):
|
|
app.router.add_get(path, fn)
|
|
runner = web.AppRunner(app)
|
|
await runner.setup()
|
|
site = web.TCPSite(runner, "127.0.0.1", 0)
|
|
await site.start()
|
|
base = f"http://127.0.0.1:{site._server.sockets[0].getsockname()[1]}"
|
|
res = {}
|
|
try:
|
|
res["ok"] = await P.https_get(base + "/ok", 3)
|
|
res["moved"] = await P.https_get(base + "/moved", 3)
|
|
for name in ("missing", "broken"):
|
|
try:
|
|
await P.https_get(base + "/" + name, 3)
|
|
res[name] = "no error"
|
|
except ClientResponseError as e:
|
|
res[name] = e.status
|
|
except Exception as e: # noqa: BLE001
|
|
res[name] = "other " + type(e).__name__
|
|
finally:
|
|
await runner.cleanup()
|
|
res["headers"] = seen[0] if seen else {}
|
|
res["followed"] = len(elsewhere)
|
|
return res
|
|
|
|
|
|
_pw = run(_plain_wire())
|
|
check("R10: the plain transport the other providers use returns the body "
|
|
"of a 200", _pw["ok"] == b'{"ok": 1}')
|
|
check("...follows a redirect, as it always did (the price service's own "
|
|
"path does not)",
|
|
_pw["moved"] == b'{"ok": 2}' and _pw["followed"] == 1)
|
|
check("...raises the library's own error for a 4xx and a 5xx, not the price "
|
|
"service's words",
|
|
_pw["missing"] == 404 and _pw["broken"] == 503)
|
|
check("...and sends no Authorization header of its own",
|
|
"Authorization" not in _pw["headers"])
|
|
|
|
for d in _TMP:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + "; ".join(fails))
|
|
sys.exit(1)
|
|
print("all checks passed")
|