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 ast
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, BANNED_IN_PLUGIN_CODE, Component, FrameworkOption,
    MAX_PLAUSIBLE_EUR_KWH, code_leaks,
    PRICE_OPTION_DEFAULTS, PROVIDERS, PluginOptions, PricePoint, PriceProduct,
    PriceProvider, PriceRead, ProviderCapabilities, RawPoint, SourceUnavailable,
    arbitrage_allowed, assemble, brussels, brussels_offset, check_egress,
    price_option,
    export_price_eur_kwh, load_prices, plan_is_provisional, register,
    volatility_regime,
)
from app import price as _price
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)


def _reason(fn):
    """The message fn() raised, so a check can assert WHICH rule refused.

    `raises(...)` only says that something refused. M7 survived because two
    different rules could refuse the same URL and the check could not tell them
    apart - so where a check is named for a specific rule, it asserts on the
    reason instead.
    """
    try:
        fn()
    except Exception as e:                                    # noqa: BLE001
        return str(e)
    return ""


def raises(fn, exc):
    """True if fn() raises exactly `exc`. A DIFFERENT exception is False, not
    True - otherwise "it was refused" and "it broke on the way to being
    refused" would grade the same, which is how M7 hid."""
    try:
        fn()
    except exc:
        return True
    except Exception:
        return False
    return False


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("4b. a plugin cannot reach the markup BY ANY ROUTE (review of PR #16)")
# WHY THIS SECTION EXISTS. The first version of this ticket claimed the markup
# was "not in scope" for a plugin because `PluginOptions` refused it. Review
# wrote a double-counting plugin three more ways, one of them touching nothing
# but `PRICE_OPTION_DEFAULTS` - a public dict in the module every plugin must
# already import for `register`. One closed door in a room with three is not a
# guarantee. Honest wholesale baseline 0.188; each attack lands 0.219.
_banned = set(BANNED_IN_PLUGIN_CODE)
check(f"the banned set covers the framework keys AND the three public routes "
      f"review actually used {sorted(_banned)}",
      _banned == set(_price.FRAMEWORK_KEYS) | {"PRICE_OPTION_DEFAULTS",
                                               "price_option", "_o"})

# code_leaks() fails OPEN when inspect.getsource() cannot find the .py - the
# right trade for a battery controller, and a silently disarmed guard if nobody
# checks. THIS is the check that makes CI prove the guard is armed.
_srclen = derived(lambda: {pid: len(inspect.getsource(cls))
                           for pid, cls in PROVIDERS.items()})
check(f"the scan really read every shipped plugin's source, so its fail-open "
      f"path is not silently in use {_srclen}",
      isinstance(_srclen, dict) and len(_srclen) == 5
      and all(n > 200 for n in _srclen.values()))
_dirty = derived(lambda: {pid: code_leaks(cls)
                          for pid, cls in PROVIDERS.items()
                          if code_leaks(cls)})
check(f"...and all five shipped plugins are clean {_dirty}", _dirty == {})

# THE DOCSTRING STRIP IS LOAD-BEARING, AND THIS PROVES IT RATHER THAN ASSERTING
# IT. Every plugin is REQUIRED to name its sourced components in prose, so
# engie_dynamic's docstring contains "markup" - a raw text scan false-positives
# on the documentation the same file mandates, and a check that fires on
# required documentation is a check somebody deletes.
_raw_hits = derived(lambda: sorted(
    pid for pid, cls in PROVIDERS.items()
    if any(b in inspect.getsource(cls) for b in _banned)))
check(f"a RAW text scan would false-positive on the mandated docstrings "
      f"{_raw_hits}, which is why code_leaks() strips them before searching",
      isinstance(_raw_hits, list) and _raw_hits != [] and _dirty == {})


class _AttackPrivateDict(PriceProvider):
    """A1: reaches into PluginOptions' private dict.

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

    id = "_attack_private_dict"

    @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.10 + o._o["price_markup_eur_kwh"], False)]


class _AttackHelper(PriceProvider):
    """A2: calls a public framework helper with the raw options dict.

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

    id = "_attack_helper"

    @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.10 + price_option(o._o, "price_markup_eur_kwh"),
                         False)]


class _AttackPublicConstant(PriceProvider):
    """A2b: touches nothing but a public module constant. THE DECIDING ONE.

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

    id = "_attack_public_constant"

    @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.10 + PRICE_OPTION_DEFAULTS["price_markup_eur_kwh"],
                         False)]


_ATTACKS = {"A1 private dict": _AttackPrivateDict,
            "A2 public helper": _AttackHelper,
            "A2b public constant": _AttackPublicConstant}
for _name, _cls in _ATTACKS.items():
    _found = derived(lambda c=_cls: code_leaks(c))
    check(f"{_name}: the scan names what it reached for {_found}",
          isinstance(_found, list) and _found != [])
    check(f"...and register() REFUSES it at import, not at fetch time",
          raises(lambda c=_cls: register(c), FrameworkOption))
    check(f"...so it never enters the registry", _cls.id not in PROVIDERS)

# THE COST OF THE DEFECT, MEASURED, so the refusal is not grading itself.
_honest_wholesale = derived(lambda: assemble(
    PROVIDERS["generic_dayahead"].capabilities(),
    [RawPoint(T0, 0.10, False)], MARKUP, TAXES)[0].eur_per_kwh)
_attacked = derived(lambda: assemble(
    _AttackPublicConstant.capabilities(),
    [RawPoint(T0, 0.10 + MARKUP, False)], MARKUP, TAXES)[0].eur_per_kwh)
check(f"an unrefused attack would bill {_attacked} against an honest "
      f"{_honest_wholesale} - one markup high, every hour, silently",
      isinstance(_attacked, float) and isinstance(_honest_wholesale, float)
      and abs((_attacked - _honest_wholesale) - MARKUP) < 1e-9)

# WHAT THIS DOES *NOT* CLOSE, AND CANNOT. A plugin that declares {ENERGY} while
# its feed is really all-in comes out one markup high while touching no
# framework name at all. No code can check that: the declaration is the
# plugin's unverifiable claim about somebody else's tariff. Stated here so the
# guarantee is read at its true size - the complement is added exactly once,
# GIVEN A CORRECT DECLARATION - and so a reviewer who trusts an absolute
# guarantee does not stop checking. Measured, not asserted:
_misdeclared = derived(lambda: assemble(
    ProviderCapabilities(product=PriceProduct.DAY_AHEAD,
                         sources=frozenset({Component.ENERGY})),
    [RawPoint(T0, 0.10 + MARKUP + TAXES, False)], MARKUP, TAXES)[0].eur_per_kwh)
check(f"a MIS-DECLARED all-in feed still doubles ({_misdeclared} vs "
      f"{_honest_wholesale}) with no framework name touched - the limit of the "
      f"guarantee, stated rather than hidden",
      isinstance(_misdeclared, float)
      and abs((_misdeclared - _honest_wholesale) - (MARKUP + TAXES)) < 1e-9
      and code_leaks(_AttackPrivateDict) != [])


# --------------------------------------------------------------------------- #
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.
_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)


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")

# M3, INHERITED FROM RIG-01 AND CLOSED HERE. `_NoData.__init__(self,
# reason="no price data")` used to go green: a required field with a DEFAULT is
# the exact idiom this whole type exists to forbid, and it survived because
# every check passed a reason explicitly. It is the identical survivor left
# open on sim/riglock.py's `_Unread.cause` - the shape was borrowed including
# its blind spot, which is what borrowing a shape does if nobody re-runs the
# mutations against the copy.
#
# Checked by SIGNATURE rather than by one call, so any default added to any of
# these constructors later is red without somebody remembering to write a new
# case.
def _defaulted(fn):
    """Parameters of `fn` that carry a default. `self` is skipped."""
    try:
        ps = list(inspect.signature(fn).parameters.values())
    except (TypeError, ValueError) as e:                      # pragma: no cover
        return _Raised(e)
    return [q.name for q in ps
            if q.name != "self" and q.default is not inspect.Parameter.empty]


_ctors = {"PriceRead.no_data": PriceRead.no_data,
          "PriceRead.series": PriceRead.series,
          "_NoData.__init__": _price._NoData.__init__,
          "_Series.__init__": _price._Series.__init__,
          "PriceRead.match": PriceRead.match}
_with_defaults = derived(lambda: {n: d for n, d in
                                  ((n, _defaulted(f)) for n, f in _ctors.items())
                                  if d})
check(f"no constructor or handler of an outcome type carries a DEFAULT, so a "
      f"reading can never be built out of nothing {_with_defaults}",
      _with_defaults == {})
check("...so no_data() with no reason at all is a TypeError at the call site",
      raises(lambda: PriceRead.no_data(), TypeError))
check("...and the variant class itself cannot be built bare either",
      raises(lambda: _price._NoData(), TypeError))
check("...nor can a series be built without its product or provenance",
      raises(lambda: _price._Series([PricePoint(brussels(T0), 0.2, False)]),
             TypeError))
# The positive arm, so the four above are not passing because everything raises.
check("...while a fully specified no_data is still perfectly constructible",
      PriceRead.no_data("the feed timed out").reason == "the feed timed out")

# `match` must also refuse POSITIONAL handlers - keyword-only is what makes a
# forgotten arm a TypeError rather than a silently swapped one.
check("match() refuses positional handlers, so the two arms cannot be swapped",
      raises(lambda: _none.match(lambda r: 1, lambda r: 2), TypeError))

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")
# M5. Deleting the `Component.ENERGY not in sources` guard used to go green,
# and the consequence is not cosmetic: `adds` becomes {ENERGY, TAXES},
# `assemble()` never adds an energy component it thinks it is adding, and the
# emitted all-in price silently omits the energy it is supposed to be pricing.
check("capabilities() refuses a feed that declares no ENERGY component",
      raises(lambda: ProviderCapabilities(
          product=PriceProduct.DAY_AHEAD,
          sources=frozenset({Component.MARKUP})), ValueError))
check("...including a completely empty sources set",
      raises(lambda: ProviderCapabilities(product=PriceProduct.DAY_AHEAD,
                                          sources=frozenset()), ValueError))
check("...and an unknown component is refused rather than ignored",
      raises(lambda: ProviderCapabilities(
          product=PriceProduct.DAY_AHEAD,
          sources=frozenset({Component.ENERGY, "grid_fees"})), ValueError))
check("...and a product given as a string is refused at the declaration too",
      raises(lambda: ProviderCapabilities(product="dynamic_dayahead",
                                          sources=frozenset({Component.ENERGY})),
             ValueError))
# THE CONSEQUENCE, derived across every shipped plugin rather than asserted on
# one: the framework must never be in a position to "add" the energy component,
# because there is nothing to add it from.
_adds_energy = derived(lambda: sorted(
    pid for pid, cls in PROVIDERS.items()
    if Component.ENERGY in cls.capabilities().adds))
check(f"so no shipped plugin ever asks the framework to invent an energy "
      f"price {_adds_energy}", _adds_energy == [])

_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))
# M7, AND THE ASSERTION IS THE FIX, NOT ONLY THE PARSER. Deleting the TLS rule
# used to leave this check GREEN: the old parser stripped a fixed
# len("https://") from the front, so `http://api.energy-charts.info/p` became
# the host `pi.energy-charts.info`, which failed the ALLOW-LIST instead. The
# feed was refused, the check named "plain http is refused" passed, and the
# rule it was named for could be deleted with no consequence - an assertion
# that does not test what it names, inside a security check.
#
# Three checks now, and the first two are what make the third able to fail:
# prove the host is declared over https, prove it parses identically over
# http, and only then assert that http is refused AND that the reason is TLS.
_tls_url = "http://api.energy-charts.info/p"
check("the same host over https is allowed, so it IS on the allow-list",
      derived(lambda: check_egress("https://api.energy-charts.info/p", _g))
      == "api.energy-charts.info")
check("...and the http form parses to that same declared host, so nothing but "
      "the TLS rule can refuse it",
      derived(lambda: _tls_url.partition("://")[2].split("/", 1)[0])
      == "api.energy-charts.info")
_tls_reason = derived(lambda: _reason(lambda: check_egress(_tls_url, _g)))
check(f"plain http is refused, and refused BY THE TLS RULE - {_tls_reason!r}",
      isinstance(_tls_reason, str) and "non-TLS" in _tls_reason)
check("a scheme that is neither is refused the same way, by TLS not by host",
      "non-TLS" in str(derived(lambda: _reason(
          lambda: check_egress("ftp://api.energy-charts.info/p", _g)))))
# ⚠️ M7b, WHICH I FOUND IN MY OWN FIX FOR M7 AND WHICH IS THE SAME DEFECT ONE
# LAYER DOWN. This check used to read `raises(..., SourceUnavailable)` and
# survived deleting the `if not sep` guard: with no guard,
# "api.energy-charts.info/p".partition("://") leaves scheme = the WHOLE string,
# so the TLS rule refuses it and the check named for the scheme guard passes.
# Refused either way - but the diagnostic an operator gets degrades from "not an
# absolute URL" to "uses 'api.energy-charts.info/p'" as a scheme, which is
# gibberish. Assert the reason, like M7 above.
_noscheme = derived(lambda: _reason(
    lambda: check_egress("api.energy-charts.info/p", _g)))
check(f"a URL with no scheme is refused BY THE SCHEME GUARD, with a diagnostic "
      f"that names the real problem - {_noscheme!r}",
      isinstance(_noscheme, str) and "not an absolute URL" in _noscheme)
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"))
# TARIFF-03 added a second source with the same silent-loser shape.
_ps = validate_startup_config(
    vbase(static_profile='{"periods": [{"from":"00:00","to":"24:00",'
                         '"days":"all","eur_per_kwh":0.31}]}',
          price_now_entity="sensor.now", price_avg_entity="sensor.avg"),
    -5000, 5000)
check("a static_profile AND the Home Assistant price entities is refused too - "
      "main.py reads one of them for `cheap` and the loser is silent",
      len(_ps) == 1 and why(_ps, "HOME ASSISTANT"))
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) == [])
# ⚠️ THIS TRIPWIRE FIRED, AND THIS IS WHAT IT LOOKS LIKE AFTERWARDS.
#
# TARIFF-02 left a check here asserting that NOTHING in app/ consumed
# `load_prices()`, because rule 7's stated reason - "the supplier you configured
# would decide nothing" - was true only for as long as that held. The
# instruction attached to it was: when this goes red, update the message in
# validate.py in the same commit, do not delete the check.
#
# TARIFF-03 landed `app/static_tariff.py`, whose `resolve_tariff()` consumes
# `load_prices()` and turns a series into a tariff plan, a cheap-window flag and
# an arbitrage verdict. The check went red, naming that file, and validate.py's
# message was rewritten in the same commit to say what is true now instead. So
# the tripwire is KEPT and RE-AIMED rather than removed, at the next thing that
# can go quietly wrong:
#
#   * the consumer set is PINNED. A second consumer appearing - OPT-04 wiring a
#     strategy in, most likely - goes red again, and whoever does that has to
#     re-read rule 7's message against what their change makes true.
#   * the EXPIRED PHRASE is asserted GONE. If the old wording is ever restored
#     by a bad merge, this says so.
#   * and the message has to keep naming its consumer, so it cannot decay into
#     a refusal with no stated reason at all.
_APP = os.path.join(os.path.dirname(os.path.abspath(__file__)), "app")
# ⚠️ AST, NOT A SUBSTRING, AND THE FIRST DRAFT PROVED WHY. A text scan for
# "load_prices" matched the COMMENT in validate.py that explains this very
# tripwire, so the check went red against its own documentation - the same
# false-positive class as the plugin docstrings in section 4b, hit twice in one
# ticket. Comments and docstrings are not in an AST, so a name/import walk sees
# code and only code.
def _imports_or_calls(path, name):
    try:
        tree = ast.parse(open(path, encoding="utf-8").read())
    except (OSError, SyntaxError):
        return False
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and any(a.name == name
                                                    for a in node.names):
            return True
        if isinstance(node, ast.Name) and node.id == name:
            return True
        if isinstance(node, ast.Attribute) and node.attr == name:
            return True
    return False


_consumers = derived(lambda: sorted(
    f for f in os.listdir(_APP)
    if f.endswith(".py") and f != "price.py"
    and _imports_or_calls(os.path.join(_APP, f), "load_prices")))
_EXPECTED_CONSUMERS = ["static_tariff.py"]
check(f"load_prices() is consumed by exactly the modules rule 7's message "
      f"names {_consumers}. When a SECOND one appears this goes red, and the "
      f"fix is to re-read that message against what the new consumer makes "
      f"true - not to widen this list without doing so",
      _consumers == _EXPECTED_CONSUMERS)
_RULE7 = derived(lambda: open(os.path.join(_APP, "validate.py"),
                              encoding="utf-8").read())
check("...and the phrase the tripwire expired - 'would decide nothing' about "
      "the provider - is GONE from validate.py, so a bad merge that restores "
      "the old reason is red rather than quiet",
      isinstance(_RULE7, str)
      and "supplier you configured would decide nothing" not in _RULE7)
check("...and the refusal still names the consumer that exists, so it cannot "
      "decay into a rule with no stated reason",
      isinstance(_RULE7, str) and "static_tariff.py" in _RULE7)
# ...and the other half: the registry IS on the startup path, which is what
# makes an unwired seam anchored rather than dead code.
check("...while validate.py does import the registry, so the seam is exercised "
      "on every boot",
      derived(lambda: "PROVIDERS" in open(
          os.path.join(_APP, "validate.py"), encoding="utf-8").read()) is True)

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")