goodwe-addon/goodwe_controller/test_price.py

"""Runnable check for the price provider framework.  `python3 test_price.py`

WHAT THIS FILE IS DEFENDING, in the order the ticket names it:

  1. Every shipped plugin emits the EXACT canonical schema (§9.3), all-in.
  2. Adding a plugin needs no optimiser change - so the framework must contain
     no plugin name, and a plugin invented HERE, in this file, must reach the
     right volatility regime without price.py being touched.
  3. `manual_csv` works with no network at all.
  4. The injection price is a separate scalar (§9.3).
  5. "No price data" is a DIFFERENT OUTCOME from "prices, all of them zero".

TWO RULES THIS FILE HOLDS ITSELF TO, because the reviewer will:

  * EVERY VERDICT IS DERIVED. There is no `check("the schema is right", True)`
    anywhere below. Where a check needs an expected number, the number is
    computed here from the same inputs by different arithmetic than price.py
    uses, and where it needs a set, the set is built from the objects rather
    than typed out.

  * NO STIMULUS IS EVALUATED WHERE THE QUANTITY IT GRADES IS DEGENERATE. That
    is not decoration here, it is the main hazard in a markup test:

      - markup and taxes must DIFFER and both be NON-ZERO, or "added markup
        twice" and "added markup and taxes once each" produce the same number
        and the double-counting check grades nothing. The shipped defaults are
        0.03 and 0.06 and 0.06 IS EXACTLY TWICE 0.03, so this file deliberately
        uses 0.031 / 0.057 for the arithmetic and asserts the non-degeneracy
        conditions as checks in their own right rather than assuming them.
      - the raw price must be non-zero, or an all-in figure equals markup+taxes
        and a plugin that dropped the energy component entirely still passes.
      - a fixture must carry BOTH a provisional and a non-provisional point, or
        `plan_is_provisional` returns the same answer under any bug.
      - the DST checks run on the 23-hour and 25-hour days, not on a normal
        24-hour day where a fixed-24-slot implementation and a correct one
        agree exactly.
"""

import asyncio
import inspect
import io
import json
import logging
import os
import sys
from datetime import datetime, timedelta, timezone

from app.price import (
    ALL_IN, Absence, Component, FrameworkOption, MAX_PLAUSIBLE_EUR_KWH,
    PRICE_OPTION_DEFAULTS, PROVIDERS, PluginOptions, PricePoint, PriceProduct,
    PriceProvider, PriceRead, ProviderCapabilities, RawPoint, SourceUnavailable,
    arbitrage_allowed, assemble, brussels, brussels_offset, check_egress,
    export_price_eur_kwh, load_prices, plan_is_provisional, register,
    volatility_regime,
)
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)


class _Raised:
    """A value a check wanted, whose computation raised.

    Compares unequal to everything, is falsy, and prints what went wrong.
    """

    def __init__(self, e):
        self.e = e

    def __eq__(self, other):
        return False

    def __hash__(self):
        return id(self)

    def __bool__(self):
        return False

    def __repr__(self):
        return f"<raised {type(self.e).__name__}: {self.e}>"


def derived(fn):
    """Compute a value a check grades. An exception becomes `_Raised`.

    ⚠️ THIS IS NOT A DEFAULT, and the distinction is the one this whole ticket
    is about. `_Raised` is not `[]`, not `None`, not 0; it equals nothing, it is
    falsy, and it renders as the exception that produced it. Its only job is
    that a mutation which makes a derivation IMPOSSIBLE - the reviewer removing
    `_NoData`'s missing `.points`, say - reports as a failed check naming the
    exception, instead of aborting this file before the sections below it run.
    A silent `except: return []` here would be the defect under test, in the
    harness that tests it.
    """
    try:
        return fn()
    except Exception as e:                                    # noqa: BLE001
        return _Raised(e)


UTC = timezone.utc

# ⚠️ NOT the shipped 0.03 / 0.06. See this module's docstring: 0.06 is exactly
# 2 x 0.03, so under the shipped pair "the framework added the markup twice and
# no taxes" and "the framework added one markup and one taxes" differ by
# 0.03 - but several OTHER confusions collapse. These two are co-prime-ish in
# the only sense that matters here, and the three degeneracy conditions are
# asserted below rather than assumed.
MARKUP = 0.031
TAXES = 0.057


def opts(**over):
    """A price configuration. Everything below is this, spoilt."""
    o = {"price_provider": "off",
         "price_markup_eur_kwh": MARKUP,
         "price_taxes_eur_kwh": TAXES,
         "price_timeout_s": 5}
    o.update(over)
    return o


def hours(n, start, step_h=1):
    return [start + timedelta(hours=i * step_h) for i in range(n)]


# --------------------------------------------------------------------------- #
print("0. the arithmetic this file grades with must not be degenerate")
# ⚠️ These four ARE the guard against a check that passes for the wrong reason.
# If any of them goes red, every markup assertion below is measuring nothing.
check("markup and taxes are both non-zero", MARKUP != 0 and TAXES != 0)
check("...and differ, so 'added the wrong one' is visible",
      MARKUP != TAXES)
check("...and markup twice != markup plus taxes, so double-counting the markup "
      "is distinguishable from correctly adding both",
      abs(2 * MARKUP - (MARKUP + TAXES)) > 1e-9)
check("...and the shipped defaults are NOT what this file grades with, because "
      f"config.yaml's {PRICE_OPTION_DEFAULTS['price_taxes_eur_kwh']} is exactly "
      f"twice its {PRICE_OPTION_DEFAULTS['price_markup_eur_kwh']}",
      abs(PRICE_OPTION_DEFAULTS["price_taxes_eur_kwh"]
          - 2 * PRICE_OPTION_DEFAULTS["price_markup_eur_kwh"]) < 1e-9
      and (MARKUP, TAXES) != (PRICE_OPTION_DEFAULTS["price_markup_eur_kwh"],
                              PRICE_OPTION_DEFAULTS["price_taxes_eur_kwh"]))


# --------------------------------------------------------------------------- #
print()
print("1. the shipped defaults live in config.yaml, and here only as a mirror")
# ⚠️ CFG-03. The add-on already carries option defaults in three places. This
# ticket adds ONE in-code mirror (PRICE_OPTION_DEFAULTS) and pins it to
# config.yaml by reading the file rather than by transcribing it - the same
# move test_validate.py makes for max_w and meter_freeze_s. If they drift, this
# goes red and somebody has to decide which is right.
_CFG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")
with open(_CFG, "r", encoding="utf-8") as fh:
    _cfg_lines = fh.read().splitlines()


def shipped(key):
    """The value config.yaml ships, from the `options:` block, parsed by hand.

    No yaml module in the shipped image, and this is two lines of parsing
    against a file whose shape is known - the alternative is a transcription,
    which is the thing being guarded against.
    """
    in_schema = False
    for line in _cfg_lines:
        if line.startswith("schema:"):
            in_schema = True
        if in_schema or not line.startswith(f"  {key}:"):
            continue
        raw = line.split(":", 1)[1].split("#", 1)[0].strip().strip('"').strip("'")
        try:
            return int(raw)
        except ValueError:
            pass
        try:
            return float(raw)
        except ValueError:
            return raw
    return KeyError


_found = {k: shipped(k) for k in PRICE_OPTION_DEFAULTS}
_missing = sorted(k for k, v in _found.items() if v is KeyError)
check(f"config.yaml ships every price option ({len(PRICE_OPTION_DEFAULTS)} keys)",
      _missing == [])
_drift = sorted(k for k, v in _found.items()
                if v is not KeyError and v != PRICE_OPTION_DEFAULTS[k])
check(f"...and each shipped default equals the in-code mirror {_found}",
      _drift == [])
# The schema block must list them too, or the Supervisor drops the option and
# the add-on silently runs on the mirror's value instead of the operator's.
_schema_block = _cfg_lines[_cfg_lines.index("schema:"):]
_unschemad = sorted(k for k in PRICE_OPTION_DEFAULTS
                    if not any(ln.startswith(f"  {k}:") for ln in _schema_block))
check("...and every one of them is in the schema block too", _unschemad == [])
# ⚠️ Derived from the registry, not typed: the schema's provider list must offer
# exactly the plugins that exist, or the UI and the code disagree about what is
# installable.
_line = next(ln for ln in _schema_block if ln.startswith("  price_provider:"))
_schema_ids = set(_line.split("list(", 1)[1].rstrip(")").split("|"))
check(f"the schema offers exactly the registered plugins plus 'off' "
      f"{sorted(_schema_ids)}",
      _schema_ids == set(PROVIDERS) | {"off"})


# --------------------------------------------------------------------------- #
print()
print("2. every shipped plugin emits the exact canonical schema (AC1)")

# One recorded fixture per plugin, none of them a real fetch. Prices are
# deliberately distinct, non-zero, and one of them NEGATIVE - §17 keeps a
# negative-price regression case, and a windy Sunday is the best charging hour
# of the year, so a framework that clamped it away would cost real money.
T0 = datetime(2026, 1, 15, 0, 0, tzinfo=UTC)
# ⚠️ THE NEGATIVE HOUR IS -95 EUR/MWh, NOT -12.5, AND THE MAGNITUDE IS THE
# POINT. All-in adds 0.088 EUR/kWh here, so a -12.5 EUR/MWh hour comes out at
# +0.0755 and the "a negative price survives" check below would be grading a
# positive number - passing whether or not the framework clamped anything.
# -95 EUR/MWh is both realistic for a windy Belgian Sunday and still negative
# after markup and taxes, which is the only version of the stimulus that grades
# what it claims to.
EPEX_EUR_MWH = [61.4, -95.0, 183.0]         # EUR/MWh, one deeply negative
RETAIL_EUR_KWH = [0.1341, 0.0872, 0.2610]   # EUR/kWh, markup already inside
EDDY_EUR_MWH = [141.0, 96.2, 268.4]         # EUR/MWh, all-in


def json_feed(key, values, provisional_at=1):
    """A feed body. `provisional_at` makes ONE point provisional and the rest
    not, so `plan_is_provisional` is graded against a mixed series - an
    all-true or all-false fixture would return the same answer under any bug."""
    return json.dumps({"data": [
        {"start": (T0 + timedelta(hours=i)).isoformat(),
         key: v,
         "provisional": (i == provisional_at)}
        for i, v in enumerate(values)]}).encode()


CSV_BODY = ("start,eur_per_kwh,provisional\n"
            "2026-01-15T00:00:00+01:00,0.1834,false\n"
            "2026-01-15T01:00:00+01:00,-0.0210,true\n"
            "2026-01-15T02:00:00+01:00,0.3115,false\n")

# The fixture, the transport that serves it, and - crucially - the ENERGY
# component of each point, computed here so the all-in expectation below is
# arithmetic this file did itself.
FIXTURES = {
    "generic_dayahead": (json_feed("price_eur_mwh", EPEX_EUR_MWH),
                         [v / 1000 for v in EPEX_EUR_MWH]),
    "engie_dynamic": (json_feed("price_eur_kwh", RETAIL_EUR_KWH),
                      list(RETAIL_EUR_KWH)),
    "totalenergies_dynamic": (json_feed("price_eur_kwh", RETAIL_EUR_KWH),
                              list(RETAIL_EUR_KWH)),
    "eneco_eddy": (json_feed("price_eur_mwh", EDDY_EUR_MWH),
                   [v / 1000 for v in EDDY_EUR_MWH]),
    "manual_csv": (None, [0.1834, -0.0210, 0.3115]),
}

check(f"all five plugins the ticket lists are registered {sorted(PROVIDERS)}",
      set(PROVIDERS) == {"generic_dayahead", "engie_dynamic",
                         "totalenergies_dynamic", "eneco_eddy", "manual_csv"})
check("...and the fixture set covers every one of them",
      set(FIXTURES) == set(PROVIDERS))


def canned(body):
    """A transport that serves recorded bytes and records that it was used."""
    calls = []

    async def t(url, timeout_s):
        calls.append((url, timeout_s))
        return body
    t.calls = calls
    return t


def load(pid, extra=None, transport=None):
    o = opts(price_provider=pid, **(extra or {}))
    return run(load_prices(o, T0, T0 + timedelta(hours=48),
                           transport=transport))


import tempfile                                       # noqa: E402

_csv_dir = tempfile.mkdtemp(prefix="tariff02-")
CSV_PATH = os.path.join(_csv_dir, "prices.csv")
with open(CSV_PATH, "w", encoding="utf-8") as fh:
    fh.write(CSV_BODY)


def load_any(pid):
    body, _ = FIXTURES[pid]
    if pid == "manual_csv":
        return load(pid, {"price_csv_path": CSV_PATH}, transport=refuse_transport)
    return load(pid, transport=canned(body))


async def _refuse(url, timeout_s):
    raise AssertionError(f"the offline plugin reached the network: {url}")

refuse_transport = _refuse

READS = {pid: load_any(pid) for pid in PROVIDERS}

_wrong_kind = sorted(p for p, r in READS.items() if r.kind != "series")
check(f"every plugin returned a series from its fixture", _wrong_kind == [])

# THE SCHEMA. Exactly three keys, in every point of every plugin, and the check
# derives the key set from the objects rather than asserting a typed literal.
_keys = derived(lambda: {frozenset(p.keys())
                         for r in READS.values()
                         for p in r.canonical()["prices"]})
check(f"every canonical point has exactly the three §9.3 keys, and only those "
      f"-> {_keys}",
      _keys == {frozenset({"start", "eur_per_kwh", "provisional"})})
check("the envelope is exactly {'prices': [...]}",
      derived(lambda: all(set(r.canonical()) == {"prices"}
                          for r in READS.values())) is True)
_badtype = derived(lambda: sorted(
    p for p, r in READS.items()
    if not all(isinstance(pt["eur_per_kwh"], float)
               and isinstance(pt["provisional"], bool)
               and isinstance(pt["start"], str)
               for pt in r.canonical()["prices"])))
check(f"...with a float price, a bool flag and an ISO string start {_badtype}",
      _badtype == [])
_naive = derived(lambda: sorted(p for p, r in READS.items()
                                if any(pt.start.utcoffset() is None
                                       for pt in r.points)))
check(f"every start is timezone-aware (§9.3 shows +01:00) {_naive}",
      _naive == [])
_offsets = derived(lambda: {pt.start.utcoffset()
                            for r in READS.values() for pt in r.points})
check(f"...and at a Belgian winter offset for a January fixture {_offsets}",
      _offsets == {timedelta(hours=1)})
check("the whole envelope round-trips through json.dumps",
      derived(lambda: all(json.loads(json.dumps(r.canonical())) == r.canonical()
                          for r in READS.values())) is True)


# --------------------------------------------------------------------------- #
print()
print("3. all-in: energy + markup + taxes, added exactly once (AC1, trap 1)")
# The expected figure is computed HERE, from the fixture's energy component and
# the plugin's own declaration, by different arithmetic than assemble() uses.
for pid, read in sorted(READS.items()):
    caps = PROVIDERS[pid].capabilities()
    _, energy = FIXTURES[pid]
    add = (MARKUP if Component.MARKUP in caps.adds else 0.0) + \
          (TAXES if Component.TAXES in caps.adds else 0.0)
    want = [round(e + add, 6) for e in energy]
    got = derived(lambda r=read: [p.eur_per_kwh for p in r.points])
    check(f"{pid}: all-in == published {sorted(c.value for c in caps.sources)} "
          f"+ framework {sorted(c.value for c in caps.adds) or 'nothing'} "
          f"-> {got}",
          got == want)

# ⚠️ THE ONE THAT MATTERS. generic_dayahead is the only plugin whose feed is
# wholesale, so it is the only one that gets both components - and the three
# retail plugins must each get strictly less. Derived by comparing the two
# plugins' output on the SAME energy figure rather than by restating numbers.
def added(pid):
    return derived(lambda: round(READS[pid].points[0].eur_per_kwh
                                 - FIXTURES[pid][1][0], 6))


_generic_add, _engie_add, _eddy_add = (added("generic_dayahead"),
                                       added("engie_dynamic"),
                                       added("eneco_eddy"))
check(f"the wholesale feed gets markup+taxes ({_generic_add})",
      isinstance(_generic_add, float)
      and abs(_generic_add - (MARKUP + TAXES)) < 1e-9)
check(f"a retail feed that already carries the markup gets taxes only "
      f"({_engie_add}), one markup less than the wholesale one",
      isinstance(_engie_add, float) and isinstance(_generic_add, float)
      and abs(_engie_add - TAXES) < 1e-9
      and abs((_generic_add - _engie_add) - MARKUP) < 1e-9)
check(f"an all-in feed gets nothing added ({_eddy_add})",
      isinstance(_eddy_add, float) and abs(_eddy_add) < 1e-9)

# Negative prices survive. A framework that clamped them would turn the best
# charging hour of the year into an average one, silently.
_neg = derived(lambda: [p.eur_per_kwh for p in READS["generic_dayahead"].points
                        if p.eur_per_kwh < 0])
check(f"a negative day-ahead price passes through, all-in {_neg}",
      _neg == [round(EPEX_EUR_MWH[1] / 1000 + MARKUP + TAXES, 6)])

# provisional is per point and comes from the feed, not from the series.
for pid, read in sorted(READS.items()):
    flags = derived(lambda r=read: [p.provisional for p in r.points])
    check(f"{pid}: provisional is per point and mixed {flags}",
          isinstance(flags, list) and any(flags) and not all(flags))
    check(f"...so plan_is_provisional is derived from those points",
          derived(lambda r=read: plan_is_provisional(r)) is True)
# ...and the same function says False when nothing is provisional. Without this
# arm, "always True" would pass every check above.
_settled = PriceRead.series(
    [PricePoint(brussels(T0), 0.2, False)], PriceProduct.DAY_AHEAD, "x", T0)
check("a fully settled series is NOT provisional",
      plan_is_provisional(_settled) is False)


# --------------------------------------------------------------------------- #
print()
print("4. double-counting the markup is detectable, and unreachable (trap 1)")


class _Rogue(PriceProvider):
    """A plugin that adds the markup itself while declaring it did not.

    NOT registered - it exists only here, as the mutation the design is meant to
    survive. It is what a plugin author writes when they copy the wholesale
    plugin, hardcode their supplier's margin, and forget to update `sources`.
    """

    id = "_rogue_not_shipped"

    @classmethod
    def capabilities(cls):
        return ProviderCapabilities(product=PriceProduct.DAY_AHEAD,
                                    sources=frozenset({Component.ENERGY}))

    async def fetch(self, o, since, until, transport):
        # The margin the author hardcoded, because they could not read the
        # configured one - see the FrameworkOption check below.
        return [RawPoint(T0, RETAIL_EUR_KWH[0], False)]


_honest_caps = PROVIDERS["engie_dynamic"].capabilities()
_honest = assemble(_honest_caps, [RawPoint(T0, RETAIL_EUR_KWH[0], False)],
                   MARKUP, TAXES)[0].eur_per_kwh
_rogue_raw = run(_Rogue().fetch(PluginOptions({}), T0, T0, None))
_rogue = assemble(_Rogue.capabilities(), _rogue_raw, MARKUP, TAXES)[0].eur_per_kwh
_delta = derived(lambda: round(_rogue - _honest, 6))
check(f"a plugin that pre-adds the markup and declares it did not comes out "
      f"{_delta} high - exactly one markup, so the defect is DETECTABLE and "
      f"attributable rather than merely wrong",
      isinstance(_delta, float) and abs(_delta - MARKUP) < 1e-9)
check("...and it is not confusable with 'the taxes were added twice'",
      isinstance(_delta, float) and abs(_delta - TAXES) > 1e-9)

# ...and now the structural half: the number is not reachable from a plugin, so
# the honest author cannot write that line even by accident.
_po = PluginOptions({"price_markup_eur_kwh": MARKUP,
                     "price_taxes_eur_kwh": TAXES,
                     "export_price_eur_kwh": 0.04,
                     "price_csv_path": "/x"})
_refused = []
for k in ("price_markup_eur_kwh", "price_taxes_eur_kwh", "export_price_eur_kwh"):
    try:
        _po.get(k)
    except FrameworkOption:
        _refused.append(k)
check(f"a plugin cannot read the framework's own options {sorted(_refused)}",
      sorted(_refused) == sorted(["price_markup_eur_kwh", "price_taxes_eur_kwh",
                                  "export_price_eur_kwh"]))
check("...and `k in opts` is false for them too, so a probe cannot find them",
      not any(k in _po for k in ("price_markup_eur_kwh", "price_taxes_eur_kwh")))
check("...while the options a plugin legitimately needs still come through",
      _po.get("price_csv_path") == "/x"
      and _po.get("price_timeout_s") == PRICE_OPTION_DEFAULTS["price_timeout_s"])

# The prose and the declaration must agree. A docstring that says "the framework
# adds taxes" over a class declaring it sources them is a 3 c/kWh error with a
# reassuring comment on top.
_doc_drift = []
for pid, cls in sorted(PROVIDERS.items()):
    line = next((ln for ln in (cls.__doc__ or "").splitlines()
                 if ln.strip().startswith("SOURCES:")), "")
    named = {w.strip() for w in line.split(":", 1)[-1].split(",") if w.strip()}
    declared = {c.value for c in cls.capabilities().sources}
    if named != declared:
        _doc_drift.append((pid, sorted(named), sorted(declared)))
check(f"every plugin's docstring names exactly the components it declares "
      f"({len(PROVIDERS)} checked)", _doc_drift == [])

# The ignored-markup warning: a plugin whose feed is already all-in must SAY it
# is ignoring the operator's numbers rather than dropping them silently.
_buf = io.StringIO()
_h = logging.StreamHandler(_buf)
logging.getLogger("goodwe.price").addHandler(_h)
load_any("eneco_eddy")
load_any("generic_dayahead")
logging.getLogger("goodwe.price").removeHandler(_h)
_warned = _buf.getvalue()
check("an all-in plugin logs that the configured markup/taxes are ignored",
      "double-count" in _warned and "eneco_eddy" in _warned)
check("...and the wholesale plugin, which really does use them, does not",
      "generic_dayahead" not in _warned)


# --------------------------------------------------------------------------- #
print()
print("5. product type is declared, not inferred from the name (AC2, trap 2)")
_products = {pid: cls.capabilities().product for pid, cls in PROVIDERS.items()}
check(f"eneco_eddy declares itself real-time/imbalance per §9.2 "
      f"({_products['eneco_eddy'].value})",
      _products["eneco_eddy"] is PriceProduct.REALTIME)
check("...and it is the only shipped one that is",
      [p for p, v in _products.items() if v is PriceProduct.REALTIME]
      == ["eneco_eddy"])
_regimes = {pid: volatility_regime(v) for pid, v in _products.items()}
check(f"...so §8.3's widened thresholds reach it through the ENUM {_regimes}",
      _regimes["eneco_eddy"] == "wide"
      and set(v for p, v in _regimes.items() if p != "eneco_eddy") == {"normal"})
check("the regime travels on the series a consumer is handed",
      derived(lambda: (READS["eneco_eddy"].regime,
                       READS["generic_dayahead"].regime)) == ("wide", "normal"))
def _accepts_name():
    """Would volatility_regime() take a plugin id? If it would, somebody can
    start passing one and AC2 rots without a single check going red."""
    try:
        volatility_regime("eneco_eddy")
        return True
    except TypeError:
        return False


check("volatility_regime refuses a plugin NAME, which is the AC2 violation "
      "written out",
      not _accepts_name())


# ⚠️ THE MECHANICAL FORM OF AC2. The framework's own code must contain no
# plugin id at all - if it does, the sixth supplier is an edit to price.py's
# shared machinery rather than a new class. Derived: build the set of ids that
# actually occur, and assert it is empty.
from app import price as _price                                   # noqa: E402

_framework = [_price.assemble, _price.load_prices, _price.volatility_regime,
              _price.arbitrage_allowed, _price.plan_is_provisional,
              _price.check_egress, _price._validate_raw, _price.PriceRead,
              _price._Series, _price._NoData, _price.PluginOptions,
              _price.register, _price.price_option]
_src = "\n".join(inspect.getsource(o) for o in _framework)
_leaked = sorted(pid for pid in PROVIDERS if pid in _src)
check(f"no plugin id appears anywhere in the framework's code {_leaked or '[]'}",
      _leaked == [])

# ...and the positive half: a plugin defined HERE, in a file price.py has never
# heard of, gets the right regime with no edit anywhere.
@register
class _NewSupplier(PriceProvider):
    """A sixth supplier, invented in the test file. AC2 in one class.

    SOURCES: energy
    The framework ADDS: markup, taxes
    """

    id = "_test_sixth_supplier"

    @classmethod
    def capabilities(cls):
        return ProviderCapabilities(product=PriceProduct.REALTIME,
                                    sources=frozenset({Component.ENERGY}),
                                    egress_hosts=("api.example.invalid",))

    async def fetch(self, o, since, until, transport):
        body = await transport("https://api.example.invalid/p",
                               float(o.get("price_timeout_s")))
        return _price._points_from_json(body, "price_eur_mwh", 0.001, False)


_sixth = load("_test_sixth_supplier",
              transport=canned(json_feed("price_eur_mwh", EPEX_EUR_MWH)))
check("a plugin invented outside price.py resolves through the registry",
      _sixth.kind == "series" and _sixth.provider == "_test_sixth_supplier")
_sixth_regime = derived(lambda: _sixth.regime)
check(f"...and lands in the wide-threshold regime purely from its declaration "
      f"({_sixth_regime})", _sixth_regime == "wide")
check("...and the framework applied its markup and taxes correctly too",
      derived(lambda: [p.eur_per_kwh for p in _sixth.points])
      == [round(v / 1000 + MARKUP + TAXES, 6) for v in EPEX_EUR_MWH])
del PROVIDERS["_test_sixth_supplier"]         # do not leak into later sections


# --------------------------------------------------------------------------- #
print()
print("6. no price data is its own outcome, not an empty list (trap 3, §11.4)")

_none = PriceRead.no_data("the feed timed out")
# ⚠️ THE DISTINCTION THE WHOLE TYPE EXISTS FOR. A day of free electricity and a
# day with no price data are different instructions: arbitrage is POINTLESS
# under the first and FORBIDDEN under the second.
_zero = PriceRead.series([PricePoint(brussels(T0 + timedelta(hours=i)), 0.0,
                                     False) for i in range(3)],
                         PriceProduct.DAY_AHEAD, "x", T0)
check("a zero-price series and no-price-data are different kinds",
      _zero.kind != _none.kind)
check("...arbitrage is allowed under zero prices and forbidden under no data",
      arbitrage_allowed(_zero) is True and arbitrage_allowed(_none) is False)
check("...the plan is provisional under no data and settled under zero prices",
      plan_is_provisional(_none) is True and plan_is_provisional(_zero) is False)
check("...and the cheapest hour is computable for the zero series",
      derived(lambda: min(p.eur_per_kwh for p in _zero.points)) == 0.0)


def raises(fn, exc):
    try:
        fn()
    except exc:
        return True
    except Exception:
        return False
    return False


check("no_data has no .points, so `min(read.points)` cannot invent a free hour",
      raises(lambda: _none.points, Absence))
check("...no .product to switch a threshold regime on",
      raises(lambda: _none.product, Absence))
check("...no .at(), so an hourly lookup cannot silently succeed",
      raises(lambda: _none.at(T0), Absence))
check("...and .get('points', []) - the idiom the whole rework exists to kill - "
      "raises rather than returning the default",
      raises(lambda: _none.get("points", []), Absence))
check("...and it is not iterable, so `for p in read:` cannot be an empty loop",
      raises(lambda: list(_none), Absence))
check("...while its one real field, the reason, is there for an operator",
      _none.reason == "the feed timed out")

check("an EMPTY series cannot be constructed at all, so 'no prices' can never "
      "wear the shape of an answer",
      raises(lambda: PriceRead.series([], PriceProduct.DAY_AHEAD, "x", T0),
             ValueError))
check("a series built with a plugin NAME instead of a product enum is refused",
      raises(lambda: PriceRead.series([PricePoint(brussels(T0), 0.2, False)],
                                      "eneco_eddy", "x", T0), ValueError))
check("no_data with a blank reason is refused - the reason is its only field",
      raises(lambda: PriceRead.no_data(""), ValueError))


def _forgot():
    return _none.match(series=lambda r: r.points)          # no_data omitted


check("a caller who forgets the no_data arm gets a TypeError at the call site, "
      "not a fall-through", raises(_forgot, TypeError))

# .at() outside the series raises rather than returning a neighbour or None.
_s = READS["generic_dayahead"]
check("series.at() inside the series returns the covering point",
      derived(lambda: _s.at(_s.points[0].start + timedelta(minutes=59))
              is _s.points[0]) is True)
check("...and raises outside it rather than returning None to be `or`-ed away",
      derived(lambda: (
          raises(lambda: _s.at(_s.points[-1].start + timedelta(days=2)),
                 Absence)
          and _s.covers(_s.points[0].start)
          and not _s.covers(_s.points[-1].start + timedelta(days=2)))) is True)

# Every failure a site can have must come back as no_data, never as a traceback
# into the tick. Derived: drive each failure and collect the kinds.
async def _boom(url, timeout_s):
    raise OSError("connection reset by peer")


_failures = {
    "the source is unreachable": load("generic_dayahead", transport=_boom),
    "the body is not JSON": load("generic_dayahead", transport=canned(b"<html>")),
    "the feed is empty": load("generic_dayahead",
                              transport=canned(b'{"data": []}')),
    "the provider id is unknown": load("no_such_supplier"),
    "no provider is configured": load("off"),
    "the csv is missing": load("manual_csv",
                               {"price_csv_path": "/no/such/file.csv"},
                               transport=refuse_transport),
}
_not_nodata = sorted(k for k, r in _failures.items() if r.kind != "no_data")
check(f"every site-level failure returns no_data instead of raising "
      f"({len(_failures)} cases)", _not_nodata == [])
_no_reason = derived(lambda: sorted(k for k, r in _failures.items()
                                    if len(r.reason) < 10))
check(f"...each with a reason an operator can act on {_no_reason}",
      _no_reason == [])
check("...and none of them is mistakable for a series",
      derived(lambda: all(not arbitrage_allowed(r)
                          for r in _failures.values())) is True)

# ⚠️ The one thing that must NOT degrade to no_data: a plugin reaching for the
# markup. That is a code defect, and swallowing it would hide exactly what
# trap 1 exists to catch.
class _Thief(PriceProvider):
    id = "_thief"

    @classmethod
    def capabilities(cls):
        return ProviderCapabilities(product=PriceProduct.DAY_AHEAD,
                                    sources=frozenset({Component.ENERGY}))

    async def fetch(self, o, since, until, transport):
        return [RawPoint(T0, 0.1 + o.get("price_markup_eur_kwh"), False)]


PROVIDERS["_thief"] = _Thief
check("a plugin that reaches for the markup raises rather than degrading to "
      "no_data - that is a bug, not a site condition",
      raises(lambda: load("_thief"), FrameworkOption))
del PROVIDERS["_thief"]


# --------------------------------------------------------------------------- #
print()
print("7. manual_csv works fully offline (AC3)")
_had_aiohttp = "aiohttp" in sys.modules
_csv_read = load("manual_csv", {"price_csv_path": CSV_PATH},
                 transport=refuse_transport)
_csv_prices = derived(lambda: [p.eur_per_kwh for p in _csv_read.points])
check("the offline plugin produced a series without touching the transport - "
      "the transport raises if called at all",
      _csv_read.kind == "series" and _csv_prices == [0.1834, -0.021, 0.3115])
check(f"...it is all-in already, so the framework added nothing {_csv_prices}",
      _csv_prices == [round(v, 6) for v in
                      [float(ln.split(",")[1]) for ln in
                       CSV_BODY.strip().splitlines()[1:]]])
check("...including the negative hour",
      isinstance(_csv_prices, list) and any(v < 0 for v in _csv_prices))
check("...and the per-point provisional flag came from the file's third column",
      derived(lambda: [p.provisional for p in _csv_read.points])
      == [ln.split(",")[2] == "true"
          for ln in CSV_BODY.strip().splitlines()[1:]])
check("...and no HTTP stack was imported to do it"
      if not _had_aiohttp else
      "...(aiohttp was already imported before this file ran; not a finding)",
      _had_aiohttp or "aiohttp" not in sys.modules)
check("manual_csv declares no egress hosts at all",
      PROVIDERS["manual_csv"].capabilities().egress_hosts == ())

_ManualCsv = PROVIDERS["manual_csv"]
check("a CSV with no header is refused rather than parsed as data",
      raises(lambda: _ManualCsv.parse("2026-01-15T00:00:00+01:00,0.18\n"),
             SourceUnavailable))
check("a CSV start with no offset is refused, naming the fix",
      raises(lambda: _ManualCsv.parse("start,eur_per_kwh\n2026-01-15 00:00,0.18\n"),
             SourceUnavailable))
check("...and the two-column form (no provisional) is accepted, defaulting to "
      "settled",
      [p.provisional for p in
       _ManualCsv.parse("start,eur_per_kwh\n2026-01-15T00:00:00+01:00,0.18\n")]
      == [False])


# --------------------------------------------------------------------------- #
print()
print("8. the two 2026 DST transitions, where a 24-slot array is wrong")
# ⚠️ RUN ON THE 23- AND 25-HOUR DAYS ON PURPOSE. On any normal day a fixed
# 24-slot implementation, a naive-local-hour key and a correct one all agree
# exactly, so a check on 15 January grades nothing about DST.
check("the 2026 transitions are where this repo expects them (last Sundays)",
      _price._last_sunday(2026, 3) == 29 and _price._last_sunday(2026, 10) == 25)


def local_day(day_start_utc, n):
    return [brussels(day_start_utc + timedelta(hours=i)) for i in range(n)]


# Spring forward: local day 2026-03-29 runs 2026-03-28T23:00Z .. 2026-03-29T22:00Z
_spring = local_day(datetime(2026, 3, 28, 23, tzinfo=UTC), 23)
_spring_end = brussels(datetime(2026, 3, 29, 22, tzinfo=UTC))
check(f"the spring day has 23 hours, ending exactly at the next midnight "
      f"({_spring[0].isoformat()} .. {_spring_end.isoformat()})",
      _spring[0].hour == 0 and _spring[0].day == 29
      and _spring_end.hour == 0 and _spring_end.day == 30)
check(f"...crossing exactly one offset change "
      f"{sorted(str(o) for o in {p.utcoffset() for p in _spring})}",
      {p.utcoffset() for p in _spring} == {timedelta(hours=1),
                                           timedelta(hours=2)})
check("...and 02:00 local never happens that day, which a 24-slot array would "
      "have invented", 2 not in {p.hour for p in _spring})

# Fall back: local day 2026-10-25 runs 2026-10-24T22:00Z .. 2026-10-25T22:00Z
_fall = local_day(datetime(2026, 10, 24, 22, tzinfo=UTC), 25)
_fall_end = brussels(datetime(2026, 10, 25, 23, tzinfo=UTC))
check(f"the autumn day has 25 hours ({_fall[0].isoformat()} .. "
      f"{_fall_end.isoformat()})",
      _fall[0].hour == 0 and _fall[0].day == 25
      and _fall_end.hour == 0 and _fall_end.day == 26)
_wall = [p.replace(tzinfo=None) for p in _fall]
_dups = len(_wall) - len(set(_wall))
check(f"...and exactly one local wall-clock hour occurs TWICE ({_dups}), which "
      f"a naive-local-hour key would collide into one mispriced hour",
      _dups == 1)
check("...distinguished by their offsets, which is why the canonical format "
      "carries one",
      len({p.isoformat() for p in _fall}) == 25)
# A normal day, stated as the control: without it, "23" and "25" prove nothing
# about whether the code counts at all.
_normal = local_day(datetime(2026, 1, 14, 23, tzinfo=UTC), 24)
check("...while an ordinary January day has 24 hours and one offset",
      len({p.replace(tzinfo=None) for p in _normal}) == 24
      and len({p.utcoffset() for p in _normal}) == 1)
check("brussels() refuses a naive datetime rather than assuming an offset",
      raises(lambda: brussels(datetime(2026, 1, 15, 0, 0)), ValueError))
check("the offset boundary is 01:00 UTC, not local midnight",
      brussels_offset(datetime(2026, 3, 29, 0, 59, tzinfo=UTC)) ==
      timezone(timedelta(hours=1))
      and brussels_offset(datetime(2026, 3, 29, 1, 0, tzinfo=UTC)) ==
      timezone(timedelta(hours=2)))

# A feed that publishes the duplicated hour must survive it. The 25-hour day is
# 25 strictly increasing instants; a plugin keyed on local hours would emit 24
# and be rejected by the boundary validator instead of silently mispricing.
_dst_feed = json.dumps({"data": [
    {"start": (datetime(2026, 10, 24, 22, tzinfo=UTC)
               + timedelta(hours=i)).isoformat(),
     "price_eur_mwh": 50.0 + i, "provisional": i == 24}
    for i in range(25)]}).encode()
_dst_read = load("generic_dayahead", transport=canned(_dst_feed))
_dst_n = derived(lambda: len(_dst_read.points))
check(f"a 25-point autumn feed is accepted whole ({_dst_n} points)",
      _dst_read.kind == "series" and _dst_n == 25)
check("...and its canonical starts are 25 distinct strings",
      derived(lambda: len({p["start"]
                           for p in _dst_read.canonical()["prices"]})) == 25)


# --------------------------------------------------------------------------- #
print()
print("9. the boundary rejects what cannot be a price, and only that")
_caps = PROVIDERS["generic_dayahead"].capabilities()


def boundary(points):
    return raises(lambda: _price._validate_raw(points, _caps), SourceUnavailable)


check("NaN is refused", boundary([RawPoint(T0, float("nan"), False)]))
check("infinity is refused", boundary([RawPoint(T0, float("inf"), False)]))
check(f"a EUR/MWh figure read as EUR/kWh is refused (>{MAX_PLAUSIBLE_EUR_KWH})",
      boundary([RawPoint(T0, 61.4, False)]))
check("a naive start is refused",
      boundary([RawPoint(datetime(2026, 1, 15), 0.1, False)]))
check("duplicate starts are refused - the DST collision, caught",
      boundary([RawPoint(T0, 0.1, False), RawPoint(T0, 0.2, False)]))
check("out-of-order starts are refused",
      boundary([RawPoint(T0 + timedelta(hours=1), 0.1, False),
                RawPoint(T0, 0.2, False)]))
check("a plugin returning something that is not a RawPoint is refused",
      boundary([PricePoint(brussels(T0), 0.1, False)]))
check("an empty list is refused", boundary([]))
# ...and the negative-price control, so the rules above are not just "reject
# everything". A validator that refused this would cost money on a windy Sunday.
check("a NEGATIVE price at a plausible magnitude is accepted",
      not boundary([RawPoint(T0, -0.35, False)]))
check("...and so is a large-but-plausible one at the limit",
      not boundary([RawPoint(T0, MAX_PLAUSIBLE_EUR_KWH - 0.01, False)]))


# --------------------------------------------------------------------------- #
print()
print("10. egress is declared by the plugin and enforced (§16)")
_g = PROVIDERS["generic_dayahead"].capabilities()
check("a declared https host is allowed",
      check_egress("https://api.energy-charts.info/price?bzn=BE", _g)
      == "api.energy-charts.info")
check("an undeclared host is refused even over TLS",
      raises(lambda: check_egress("https://evil.example/p", _g),
             SourceUnavailable))
check("plain http is refused",
      raises(lambda: check_egress("http://api.energy-charts.info/p", _g),
             SourceUnavailable))
check("a userinfo prefix cannot smuggle a declared host past the check",
      raises(lambda: check_egress(
          "https://api.energy-charts.info@evil.example/p", _g),
          SourceUnavailable))
check("an operator-supplied URL to another supplier's host is refused",
      load("generic_dayahead",
           {"price_feed_url": "https://api.eneco.be/eddy/prices"},
           transport=canned(json_feed("price_eur_mwh", EPEX_EUR_MWH))
           ).kind == "no_data")
_declared = {pid: cls.capabilities().egress_hosts
             for pid, cls in PROVIDERS.items()}
check(f"every networked plugin declares at least one host, and manual_csv "
      f"declares none {_declared}",
      all(bool(h) for p, h in _declared.items() if p != "manual_csv")
      and _declared["manual_csv"] == ())
check("the timeout the plugin passes to the transport is the configured one, "
      "not a hardcoded default",
      (lambda t: (load("generic_dayahead", {"price_timeout_s": 7},
                       transport=t), t.calls[0][1] == 7.0)[1])(
          canned(json_feed("price_eur_mwh", EPEX_EUR_MWH))))


# --------------------------------------------------------------------------- #
print()
print("11. injection price is a separate scalar (AC4, §9.3)")
check(f"the shipped default is config.yaml's "
      f"{PRICE_OPTION_DEFAULTS['export_price_eur_kwh']}",
      export_price_eur_kwh({}) == PRICE_OPTION_DEFAULTS["export_price_eur_kwh"])
check("an operator's value wins over it",
      export_price_eur_kwh({"export_price_eur_kwh": 0.0}) == 0.0)
check("...including a negative one - Belgian injection has gone below zero",
      export_price_eur_kwh({"export_price_eur_kwh": -0.02}) == -0.02)
check("it is NOT folded into the consumption series any plugin emits",
      derived(lambda: all(export_price_eur_kwh({}) not in
                          [p.eur_per_kwh for p in r.points]
                          for r in READS.values())) is True)
check("...and no plugin can reach it to fold it in",
      raises(lambda: PluginOptions({}).get("export_price_eur_kwh"),
             FrameworkOption))


# --------------------------------------------------------------------------- #
print()
print("12. the startup gate (validate.py rule 7)")


def vbase(**over):
    o = {"setpoint_entity": "number.goodwe_setpoint", "max_w": 2000,
         "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
         "meter_source": "off",
         "maintenance_charge_w": 2000, "maintenance_discharge_w": 2000}
    o.update(over)
    return o


def why(problems, needle):
    return any(needle in p for p in problems)


check("the shipped price defaults start (price_provider is 'off')",
      validate_startup_config(vbase(**{k: v for k, v in
                                       PRICE_OPTION_DEFAULTS.items()}),
                              -5000, 5000) == [])
check("...and so does a configured provider on its own",
      validate_startup_config(vbase(price_provider="generic_dayahead"),
                              -5000, 5000) == [])
check("an unknown provider id is refused rather than defaulted to 'off'",
      why(validate_startup_config(vbase(price_provider="nordpool"), -5000, 5000),
          "not a shipped plugin"))
# ⚠️ THE HA OVERLAP, DECIDED. The two entities keep working alone; combining
# them with a provider is what is refused, because the entities win silently.
_p = validate_startup_config(
    vbase(price_provider="engie_dynamic", price_now_entity="sensor.now",
          price_avg_entity="sensor.avg"), -5000, 5000)
check("a provider AND the Home Assistant price entities is refused - two price "
      "sources on one site", len(_p) == 1)
check("...and the message says which one actually wins",
      why(_p, "HOME ASSISTANT ENTITIES WIN"))
check("the Home Assistant entities alone still start, so nothing is taken away "
      "from a live install",
      validate_startup_config(vbase(price_now_entity="sensor.now",
                                    price_avg_entity="sensor.avg"),
                              -5000, 5000) == [])
check("manual_csv with no path is refused - the offline fallback with no file "
      "is §11.4 reached by accident",
      why(validate_startup_config(vbase(price_provider="manual_csv",
                                        price_csv_path=""), -5000, 5000),
          "price_csv_path"))
check("...and with a path it starts",
      validate_startup_config(vbase(price_provider="manual_csv",
                                    price_csv_path=CSV_PATH),
                              -5000, 5000) == [])
# ⚠️ price_provider 'off' must NEVER become a refusal: §9.2 puts a large share
# of the installed base on a fixed or static tariff, and §11.4 makes no-price
# data a supported state.
check("'off' is not refused, on any of the other configurations",
      all(validate_startup_config(vbase(price_provider="off", **extra),
                                  -5000, 5000) == []
          for extra in ({}, {"price_now_entity": "sensor.now",
                             "price_avg_entity": "sensor.avg"},
                        {"price_csv_path": ""})))


# --------------------------------------------------------------------------- #
print()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")