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.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ, and a size-preserving
edit reverted inside one second is invisible to CPython's staleness test. That
is TEST-15, and it has already turned one live mutation into a false survivor.
Read MUTATION.md before you report a survivor.
"""

import ast
import asyncio
import hashlib
import inspect
import io
import json
import logging
import os
import shutil
import sys
import tempfile
from datetime import date, datetime, timedelta, timezone
from textwrap import dedent

from app.price import (
    ALL_IN, Absence, BANNED_IN_PLUGIN_CODE, Component, FrameworkOption,
    RETIRED_PROVIDERS,
    MAX_PLAUSIBLE_EUR_KWH, code_leaks,
    PRICE_OPTION_DEFAULTS, PROVIDERS, PluginOptions, PricePoint, PriceProduct,
    PriceProvider, PriceRead, ProviderCapabilities, RawPoint, SourceUnavailable,
    arbitrage_allowed, assemble, brussels, brussels_date_range,
    brussels_offset, check_egress, dated_feed_url,
    price_option,
    export_price_eur_kwh, load_prices, plan_is_provisional, register,
    volatility_regime, uncertainty_of, widen_factor, Uncertainty,
)
from app import price as _price
from app import arbitrage as _arb
from app import main as _main
from app.p1 import make_sample, split_signed
from app.static_tariff import TariffPlan
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
HERE = os.path.dirname(os.path.abspath(__file__))

# ⚠️ 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("'")
        # OPT-04 added the first BOOLEAN price option, and without this the
        # mirror check would compare the string "false" against `False` and go
        # red for a file that is perfectly in step. Parsing a bool as a bool is
        # what the yaml module this image does not carry would have done; it
        # widens nothing, because a key whose shipped value disagrees with the
        # mirror still drifts.
        if raw in ("true", "false"):
            return raw == "true"
        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 PLUS the retired ids, or the UI and the code
# disagree about what is installable.
#
# ⚠️ THE RETIRED IDS ARE IN THE SCHEMA ON PURPOSE AND THIS IS THE CHECK THAT
# KEEPS THEM THERE (TARIFF-07). The Supervisor validates the options file
# against this line BEFORE the add-on runs. Drop a retired id from it and an
# EXISTING config naming that id becomes schema-invalid: the operator gets an
# opaque "expected one of ..." and validate.py's retired-id refusal - the whole
# point of the withdrawal - is never reached, on exactly the sites that need
# it. So the schema is deliberately WIDER than the registry, by exactly
# RETIRED_PROVIDERS and by nothing else. Both halves are graded: an id that is
# in neither set, and a retired id that has gone missing.
_line = next(ln for ln in _schema_block if ln.startswith("  price_provider:"))
_schema_ids = set(_line.split("list(", 1)[1].rstrip(")").split("|"))
_wanted = set(PROVIDERS) | set(RETIRED_PROVIDERS) | {"off"}
check(f"the schema offers exactly the registered plugins, the retired ids and "
      f"'off' {sorted(_schema_ids)}",
      _schema_ids == _wanted)
check(f"...and every retired id is STILL offered, so a config already naming "
      f"one reaches validate.py rather than the Supervisor's schema error "
      f"{sorted(set(RETIRED_PROVIDERS) - _schema_ids) or 'none missing'}",
      set(RETIRED_PROVIDERS) <= _schema_ids and set(RETIRED_PROVIDERS) != set())
check("...and no retired id is ALSO a live plugin, which would silently take "
      "that refusal away",
      set(RETIRED_PROVIDERS) & set(PROVIDERS) == set())


# --------------------------------------------------------------------------- #
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


# ⚠️ THE INVENTED ROW-PER-POINT FIXTURE IS GONE, AND SO IS ITS PARSER
# (TARIFF-07). Until this card every price fixture here was a
# `{"data": [{"start": ..., "price_eur_mwh": ...}]}` document that
# `_points_from_json()` was written to read and that NO FEED ANYBODY REACHED
# HAS EVER PUBLISHED. The three plugins fed from it - `engie_dynamic`,
# `totalenergies_dynamic`, `eneco_eddy` - had no captured body because their
# endpoints do not exist, so the loop was closed twice over: an invented parser
# graded against an invented body, on behalf of plugins that could never fetch.
# Both plugins that remain are replayed from something real - `generic_dayahead`
# from captured bytes (section 2b), `manual_csv` from the CSV a human types.
#
# The shape itself survives in ONE place and in one direction only: as the body
# section 9 feeds `generic_dayahead` to prove that a row-per-point document is a
# FAILURE for it rather than a second dialect it might sniff. It is named for
# what it is.


def row_shape_body(key, values):
    """A `{"data": [...]}` document. ⚠️ EXISTS ONLY TO BE REFUSED.

    No feed this repository has ever reached publishes this, and the parser
    that read it was withdrawn with the three plugins it served (TARIFF-07).
    It is kept as a NEGATIVE stimulus: if `_points_from_columnar_json()` ever
    grows a sniff for `data`, section 9's "the body is the OTHER shape" row
    turns from `no_data` into a series and a check goes red, rather than
    TARIFF-08's scope rule being widened by a review nobody remembers.
    """
    return json.dumps({"data": [
        {"start": (T0 + timedelta(hours=i)).isoformat(), key: v,
         "provisional": False}
        for i, v in enumerate(values)]}).encode()



def columnar_feed(values, provisional_at=None, unit="EUR / MWh",
                  step_min=60, start=None, prices=None):
    """A body in the shape `api.energy-charts.info` really publishes.

    Parallel arrays and no per-point object; `unix_seconds` are UTC epoch
    seconds. `prices` overrides the price column outright, which is how the
    unequal-length case below is built without hand-writing JSON.

    ⚠️ THERE IS NO PER-POINT `provisional` IN THIS SHAPE, so `provisional_at`
    exists only to be REFUSED - the columnar document cannot carry the flag and
    the plugin supplies it. It is a keyword the callers do not pass; the
    parameter is here so that a future caller who reaches for it gets a
    TypeError rather than a silently ignored argument.
    """
    if provisional_at is not None:
        raise TypeError(
            "the columnar document has no per-point provisional flag; "
            "`generic_dayahead` declares it in fetch(). See "
            "_points_from_columnar_json().")
    t0 = start if start is not None else T0
    secs = [int((t0 + timedelta(minutes=step_min * i)).timestamp())
            for i in range(len(values))]
    doc = {"unix_seconds": secs,
           "price": list(values) if prices is None else list(prices),
           "unit": unit,
           "license_info": "CC BY 4.0 (creativecommons.org/licenses/by/4.0) "
                           "from Bundesnetzagentur | SMARD.de",
           "deprecated": False}
    return json.dumps(doc).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.
# ⚠️ `generic_dayahead` IS FED THE COLUMNAR SHAPE AND THE OTHER THREE ARE NOT,
# and the asymmetry is the honest one rather than an oversight: the columnar
# document is what the one endpoint anybody has ever reached actually returns.
FIXTURES = {
    "generic_dayahead": (columnar_feed(EPEX_EUR_MWH),
                         [v / 1000 for v in EPEX_EUR_MWH]),
    "manual_csv": (None, [0.1834, -0.0210, 0.3115]),
}

check(f"the two plugins that survive TARIFF-07 are registered "
      f"{sorted(PROVIDERS)}",
      set(PROVIDERS) == {"generic_dayahead", "manual_csv"})
check(f"...and the three withdrawn ids resolve to NOTHING "
      f"{sorted(RETIRED_PROVIDERS)}",
      set(RETIRED_PROVIDERS) == {"engie_dynamic", "totalenergies_dynamic",
                                 "eneco_eddy"}
      and not (set(RETIRED_PROVIDERS) & set(PROVIDERS)))
def _readd_retired():
    return register(type("_ReAdd", (PriceProvider,), {
        "id": "eneco_eddy",
        "capabilities": classmethod(lambda c: ProviderCapabilities(
            product=PriceProduct.DAY_AHEAD,
            sources=frozenset({Component.ENERGY}))),
        "fetch": lambda s, o, a, b, t: None}))


# ⚠️ GRADED ON THE REASON, NOT THE VERDICT. `register()` already refuses a
# DUPLICATE id, so "it raised ValueError" would pass on a tree where the
# retired guard is deleted and the id happens to be live again - the masking
# shape MUTATION.md's TEST-16 section is about.
_readd_why = _reason(_readd_retired)
check(f"re-registering a retired id is refused, and refused for the RETIRED "
      f"reason rather than the duplicate one ({_readd_why[:60]!r})",
      "RETIRED_PROVIDERS" in _readd_why and "eneco_eddy" in _readd_why)
check("...and the refusal did not take effect anyway",
      "eneco_eddy" not in PROVIDERS)
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("2b. TARIFF-08 - the SHIPPED parser, replayed against REAL captured "
      "bodies")
#
# ⚠️ WHY THIS SECTION EXISTS, WHICH IS THE MORE IMPORTANT HALF OF THE TICKET.
# Until 2026-09-10 every price fixture in this repository - here, in
# test_price_history.py and in test_static_tariff.py - was a
# `{"data": [{"start": ..., "price_eur_mwh": ...}]}` document invented by
# whoever wrote `_points_from_json()`. `api.energy-charts.info` has never
# published that shape. So the suite validated the parser against the parser's
# own assumption: a closed loop that cannot notice the assumption is wrong, and
# it stayed green while `generic_dayahead` raised `SourceUnavailable` on every
# live fetch it ever made. §11.4 then made that silent - a parse error becomes
# `no_data` becomes the static profile with arbitrage off, which is the CORRECT
# degradation and says nothing at all about the cause.
#
# The only thing that breaks the loop is bytes nobody here wrote. Both fixtures
# below are unmodified response bodies, captured by hand with curl on
# 2026-09-10; `scenarios/energy-charts-be-price-captures.json` records the URL,
# the retrieval time, the licence as returned and a sha256 of each, and the
# checksums are RE-DERIVED here so a fixture edited to make a check pass stops
# being a capture and starts being a failure.
#
# ⚠️ THERE IS NOW EXACTLY ONE NETWORKED PLUGIN, WHICH IS TARIFF-07's DOING.
# `engie_dynamic`, `totalenergies_dynamic` and `eneco_eddy` had no capture
# because their endpoints do not exist - verified by hand on 2026-09-13: a 404
# from a gateway with no route, a TLS hostname mismatch, and a 404 with an empty
# body. They were withdrawn rather than left fed by an invented fixture. The
# check below is derived from the shipped `PROVIDERS` registry, so a plugin
# added with no capture forces the question rather than inheriting silence.

_CAPTURES = json.load(open(
    os.path.join(HERE, "scenarios", "energy-charts-be-price-captures.json"),
    encoding="utf-8"))


def _capture(name):
    with open(os.path.join(HERE, "scenarios", name), "rb") as fh:
        return fh.read()


_REAL = {n: _capture(n) for n in _CAPTURES["files"]}
_bad_sha = sorted(n for n, b in _REAL.items()
                  if hashlib.sha256(b).hexdigest()
                  != _CAPTURES["files"][n]["sha256"])
check(f"all three captured bodies are the bytes the provenance file "
      f"recorded {sorted(_REAL)}",
      len(_REAL) == 3 and _bad_sha == [])

# The document really is columnar, derived from the bytes rather than asserted:
# if it ever grows a `data` key this check says so before any parser does.
_shapes = {n: sorted(json.loads(b.decode("utf-8"))) for n, b in _REAL.items()}
check(f"...and every one is COLUMNAR - parallel arrays, no 'data' key "
      f"{sorted({tuple(v) for v in _shapes.values()})}",
      all(v == ["deprecated", "license_info", "price", "unit", "unix_seconds"]
          for v in _shapes.values()))
check("...which is precisely the shape the shipped parser used to demand a "
      "'data' list from",
      all("data" not in v for v in _shapes.values()))

# ⚠️ CC BY 4.0 IS WHY THESE BYTES MAY BE IN THE REPOSITORY AT ALL. DATA-02
# established that the BE zone is redistributable with attribution; the feed
# restates the licence per response, so a capture whose own licence line has
# changed is a capture that must not stay committed.
_lic = {n: json.loads(b.decode("utf-8"))["license_info"] for n, b in _REAL.items()}
check(f"...and each carries the CC BY 4.0 line the provenance block cites",
      all("CC BY 4.0" in v for v in _lic.values())
      and all(_lic[n] == _CAPTURES["files"][n]["licence_as_returned"]
              for n in _REAL))

_QH = "energy-charts-be-price-2026-09-10-quarterhourly.json"
_H1 = "energy-charts-be-price-2025-09-10-hourly.json"

# ⚠️ THE TWO GRANULARITIES ARE NOT DECORATION. Belgium moved to a 15-minute MTU
# on 2025-10-01 and the same endpoint answers 96 points at 900 s for a day after
# it and 24 at 3600 s for a day before it. A fixture at one spacing cannot grade
# whether anything assumes the other, so both are replayed and the spacings are
# DERIVED from the bodies rather than typed.
_spacing = {}
for _n, _b in _REAL.items():
    _u = json.loads(_b.decode("utf-8"))["unix_seconds"]
    _spacing[_n] = sorted({_u[i + 1] - _u[i] for i in range(len(_u) - 1)})
check(f"the two captures really differ in granularity - {_spacing[_QH]} s over "
      f"{len(json.loads(_REAL[_QH])['price'])} points and {_spacing[_H1]} s "
      f"over {len(json.loads(_REAL[_H1])['price'])} points - or replaying both "
      f"grades nothing more than replaying one",
      _spacing[_QH] == [900] and _spacing[_H1] == [3600])

_REPLAY = {n: load("generic_dayahead", transport=canned(b))
           for n, b in _REAL.items()}
_not_series = sorted(n for n, r in _REPLAY.items() if r.kind != "series")
check(f"the SHIPPED generic_dayahead parses both real bodies "
      f"{[r.words for r in _REPLAY.values()]}",
      _not_series == [])

# The point count is the source's own interval count, taken from the document.
_counts = derived(lambda: {n: len(r.points) for n, r in _REPLAY.items()})
_want_counts = {n: len(json.loads(b.decode("utf-8"))["price"])
                for n, b in _REAL.items()}
check(f"...emitting one point per published interval, no decimation "
      f"{_counts}", _counts == _want_counts)
check(f"...which for the quarter-hourly body is 96 points and not 24 - the "
      f"15-minute MTU is the shape of the real feed, not a hypothetical",
      _want_counts[_QH] == 96 and _want_counts[_H1] == 24)

# §9.3: unix_seconds are UTC instants, RENDERED at the Brussels offset. Both
# captured days are inside CEST, so the correct first start is local midnight at
# +02:00. ⚠️ A NAIVE READING - `datetime.fromtimestamp(t)` with no tzinfo - is
# what this grades: it would produce the host's local wall clock, and on a UTC
# host that is 22:00 of the previous day. The expectation is computed here from
# the raw epoch second by different arithmetic than price.py uses.
_first_raw = {n: json.loads(b.decode("utf-8"))["unix_seconds"][0]
              for n, b in _REAL.items()}
_first_utc = {n: datetime(1970, 1, 1, tzinfo=UTC) + timedelta(seconds=t)
              for n, t in _first_raw.items()}
_first_got = derived(lambda: {n: r.points[0].start
                              for n, r in _REPLAY.items()})
check(f"every start is the UTC instant the feed published "
      f"{ {n: v.isoformat() for n, v in _first_utc.items()} }",
      isinstance(_first_got, dict)
      and all(_first_got[n] == _first_utc[n] for n in _REAL))
check(f"...rendered at the Brussels offset, so a September day starts at "
      f"00:00+02:00 and NOT at the 22:00 a naive local reading gives "
      f"{ {n: v.isoformat() for n, v in (_first_got if isinstance(_first_got, dict) else {}).items()} }",
      isinstance(_first_got, dict)
      and all(v.utcoffset() == timedelta(hours=2) for v in _first_got.values())
      and all((v.hour, v.minute) == (0, 0) for v in _first_got.values()))

# EUR/MWh -> all-in EUR/kWh, recomputed here from the raw column.
_allin_want = {n: [round(v / 1000 + MARKUP + TAXES, 6)
                   for v in json.loads(b.decode("utf-8"))["price"]]
               for n, b in _REAL.items()}
_allin_got = derived(lambda: {n: [p.eur_per_kwh for p in r.points]
                              for n, r in _REPLAY.items()})
check(f"...and every price is the published EUR/MWh figure over 1000 plus "
      f"markup and taxes (first: {[v[0] for v in _allin_want.values()]})",
      _allin_got == _allin_want)
# ⚠️ NON-DEGENERACY. If the captured prices happened to be all equal, the
# arithmetic check above would pass under a parser that emitted the same number
# for every interval.
_raw_spread = {n: round(max(v) - min(v), 6)
               for n, v in ((k, json.loads(b.decode("utf-8"))["price"])
                            for k, b in _REAL.items())}
check(f"...on bodies whose prices actually vary, or that check grades a "
      f"constant {_raw_spread}",
      all(v > 1.0 for v in _raw_spread.values()))


# --- the refusals, driven from the real body by breaking one thing ---------- #

def _edited(name, **fields):
    """The captured body with one field replaced. Everything else is real."""
    doc = json.loads(_REAL[name].decode("utf-8"))
    doc.update(fields)
    return json.dumps(doc).encode()


_full = json.loads(_REAL[_QH].decode("utf-8"))
_short = load("generic_dayahead",
              transport=canned(_edited(_QH, price=_full["price"][:-4])))
check(f"a `price` array 4 entries short of `unix_seconds` is no_data, NOT a "
      f"series truncated to {len(_full['price']) - 4} points - the missing "
      f"tail of a Belgian day is the evening peak, and a short series is "
      f"traded on where no data at all switches arbitrage off (§11.4)",
      _short.kind == "no_data")
check("...and the reason names the two lengths, so an operator can see it was "
      "a length mismatch and not an outage",
      _short.kind == "no_data"
      and str(len(_full["unix_seconds"])) in _short.reason
      and str(len(_full["price"]) - 4) in _short.reason)
# The other direction, so the check is not a one-sided `len(price) < len(time)`.
_long = load("generic_dayahead",
             transport=canned(_edited(_QH, price=_full["price"] + [1.0])))
check("...and a `price` array one entry LONGER is refused too",
      _long.kind == "no_data")
# ⚠️ THE CONTROL. Without it, "refuses everything" passes both checks above.
check("...while the untouched body is still a series, so the refusal is the "
      "length and not the editing",
      load("generic_dayahead",
           transport=canned(_edited(_QH))).kind == "series")

_unit = load("generic_dayahead", transport=canned(_edited(_QH,
                                                          unit="EUR / kWh")))
check("a feed that changes its unit to EUR/kWh is no_data, not every price "
      "silently divided by 1000 - MAX_PLAUSIBLE_EUR_KWH only bounds prices "
      "that are too LARGE, so nothing downstream could catch it",
      _unit.kind == "no_data")

_nulls = _full["price"][:]
for _i in (3, 17):
    _nulls[_i] = None
_nulled = load("generic_dayahead", transport=canned(_edited(_QH,
                                                            price=_nulls)))
check(f"a published null is DROPPED, leaving a hole rather than a zero - a "
      f"missing quarter is not a free quarter",
      _nulled.kind == "series"
      and derived(lambda: len(_nulled.points)) == len(_full["price"]) - 2)
check("...and the dropped instants are the ones the feed left null, not the "
      "first two",
      derived(lambda: sorted(
          int(p.start.timestamp()) for p in _nulled.points))
      == sorted(t for i, t in enumerate(_full["unix_seconds"])
                if i not in (3, 17)))
_allnull = load("generic_dayahead",
                transport=canned(_edited(_QH,
                                         price=[None] * len(_full["price"]))))
check("...but a body with a price for nothing at all is no_data, not an empty "
      "series", _allnull.kind == "no_data")
# ⚠️ WHICH RULE REFUSED IT, not merely that something did. `_validate_raw()`
# would also refuse an empty list, one layer further out and with a sentence
# that says nothing about the feed - so asserting only `no_data` here would
# grade the same under a parser that had lost this refusal entirely.
check(f"...refused by the COLUMNAR parser, naming what arrived, rather than "
      f"reaching the boundary validator as a generic empty list - "
      f"{_allnull.reason!r}",
      _allnull.kind == "no_data"
      and f"{len(_full['price'])} instants" in _allnull.reason
      and "a price for none of them" in _allnull.reason)

# ⚠️ THE TWO WARNINGS ARE GRADED, NOT ASSUMED. §11.4 is exactly what makes a
# price defect quiet - a parse failure becomes no_data becomes the static
# profile, correctly and silently - so an unverified diagnostic is the first
# thing here to rot. Both are captured off the shipped logger.
_wbuf = io.StringIO()
_wh = logging.StreamHandler(_wbuf)
logging.getLogger("goodwe.price").addHandler(_wh)
load("generic_dayahead", transport=canned(_edited(_QH, price=_nulls)))
_dep = load("generic_dayahead", transport=canned(_edited(_QH,
                                                         deprecated=True)))
logging.getLogger("goodwe.price").removeHandler(_wh)
_warn = _wbuf.getvalue()
_n_null = len([v for v in _nulls if v is None])
check(f"the dropped nulls are reported with a count ({_n_null} of "
      f"{len(_nulls)}), so a series with holes in it is visible to an operator "
      f"rather than merely short",
      f"{_n_null} of {len(_nulls)}" in _warn)
check("...and a feed announcing deprecated=true says so WHILE IT STILL WORKS - "
      "the day it stops, §11.4 makes its death look exactly like a quiet one",
      "deprecated" in _warn and _dep.kind == "series")

# Which plugins are replayed from real bytes, derived from the registry so a
# sixth supplier cannot be added without this line noticing.
_networked = sorted(pid for pid, cls in PROVIDERS.items()
                    if cls.capabilities().egress_hosts)
_replayed = ["generic_dayahead"]
check(f"the networked plugins are {_networked}, and EVERY one of them is "
      f"replayed from a captured real body ({_replayed}) - no shipped plugin "
      f"reaches a network this file has never seen answer (TARIFF-07)",
      _networked == _replayed)


# --------------------------------------------------------------------------- #
print()
print("2c. TARIFF-09 - the RANGE the caller asked for, on the Brussels "
      "calendar")
#
# ⚠️ WHAT WAS WRONG, AND WHY NO CHECK IN THIS FILE COULD SEE IT.
# `generic_dayahead.fetch()` accepted `since` and `until` and requested the
# bare `https://api.energy-charts.info/price?bzn=BE`, which answers ONE
# Brussels day. Everything past today was then filled from yesterday's profile
# by FCST-04, flagged provisional, and §8.3 dropped all of it from the trade
# set - so the optimiser never saw tomorrow as firm, at any hour of any day.
# Measured on the merged tree at 2026-09-10T09:00+02:00 over a 48 h horizon:
# 15 firm whole hours and 33 carried, both ends of the only permitted cycle
# inside ONE calendar day; and at 23:00, 1 firm hour and 47 carried, where the
# firm spread no longer clears the round-trip cost at all.
#
# The reason section 2b could not catch it is that section 2b fed the parser
# BYTES and never looked at the REQUEST. `canned()` has always recorded the URL
# it was asked for and nothing asserted on it - so the transport knew the
# plugin was ignoring its arguments and no check ever asked. The checks below
# assert on the URL, and the two-day body they replay is a capture of what that
# URL really answers.

_TD = "energy-charts-be-price-2026-09-09_2026-09-10-twoday.json"

# ⚠️ NON-DEGENERACY FIRST (habit 1): the two-day capture must really cover TWO
# Brussels days and the one-day capture ONE, or "the range reached the feed"
# is graded against two fixtures that are the same shape.
_bdays = {n: sorted({brussels(datetime.fromtimestamp(t, UTC)).date().isoformat()
                     for t in json.loads(b.decode("utf-8"))["unix_seconds"]})
          for n, b in _REAL.items()}
check(f"the captures span the local-day counts this section needs - "
      f"{ {n: len(v) for n, v in _bdays.items()} } - or the range check has "
      f"nothing to distinguish {_bdays[_TD]} from {_bdays[_QH]}",
      len(_bdays[_TD]) == 2 and len(_bdays[_QH]) == 1
      and len(_bdays[_H1]) == 1)
_npts = {n: len(json.loads(b.decode("utf-8"))["price"])
         for n, b in _REAL.items()}
check(f"...and the two-day body carries 192 quarter-hours where the one-day "
      f"body carries 96, which is the counterfactual this card is built on "
      f"{_npts}",
      len(json.loads(_REAL[_TD].decode("utf-8"))["price"]) == 192
      and len(json.loads(_REAL[_QH].decode("utf-8"))["price"]) == 96)

# --- the URL the SHIPPED plugin asks for ----------------------------------- #


def asked_url(since, until, extra=None):
    """The URL `generic_dayahead` really requests for a window. Behaviour,
    not source text: it comes back off the injected transport."""
    t = canned(_REAL[_TD])
    o = opts(price_provider="generic_dayahead", **(extra or {}))
    run(load_prices(o, since, until, transport=t))
    return t.calls[0][0]


# The expectation is built HERE, from the window, by different arithmetic than
# price.py uses: a literal string with the two dates spelled out.
_w0 = datetime.fromisoformat("2026-09-09T09:17:43+02:00")
check(f"a 48 h horizon from 09:17 local asks the feed for the three Brussels "
      f"days it touches -> {asked_url(_w0, _w0 + timedelta(hours=48))!r}",
      asked_url(_w0, _w0 + timedelta(hours=48))
      == "https://api.energy-charts.info/price?bzn=BE"
         "&start=2026-09-09&end=2026-09-11")
check("...and the shipped default no longer asks the bare URL that answers a "
      "single day",
      asked_url(_w0, _w0 + timedelta(hours=48))
      != "https://api.energy-charts.info/price?bzn=BE")

# ⚠️ THE BRUSSELS/UTC BOUNDARY, WHICH IS THE TRAP. `since`/`until` are
# INSTANTS and the API takes DATES. At 22:30 UTC in September Belgium is
# already on the NEXT calendar day, so a range derived from `since.date()` -
# or from `since.astimezone(utc).date()`, which is the same thing - asks for
# the day the controller has just left and drops the day it is standing in.
# This is `_bdate()`'s lesson one layer up: price_history.py MISFILES an hour
# when it gets this wrong; here the hours are never fetched at all.
_edge = datetime.fromisoformat("2026-09-10T22:30:00+00:00")
check(f"the precondition: {_edge.isoformat()} really is a different day in "
      f"Brussels ({brussels(_edge).date().isoformat()}) than in UTC "
      f"({_edge.date().isoformat()}), or the boundary check grades nothing",
      brussels(_edge).date().isoformat() != _edge.date().isoformat())
check(f"...so the range starts on the BRUSSELS day, 2026-09-11, and not on "
      f"the UTC day 2026-09-10 -> "
      f"{asked_url(_edge, _edge + timedelta(hours=12))!r}",
      asked_url(_edge, _edge + timedelta(hours=12))
      == "https://api.energy-charts.info/price?bzn=BE"
         "&start=2026-09-11&end=2026-09-11")

# ...and the SAME trap on the other end. `until` at 22:30 UTC is already
# 00:30 of the next Brussels day, so a UTC-derived END stops one day short and
# the last hours of the horizon are never fetched. Both ends of the range
# derivation are graded, or a mutation only has to get one of them right.
_edge2_a = datetime.fromisoformat("2026-09-09T12:00:00+00:00")
_edge2_b = datetime.fromisoformat("2026-09-10T22:30:00+00:00")
check(f"the precondition: {_edge2_b.isoformat()} is "
      f"{brussels(_edge2_b).date().isoformat()} in Brussels and "
      f"{_edge2_b.date().isoformat()} in UTC, so the two answers differ",
      brussels(_edge2_b).date().isoformat() != _edge2_b.date().isoformat())
check(f"...so the range ENDS on the Brussels day 2026-09-11, not on the UTC "
      f"day 2026-09-10 -> {asked_url(_edge2_a, _edge2_b)!r}",
      asked_url(_edge2_a, _edge2_b)
      == "https://api.energy-charts.info/price?bzn=BE"
         "&start=2026-09-09&end=2026-09-11")

# ⚠️ `until` IS EXCLUSIVE, so local midnight belongs to the day BEFORE it. A
# 24 h horizon from local midnight is ONE day, not two.
_mid = datetime.fromisoformat("2026-09-09T00:00:00+02:00")
check(f"a horizon ending exactly at local midnight asks for one day, because "
      f"`until` is exclusive -> {asked_url(_mid, _mid + timedelta(hours=24))!r}",
      asked_url(_mid, _mid + timedelta(hours=24))
      == "https://api.energy-charts.info/price?bzn=BE"
         "&start=2026-09-09&end=2026-09-09")
check(f"...while one second past midnight is two days, so the rule is the "
      f"exclusive END and not a fixed subtraction -> "
      f"{asked_url(_mid, _mid + timedelta(hours=24, seconds=1))!r}",
      asked_url(_mid, _mid + timedelta(hours=24, seconds=1))
      == "https://api.energy-charts.info/price?bzn=BE"
         "&start=2026-09-09&end=2026-09-10")
check(f"an empty or inverted window collapses to a single day rather than "
      f"asking for a backwards range no feed has a defined answer for -> "
      f"{asked_url(_mid, _mid - timedelta(hours=5))!r}",
      asked_url(_mid, _mid - timedelta(hours=5))
      == "https://api.energy-charts.info/price?bzn=BE"
         "&start=2026-09-09&end=2026-09-09")

# The helper on its own, over the DST transitions, where a fixed-offset
# assumption and a correct one disagree.
_mar = datetime.fromisoformat("2026-03-29T00:30:00+00:00")   # 02:30 CEST
_bdr = derived(lambda: brussels_date_range(_mar, _mar + timedelta(hours=1)))
check(f"brussels_date_range() is right across the spring transition, where "
      f"00:30 UTC is already 02:30 local -> {_bdr}",
      _bdr == (date(2026, 3, 29), date(2026, 3, 29)))

# --- the two-day body, end to end through the SHIPPED parser --------------- #

_two = load("generic_dayahead", transport=canned(_REAL[_TD]))
check(f"the shipped generic_dayahead parses the real two-day body "
      f"{_two.words}", _two.kind == "series")
def _series_days(read):
    return derived(lambda: sorted({brussels(p.start).date().isoformat()
                                   for p in read.points}))


_two_days = _series_days(_two)
check(f"...and the series it produces spans TWO Brussels local days "
      f"{_two_days}, where the one-day body's spans one "
      f"{_series_days(_REPLAY[_QH])}",
      _two_days == ["2026-09-09", "2026-09-10"]
      and len(_two_days) == 2)
check(f"...one point per published interval, 192 of them, no decimation and "
      f"no day dropped ({derived(lambda: len(_two.points))})",
      derived(lambda: len(_two.points)) == 192)

# --- the SHORT ANSWER, which is normal and must stay a series -------------- #
#
# ⚠️ ASKING FOR TOMORROW BEFORE THE ~13:00-14:00 CET AUCTION RETURNS FEWER DAYS
# THAN REQUESTED, AND THAT IS NOT AN ERROR. Measured: at 2026-09-10T07:46:55Z
# the request `?bzn=BE&start=2026-09-10&end=2026-09-11` came back BYTE-IDENTICAL
# to the bare-URL capture - one day, 96 quarter-hours. A `no_data` there would
# switch a working site's arbitrage off every morning, so the requirement is
# that the short body stays a `series` of what EXISTS and FCST-04's fill covers
# the rest with flagged, provisional hours. The identity is re-derived from the
# bytes rather than restated.
_prepub = _CAPTURES["files"][_QH]["also_returned_verbatim_by"]
check(f"the pre-publication range request {_prepub['url']!r} returned the "
      f"SAME BYTES as the bare default URL - one day where two were asked for",
      hashlib.sha256(_REAL[_QH]).hexdigest()
      == _CAPTURES["files"][_QH]["sha256"]
      and "start=2026-09-10&end=2026-09-11" in _prepub["url"])
check(f"...and the shipped parser answers that short body with a SERIES of "
      f"what exists, never no_data - {_REPLAY[_QH].words}",
      _REPLAY[_QH].kind == "series"
      and derived(lambda: len(_REPLAY[_QH].points)) == 96)
# ⚠️ THE COUNTS ARE TAKEN THROUGH `derived()` AND COMPARED ONLY WHEN BOTH ARE
# NUMBERS. A mutation that turns the short answer into `no_data` makes
# `.points` raise `Absence` by design, and `_Raised < _Raised` is a TypeError -
# which takes the file down with a traceback instead of reporting a FAIL. A
# check whose failure mode is a crash has no name attached (habit 5's shape,
# with an exception in place of a hang).
_n_short = derived(lambda: len(_REPLAY[_QH].points))
_n_two = derived(lambda: len(_two.points))
check(f"...which is strictly fewer points than the two-day body "
      f"({_n_short} < {_n_two}), so 'short' is a real property of this "
      f"stimulus and not a name for it",
      isinstance(_n_short, int) and isinstance(_n_two, int)
      and _n_short < _n_two)

# --- price_feed_url: the exact URL, and the cost said out loud ------------- #
#
# ⚠️ THE DECISION. An override is used VERBATIM and this plugin appends
# nothing to it. Merging our `start`/`end` into an operator's query string
# would mean re-serialising a URL they wrote, and `start`/`end` are
# energy-charts' spelling - and this provider also DECLARES a SECOND host,
# `web-api.tp.entsoe.eu`, whose date parameters are not those and are not
# verified here, so appending our vocabulary there yields two ignored
# parameters and the short horizon anyway. The cost is
# that an overriding site silently gets the horizon this card exists to fix,
# so it is NOT silent: a warning names the consequence at fetch time.
_ovr = "https://api.energy-charts.info/price?bzn=BE&somethingelse=1"
check(f"an override is requested EXACTLY as given, with nothing appended -> "
      f"{asked_url(_w0, _w0 + timedelta(hours=48), {'price_feed_url': _ovr})!r}",
      asked_url(_w0, _w0 + timedelta(hours=48),
                {"price_feed_url": _ovr}) == _ovr)

_obuf = io.StringIO()
_oh = logging.StreamHandler(_obuf)
logging.getLogger("goodwe.price").addHandler(_oh)
asked_url(_w0, _w0 + timedelta(hours=48), {"price_feed_url": _ovr})
_ovr_log = _obuf.getvalue()
_obuf2 = io.StringIO()
_oh2 = logging.StreamHandler(_obuf2)
logging.getLogger("goodwe.price").removeHandler(_oh)
logging.getLogger("goodwe.price").addHandler(_oh2)
asked_url(_w0, _w0 + timedelta(hours=48))
_dflt_log = _obuf2.getvalue()
logging.getLogger("goodwe.price").removeHandler(_oh2)
check(f"...and the operator is TOLD what it costs them, at fetch time, naming "
      f"the horizon consequence rather than only the substitution",
      "price_feed_url" in _ovr_log and _ovr in _ovr_log
      and "provisional" in _ovr_log and "2026-09-11" in _ovr_log)
# ⚠️ THE CONTROL. Without it, "warns about everything" passes the line above.
check("...while the DEFAULT path says nothing, so the warning is about the "
      "override and not about fetching",
      "price_feed_url is set" not in _dflt_log)

# The builder refuses to append to a base that already has a query string,
# which is the "URL nobody intended" this decision is meant to make
# unreachable. Driven, not read off the source.
check(f"dated_feed_url() refuses a base that already carries a query string, "
      f"rather than producing a second '?' - "
      f"{_reason(lambda: dated_feed_url('https://h/p?a=1', b='2'))!r}",
      raises(lambda: dated_feed_url("https://h/p?a=1", b="2"),
             SourceUnavailable))
check("...and a clean base still builds, so the refusal is the existing query "
      "string and not the builder",
      dated_feed_url("https://h/p", b="2") == "https://h/p?b=2")

# --- the plugin contract: who consumes the range, pinned ------------------- #
#
# ⚠️ THIS IS A SOURCE-SHAPE PIN AND IS NAMED AS ONE (habit 4). The claim is
# about the CODE - which plugins reference the range arguments they are handed
# - which is the same kind of claim `code_leaks()` makes, and it is not
# evidence that any feed parses. `generic_dayahead`'s BEHAVIOUR is graded above
# by the URL it really requests.
#
# It exists because `fetch(self, opts, since, until, transport)` promises every
# plugin the range and the promise is NOT enforced. Enforcing it would oblige
# the three plugins whose endpoints nobody has verified (TARIFF-07) to invent a
# date-parameter vocabulary - TARIFF-08's defect moved from the response to the
# request. So the split is pinned instead, and a sixth supplier has to change
# this line rather than inherit the silence.
def _uses_range(cls):
    src = dedent(inspect.getsource(cls.fetch))
    tree = ast.parse(src)
    names = {n.id for n in ast.walk(tree) if isinstance(n, ast.Name)}
    return bool(names & {"since", "until"})


_uses = sorted(pid for pid, cls in PROVIDERS.items() if _uses_range(cls))
check(f"the plugins that CONSUME their `since`/`until` are {_uses}; the rest "
      f"take the range and ignore it, which is what TARIFF-09 fixed for "
      f"generic_dayahead - and after TARIFF-07 the only one left that does not "
      f"is the offline plugin, for the honest reason below",
      _uses == ["generic_dayahead"])
check("...and `manual_csv` is honestly among the ones that do not: it reads a "
      "whole file and the fill windows it downstream, so a range parameter "
      "would have nowhere to go",
      "manual_csv" in sorted(set(PROVIDERS) - set(_uses)))


# --------------------------------------------------------------------------- #
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 shipped plugin whose
# feed is wholesale, so it is the only one that gets both components - and a
# feed declaring more must get strictly less. Derived by comparing output on the
# SAME energy figure rather than by restating numbers.
#
# ⚠️ THE MIDDLE RUNG - a feed sourcing {ENERGY, MARKUP}, which gets TAXES only -
# HAS NO SHIPPED DECLARER SINCE TARIFF-07, and is graded through `assemble()`
# with a hand-built declaration instead of being dropped. It is the only rung
# that distinguishes "adds the complement" from "adds everything or nothing",
# so deleting it with the three retail plugins would have left the partial case
# ungraded on the exact function that must not get it wrong.
def added(pid):
    return derived(lambda: round(READS[pid].points[0].eur_per_kwh
                                 - FIXTURES[pid][1][0], 6))


def added_for(sources):
    """What assemble() adds to ONE point declared as sourcing `sources`."""
    caps = ProviderCapabilities(product=PriceProduct.DAY_AHEAD,
                                sources=frozenset(sources))
    e = RETAIL_EUR_KWH[0]
    return derived(lambda: round(
        assemble(caps, [RawPoint(T0, e, False)], MARKUP, TAXES)[0].eur_per_kwh
        - e, 6))


_generic_add = added("generic_dayahead")
_csv_add = added("manual_csv")
_retail_add = added_for({Component.ENERGY, Component.MARKUP})
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 feed that already carries the markup gets taxes ONLY "
      f"({_retail_add}), exactly one markup less than the wholesale one",
      isinstance(_retail_add, float) and isinstance(_generic_add, float)
      and abs(_retail_add - TAXES) < 1e-9
      and abs((_generic_add - _retail_add) - MARKUP) < 1e-9)
check(f"an all-in feed gets nothing added ({_csv_add})",
      isinstance(_csv_add, float) and abs(_csv_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.
#
# ⚠️ `generic_dayahead` IS EXCLUDED FROM THE "MIXED" REQUIREMENT, AND THE
# EXCLUSION IS THE POINT RATHER THAN A GAP. The columnar document the real feed
# publishes has no per-point provisional field - it is a settled auction result
# - so the plugin declares `False` for all of them. Demanding a mixed series
# from it would mean feeding it an invented body carrying an invented flag,
# which is exactly the closed loop TARIFF-08 exists to remove. It is graded on
# what its real body can grade instead, immediately below.
for pid in sorted(set(READS) - {"generic_dayahead"}):
    read = READS[pid]
    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)
_gen_flags = derived(lambda: [p.provisional
                              for p in READS["generic_dayahead"].points])
check(f"generic_dayahead's settled auction comes out settled at every point "
      f"{_gen_flags}",
      isinstance(_gen_flags, list) and len(_gen_flags) == len(EPEX_EUR_MWH)
      and not any(_gen_flags))
check("...so plan_is_provisional is False for it, derived from the same points",
      derived(lambda: plan_is_provisional(READS["generic_dayahead"])) is False)
# ...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)]


# The honest counterpart to _Rogue: the SAME declaration _Rogue makes, on a
# feed that really does publish only energy. No shipped plugin is needed for it
# and none is used - the comparison is between two declarations, not two
# suppliers.
_honest_caps = ProviderCapabilities(
    product=PriceProduct.DAY_AHEAD,
    sources=frozenset({Component.ENERGY, Component.MARKUP}))
_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("manual_csv")
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 "manual_csv" 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 set(_srclen) == set(PROVIDERS)
      and _srclen != {} 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 every shipped plugin is 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, and a
# plugin whose feed already carries the markup naturally writes the option key
# out to say the framework must not add it again - which a RAW text scan reads
# as a leak. A check that fires on the documentation the same file mandates is
# a check somebody deletes.
#
# ⚠️ THE STIMULUS IS BUILT HERE NOW, AND SAYING SO IS THE POINT. Until
# TARIFF-07 this asserted on the SHIPPED set, because `engie_dynamic`'s
# docstring happened to spell `price_markup_eur_kwh` in full. It was withdrawn,
# no remaining shipped docstring spells a full key, and grading the shipped set
# would therefore now pass on a tree with the strip DELETED - a check that had
# quietly stopped being one. So the false-positive is constructed, and the
# shipped set is asserted to be clean separately, above.
class _DocumentedPlugin(PriceProvider):
    """A plugin documenting, correctly, that price_markup_eur_kwh is not added.

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

    id = "_documented_not_shipped"

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

    async def fetch(self, o, since, until, transport):
        return [RawPoint(T0, RETAIL_EUR_KWH[0], False)]


_raw_hit = derived(lambda: sorted(
    b for b in _banned if b in inspect.getsource(_DocumentedPlugin)))
check(f"a RAW text scan false-positives on a mandated docstring {_raw_hit}, "
      f"which is why code_leaks() strips docstrings before searching - and on "
      f"the same class code_leaks() reports {code_leaks(_DocumentedPlugin)}",
      _raw_hit == ["price_markup_eur_kwh"]
      and code_leaks(_DocumentedPlugin) == []
      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)")
# ⚠️ NO SHIPPED PLUGIN DECLARES `PriceProduct.REALTIME`, AND THAT IS RECORDED
# HERE RATHER THAN LEFT TO BE REDISCOVERED (TARIFF-07). `eneco_eddy` was its
# sole declarer and it is withdrawn, so `_WIDE`, `Uncertainty.REALTIME_PRODUCT`
# and `WIDEN_FACTOR`'s 2.5 have no shipped caller. They are KEPT - a reviewer
# finding a mechanism with zero callers would reasonably delete it, and the
# first real imbalance product would then have to rebuild the AC2 seam - and
# this pin is what tells that reviewer the emptiness is intentional. See
# DECISIONS/TARIFF-07-withdrawing-three-plugins.md.
#
# ⚠️ IT IS A PIN, NOT A PROHIBITION. If a plugin that really is real-time ever
# ships, this check goes red and the right fix is to EDIT it - and the decision
# record with it - not to delete the seam.
_products = {pid: cls.capabilities().product for pid, cls in PROVIDERS.items()}
_realtime = sorted(p for p, v in _products.items()
                   if v is PriceProduct.REALTIME)
check(f"no SHIPPED plugin declares real-time/imbalance {_realtime or '[]'} - "
      f"§9.2's row 2 is unoccupied since TARIFF-07 withdrew eneco_eddy, and "
      f"the widening path below is kept anyway, on purpose",
      _realtime == [])
_regimes = {pid: volatility_regime(v) for pid, v in _products.items()}
check(f"...so every shipped plugin is in the NORMAL regime {_regimes}",
      set(_regimes.values()) == {"normal"})

# ...and the machinery is still LIVE, graded from a declaration this file makes
# rather than from a plugin. This is the same stimulus AC2 promises a sixth
# supplier: declare REALTIME, get the wide regime, edit nothing in price.py.
_rt_caps = ProviderCapabilities(product=PriceProduct.REALTIME,
                                sources=frozenset({Component.ENERGY}))
check(f"the widening path still works from a declaration alone "
      f"({volatility_regime(_rt_caps.product)}), which is why §9.2 row 2 "
      f"having no occupant is not the same as it having no mechanism",
      volatility_regime(_rt_caps.product) == "wide"
      and volatility_regime(PriceProduct.DAY_AHEAD) == "normal")
check(f"...and it is the PRODUCT that carries it: uncertainty_of() answers "
      f"{sorted(u.value for u in uncertainty_of(PriceProduct.REALTIME))} for "
      f"real-time and nothing for day-ahead, and widen_factor() turns that "
      f"into {widen_factor(uncertainty_of(PriceProduct.REALTIME))}",
      uncertainty_of(PriceProduct.REALTIME)
      == frozenset({Uncertainty.REALTIME_PRODUCT})
      and uncertainty_of(PriceProduct.DAY_AHEAD) == frozenset()
      and widen_factor(uncertainty_of(PriceProduct.REALTIME)) == 2.5
      and widen_factor(uncertainty_of(PriceProduct.DAY_AHEAD)) == 1.0)


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_columnar_json(body, expect_unit="EUR / MWh",
                                                 scale=0.001,
                                                 provisional_default=False)


# ⚠️ THIS CLASS IS ALSO THE ONLY LIVE DECLARER OF `PriceProduct.REALTIME` IN
# THE TREE (TARIFF-07), which makes the widened-threshold path a thing this
# suite exercises end to end - registry, series, regime - and not merely a
# constant section 5 reads. Deleting it would strand the seam twice over.
_sixth = load("_test_sixth_supplier",
              transport=canned(columnar_feed(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(columnar_feed([]))),
    # ⚠️ THE ROW SHAPE IS A FAILURE FOR THIS PLUGIN, NOT A SECOND DIALECT. If
    # `_points_from_columnar_json()` ever grows a sniff for `data`, this case
    # goes from no_data to a series and TARIFF-08's scope rule is broken by a
    # check rather than by a review.
    "the body is the OTHER shape": load(
        "generic_dayahead",
        transport=canned(row_shape_body("price_eur_mwh", EPEX_EUR_MWH))),
    "the provider id is unknown": load("no_such_supplier"),
    "the provider id is RETIRED": load("eneco_eddy"),
    "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 = columnar_feed([50.0 + i for i in range(25)],
                          start=datetime(2026, 10, 24, 22, tzinfo=UTC))
_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(columnar_feed(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(columnar_feed(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"))

# ⚠️ TARIFF-07: A RETIRED ID IS REFUSED BY A DIFFERENT MESSAGE, AND EVERY CHECK
# HERE GRADES THE REASON RATHER THAN THE VERDICT. The failure this section
# exists to catch is NOT "a retired id starts" - it never will, because it is
# absent from PROVIDERS and the unknown-id branch would catch it. It is the
# retired branch COLLAPSING INTO the unknown-id one: same refusal, same exit
# code, and an operator told they mistyped a name they chose from a dropdown
# this add-on offered them. The two messages are asserted to DIFFER and each
# needle is asserted to be absent from the other, which is MUTATION.md's
# "grade the reason, not the verdict" applied to two rungs that answer one
# stimulus.
_unknown = validate_startup_config(vbase(price_provider="nordpool"),
                                   -5000, 5000)
_retired = {pid: validate_startup_config(vbase(price_provider=pid),
                                         -5000, 5000)
            for pid in sorted(RETIRED_PROVIDERS)}
check(f"every retired id is refused at startup {sorted(_retired)}",
      len(_retired) == 3 and all(len(v) == 1 for v in _retired.values()))
check("...and NOT by the unknown-id message, which would tell an operator they "
      "mistyped an id this add-on offered them and accepted for months",
      all("not a shipped plugin" not in v[0] for v in _retired.values())
      and "WITHDRAWN" not in _unknown[0])
check("...and each message NAMES ITS OWN ID, so the operator can tell it is "
      "about their configuration and not a generic notice",
      all(repr(pid) in v[0] for pid, v in _retired.items()))
check("...and (a) says the plugin never fetched a price, naming the endpoint "
      "that does not exist",
      all("NEVER FETCHED A PRICE" in v[0] for v in _retired.values())
      and "api.engie.be/dynamic/be" in _retired["engie_dynamic"][0]
      and "api.totalenergies.be/prices/dynamic"
          in _retired["totalenergies_dynamic"][0]
      and "api.eneco.be/eddy/prices" in _retired["eneco_eddy"][0])
check("...and (b) says what the site has ACTUALLY been running instead - the "
      "half §11.4 made invisible",
      all("SILENT" in v[0] and "static_profile" in v[0]
          for v in _retired.values()))
check("...and (c) says what to do, naming a live plugin and refusing to "
      "pretend an alias could exist",
      all("generic_dayahead" in v[0] and "manual_csv" in v[0]
          and "NO alias" in v[0] for v in _retired.values()))

# ⚠️ AND THE SENTENCE CHANGES WITH THE SITE. "You have been on the static
# profile" is FALSE where no profile was ever set, and that site is in a WORSE
# state - no tariff at all - so the message must not flatten the two. Graded by
# asserting the two renderings DIFFER, because a check that only looked for
# "static_profile" would pass on a message that says the same thing to both.
_PROF = ('{"periods": [{"from":"00:00","to":"24:00","days":"all",'
         '"eur_per_kwh":0.31}]}')
_with_profile = validate_startup_config(
    vbase(price_provider="eneco_eddy", static_profile=_PROF), -5000, 5000)[0]
_without = _retired["eneco_eddy"][0]
check("a retired id on a site WITH a static profile is told it has been "
      "running that profile",
      "the static day/night tariff in static_profile" in _with_profile)
check("...and a site WITHOUT one is told it has been running nothing at all, "
      "which is a different and worse state",
      "NOTHING AT ALL" in _without
      and "the static day/night tariff in static_profile" not in _without)
check("...so the two renderings are not the same sentence",
      _with_profile != _without)

# ⚠️ 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="generic_dayahead", 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()
print("13. TARIFF-10 - a configured provider is FETCHED, and trading is still "
      "its own switch")
# WHAT SHIPPED, AND WHY THIS SECTION IS NOT A CHECK ON AN `if`. `amain()` read
#
#     if arbitrage.enabled(opts):
#         tasks.append(asyncio.create_task(controller.run_tariff_refresh()))
#
# `run_tariff_refresh` is the only caller of `resolve_tariff()`, which is the
# only caller of `price_history.carry()`, and `arbitrage_enabled` ships FALSE.
# So on a shipped install with a provider configured NOTHING was fetched, no
# plan was resolved, /data/price_history.json was never created, the *Price
# history* row read `empty` for the life of the install, and TARIFF-09's
# short-horizon warning could not fire because no fetch happened to warn about.
# Measured over a 40 h two-instance soak: zero feed requests and
# `present: false` across all 802 samples.
#
# ⚠️ A CHECK THAT PINS THE `if` WOULD NOT HAVE CAUGHT IT, because the `if` was
# EXACTLY WHAT SOMEBODY MEANT TO WRITE. What was wrong is the SET OF
# CONFIGURATIONS FROM WHICH A FETCH CAN BE REACHED, so that is what is graded:
# the guards in `amain()` are EVALUATED against real option dicts (whatever they
# are spelled as), the call graph under `app/` is walked to see whether the
# tasks those guards start can arrive at `resolve_tariff` (wherever it has been
# moved to), and then the fetch is DRIVEN and the requests counted, because a
# reachability graph is an argument and a counted request is an observation.


# -- the predicate the gate is built on, before anything is wired to it -------
# ⚠️ ONE SPELLING, AND IT NOW HAS THREE READERS. `str(price_option(opts,
# "price_provider") or "off").strip()` was written out TWICE - in
# `load_prices()` and in `static_tariff.resolve_tariff()` - and main.py's
# startup gate would have been a third. "Is a provider configured" is now a
# DECISION rather than a detail, because it is what starts a background task,
# so the three readers have to agree about the edge cases and `provider_id()`
# is the one place they can. (`validate.py` keeps its own reader on purpose: it
# validates the RAW options blob and must not route through the defaults.)
check("`provider_configured()` is False on the shipped default, on an explicit "
      "'off', and on a key that is absent, empty or None - `price_option()` "
      "maps the last three onto the default and none of them may read as a "
      "configured feed",
      not any(_price.provider_configured(o) for o in
              ({}, {"price_provider": "off"}, {"price_provider": ""},
               {"price_provider": None})))
check("...and True on each shipped plugin id, so it is a predicate and not a "
      f"constant {sorted(PROVIDERS)}",
      all(_price.provider_configured({"price_provider": pid})
          for pid in PROVIDERS))
# ⚠️ WHITESPACE IS THE EDGE THE `.strip()` IS FOR, AND NOTHING GRADED IT. The
# Supervisor's `list(off|...)` schema cannot produce `"  off  "`, but
# `validate_startup_config()` is also sim/inject.py's entry point and takes any
# dict at all - and dropping the `.strip()` left all four price suites green
# (measured, -8 B) until this pair was written.
check("...and a padded id is normalised on both sides of the decision: "
      "'  off  ' is still off, and ' generic_dayahead ' is still that plugin",
      _price.provider_id({"price_provider": "  off  "}) == "off"
      and not _price.provider_configured({"price_provider": "  off  "})
      and _price.provider_id({"price_provider": " generic_dayahead "})
          == "generic_dayahead"
      and _price.provider_configured(
          {"price_provider": " generic_dayahead "}))
check("...and `load_prices()` answers `no_data` for exactly the dicts "
      "`provider_configured()` calls unconfigured, so the startup gate and the "
      "fetch cannot disagree about what 'a provider is set' means",
      all((run(load_prices(o, datetime(2026, 1, 1, tzinfo=UTC),
                           datetime(2026, 1, 2, tzinfo=UTC),
                           transport=None)).kind == "no_data")
          == (not _price.provider_configured(o))
          for o in ({}, {"price_provider": "off"},
                    {"price_provider": "  off  "},
                    {"price_provider": "manual_csv",
                     "price_csv_path": CSV_PATH})))


def _coroutine_name(node):
    """`controller.run_tariff_refresh()` -> `run_tariff_refresh`."""
    if isinstance(node, ast.Call):
        if isinstance(node.func, ast.Attribute):
            return node.func.attr
        if isinstance(node.func, ast.Name):
            return node.func.id
    return ast.unparse(node)


def _task_sites(node, guards=()):
    """Every `create_task(...)` under `node`, with the `if`s that guard it.

    The guard is kept as SOURCE and evaluated later against an options dict, so
    this reports what the tree says rather than what any one ticket expected to
    find there. An `else` branch is recorded as `not (<test>)`.
    """
    out = []
    for child in ast.iter_child_nodes(node):
        if isinstance(child, ast.If):
            test = ast.unparse(child.test)
            for stmt in child.body:
                out += _task_sites(stmt, guards + (test,))
            for stmt in child.orelse:
                out += _task_sites(stmt, guards + (f"not ({test})",))
            continue
        if isinstance(child, ast.Call) and \
                ast.unparse(child.func).endswith("create_task"):
            if child.args:
                out.append((guards, _coroutine_name(child.args[0])))
            continue
        out += _task_sites(child, guards)
    return out


def _call_graph(app_dir):
    """name -> the names it calls, over every module in `app/`.

    ⚠️ IT IS KEYED BY NAME AND THEREFORE OVER-APPROXIMATES, AND THAT IS THE
    SAFE DIRECTION HERE. Two functions called `run` in two modules are one node;
    a method call on an unrelated object joins the same node. So a `reachable`
    answer is an upper bound and a `NOT reachable` answer is the strong one -
    which is why every positive row below is also DRIVEN, with the requests
    counted, rather than being left to this graph (MUTATION.md habit 13).
    """
    graph = {}
    for name in sorted(os.listdir(app_dir)):
        if not name.endswith(".py"):
            continue
        tree = ast.parse(open(os.path.join(app_dir, name),
                              encoding="utf-8").read())
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            called = set()
            for inner in ast.walk(node):
                if isinstance(inner, ast.Call):
                    fn = inner.func
                    if isinstance(fn, ast.Attribute):
                        called.add(fn.attr)
                    elif isinstance(fn, ast.Name):
                        called.add(fn.id)
            graph.setdefault(node.name, set()).update(called)
    return graph


def _reaches(graph, start, target):
    """Can `start` arrive at `target` through the call graph?"""
    seen, stack = set(), [start]
    while stack:
        cur = stack.pop()
        if cur in seen:
            continue
        seen.add(cur)
        for nxt in graph.get(cur, ()):
            if nxt == target:
                return True
            stack.append(nxt)
    return False


def _started(sites, opts):
    """[(coroutine, guards, verdict)] for `opts`. Verdict True/False/'?'.

    `'?'` is a guard this cannot decide from the options alone - `p1_source is
    not None` is one - and it is reported rather than assumed either way. A
    fetch-reaching task whose guard is undecidable would make every row below
    vacuous, so that is a check of its own.
    """
    out = []
    for guards, name in sites:
        verdict = True
        for guard in guards:
            try:
                answer = bool(eval(guard, vars(_main), {"opts": opts}))
            except NameError:
                verdict = "?"
                break
            if not answer:
                verdict = False
                break
        out.append((name, guards, verdict))
    return out


_MAIN_TREE = derived(lambda: ast.parse(open(os.path.join(_APP, "main.py"),
                                            encoding="utf-8").read()))
_AMAIN = derived(lambda: next(n for n in ast.walk(_MAIN_TREE)
                              if isinstance(n, ast.AsyncFunctionDef)
                              and n.name == "amain"))
_SITES = derived(lambda: _task_sites(_AMAIN))
_GRAPH = derived(lambda: _call_graph(_APP))
_FETCH_ENTRY = "resolve_tariff"

print(f"    create_task sites found in amain(): "
      f"{[(n, list(g)) for g, n in _SITES]}")
print(f"    call graph: {len(_GRAPH)} named functions across app/")
print("    ⚠️ what this instrument CANNOT see, largest hole first: a coroutine "
      "started by anything other than `create_task` (gather, ensure_future, a "
      "thread, a timer); a guard that is not an enclosing `if` in amain() (a "
      "while, a try/except, an early return above the site, or a refusal "
      "INSIDE the coroutine); any module outside app/; and - because the graph "
      "is keyed by NAME - two same-named functions as one node, which makes "
      "every `reachable` answer an UPPER bound.")

# -- the instrument is calibrated before anything is graded with it (habit 1) --
check(f"the scanner finds amain()'s create_task sites at all - it found "
      f"{len(_SITES) if not isinstance(_SITES, _Raised) else _SITES}, and at "
      f"zero every row below would be vacuously true",
      not isinstance(_SITES, _Raised) and len(_SITES) >= 2)
check(f"...and the call graph can tell reaching from not-reaching: "
      f"`run_tariff_refresh` arrives at {_FETCH_ENTRY!r} and `run_control` - "
      f"the control tick, which must never await a fetch - does not. A graph "
      f"that answered the same for both would pass every row below however the "
      f"gate was wired",
      _reaches(_GRAPH, "run_tariff_refresh", _FETCH_ENTRY)
      and not _reaches(_GRAPH, "run_control", _FETCH_ENTRY))


def _fetchers(opts):
    """The coroutines amain() STARTS, for `opts`, that can reach the feed."""
    return sorted(n for n, _, v in _started(_SITES, opts)
                  if v is True and _reaches(_GRAPH, n, _FETCH_ENTRY))


def _undecidable_fetchers(opts):
    return sorted(n for n, _, v in _started(_SITES, opts)
                  if v == "?" and _reaches(_GRAPH, n, _FETCH_ENTRY))


# The four configurations. `_T10_BASE` carries no price keys at all, so the
# shipped defaults in PRICE_OPTION_DEFAULTS are what answer - which is the
# install the defect was measured on.
_T10_BASE = {"setpoint_entity": "number.sp", "max_w": 5000,
             "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
             "meter_source": "off", "maintenance_charge_w": 2000,
             "maintenance_discharge_w": 2000}
_T10_PROFILE = json.dumps({"periods": [
    {"name": "night", "days": "all", "from": "22:00", "to": "07:00",
     "eur_per_kwh": 0.20},
    {"name": "day", "days": "all", "from": "07:00", "to": "22:00",
     "eur_per_kwh": 0.284}]})
_T10_CASES = (
    ("shipped default: no provider, no trading", {}, False),
    ("PROVIDER configured, trading OFF (the defect)",
     {"price_provider": "generic_dayahead"}, True),
    ("PROVIDER configured, trading ON",
     {"price_provider": "generic_dayahead", "arbitrage_enabled": True}, True),
    ("no provider, trading ON on a static profile",
     {"arbitrage_enabled": True, "static_profile": _T10_PROFILE}, True),
)
print("    startup reachability of resolve_tariff(), by configuration:")
for _label, _over, _want in _T10_CASES:
    print(f"      {'reachable' if _fetchers(dict(_T10_BASE, **_over)) else '-        '}"
          f"  {_label}")

check(f"the shipped defaults really do leave `arbitrage_enabled` off "
      f"({PRICE_OPTION_DEFAULTS['arbitrage_enabled']!r}) and `price_provider` "
      f"off ({PRICE_OPTION_DEFAULTS['price_provider']!r}) - the precondition "
      f"every row below varies from, and if a later card changes either "
      f"default these rows stop meaning what they say. Section 1 is what pins "
      f"this dict to config.yaml",
      PRICE_OPTION_DEFAULTS["price_provider"] == "off"
      and PRICE_OPTION_DEFAULTS["arbitrage_enabled"] is False
      and not _arb.enabled(dict(_T10_BASE))
      and not _price.provider_configured(dict(_T10_BASE)))
check(f"AC1: with a `price_provider` set and `arbitrage_enabled` at its shipped "
      f"default, the startup path REACHES the feed "
      f"{_fetchers(dict(_T10_BASE, price_provider='generic_dayahead'))}. This "
      f"is the shipped defect: it read `[]`, so nothing was ever fetched",
      _fetchers(dict(_T10_BASE, price_provider="generic_dayahead")) != [])
check(f"AC2: with `price_provider: off` and trading off - the true shipped "
      f"default - NO task reaches the feed "
      f"{_fetchers(dict(_T10_BASE))}, so an install that asked for no prices "
      f"acquires no periodic work from this card",
      _fetchers(dict(_T10_BASE)) == [])
check("...and the guard on the fetch-reaching task is decidable from the "
      "OPTIONS alone. A gate moved onto a local variable would make every row "
      "in this section vacuous rather than red",
      _undecidable_fetchers(dict(_T10_BASE, price_provider="generic_dayahead"))
      == [])
# ⚠️ THE `or` IS LOAD-BEARING AND THIS IS THE ROW THAT SAYS SO. `price_provider:
# off` + a `static_profile` + `arbitrage_enabled: true` is a configuration
# validate.py ACCEPTS and that trades (measured: ARBITRAGE_CHARGE -5000 W at
# `cycle_cost_threshold_eur: 0.05` on a 0.20/0.284 day/night profile). Gating
# the refresh on the provider ALONE would leave that site's `tariff_plan` at
# None forever and silently disarm a strategy the UI says is running.
_T10_OLD = {lbl: _arb.enabled(dict(_T10_BASE, **over))
            for lbl, over, _ in _T10_CASES}
_T10_NEW = {lbl: bool(_fetchers(dict(_T10_BASE, **over)))
            for lbl, over, _ in _T10_CASES}
print(f"    old gate (arbitrage.enabled alone) -> {list(_T10_OLD.values())}")
print(f"    shipped gate                       -> {list(_T10_NEW.values())}")
check(f"the gate is a STRICT WIDENING: every configuration that started the "
      f"refresh before still starts it, so nothing was taken away from a live "
      f"install {[l for l in _T10_OLD if _T10_OLD[l] and not _T10_NEW[l]]}",
      all(_T10_NEW[l] for l in _T10_OLD if _T10_OLD[l]))
check("...and it is a widening rather than a no-op: at least one configuration "
      "fetches now that did not before, which is the whole ticket",
      any(_T10_NEW[l] and not _T10_OLD[l] for l in _T10_OLD))


# -- and the fetch is DRIVEN, because a graph is an argument and a counted ----
# -- request is an observation (habit 4) -------------------------------------
_CAPTURE = derived(lambda: json.load(open(
    os.path.join(HERE, "scenarios",
                 "energy-charts-be-price-2026-09-09_2026-09-10-twoday.json"),
    encoding="utf-8")))


def _restamped_body(now):
    """The COMMITTED capture, with only its instants moved onto `now`.

    ⚠️ THE SHAPE AND THE PRICES ARE THE CAPTURE'S (TARIFF-08, habit 19). What
    is synthetic is the CLOCK and nothing else: `run_tariff_refresh` reads the
    wall clock and there is no seam to inject one, so a 2026-09-09 capture
    would fall outside every horizon it asks for. Section 2b replays the
    capture unmodified against the parser; this one is grading the WIRING.
    """
    base = int(now.timestamp()) // 3600 * 3600
    doc = dict(_CAPTURE)
    doc["unix_seconds"] = [base + 900 * i
                           for i in range(len(_CAPTURE["price"]))]
    return json.dumps(doc).encode()


class _T10Hass:
    async def limits(self, entity):
        return (-5000.0, 5000.0, 1.0)

    async def number(self, entity, invert=False):
        return {"sensor.grid": 0.0, "sensor.soc": 55.0,
                "sensor.batt": 0.0}.get(entity)

    async def set_number(self, entity, value):
        return True

    async def call(self, domain, service, data):
        return True


class _T10Store:
    data = {}

    def set(self, *a):
        pass

    def get_time(self, key):
        return None


class _T10Pub:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


def _drive_startup(over):
    """Start exactly what amain() would start for `opts`, and watch the wire.

    ⚠️ THE TASK IS STARTED ONLY IF THE SHIPPED GUARD SAYS SO. That is the whole
    point: the coroutine is not called because this file knows its name, it is
    called because `_fetchers()` evaluated amain()'s own guard against these
    options. A gate that goes back to `arbitrage.enabled(opts)` leaves the
    coroutine uncalled and the request count at zero.

    ⚠️ AND IT RUNS UNDER A DEADLINE (habit 5). The refresh loop is infinite by
    design; the failure mode of a mutation that never resolves a plan is a HANG,
    which reads like flaky infrastructure rather than like a named check.
    """
    opts = dict(_T10_BASE, **over)
    tmp = tempfile.mkdtemp(prefix="tariff10-")
    _tmpdirs_t10.append(tmp)
    store_path = os.path.join(tmp, "price_history.json")
    seen = []

    async def _counting_get(url, timeout_s):
        seen.append(url)
        return _restamped_body(datetime.now(timezone.utc))

    real_get = _price.https_get
    _price.https_get = _counting_get
    try:
        async def _go():
            stop = asyncio.Event()
            ctl = _main.Controller(opts, _T10Hass(), _T10Store(), _T10Pub(),
                                   stop=stop,
                                   peak_db_path=os.path.join(tmp, "tariff.db"),
                                   price_history_path=store_path)
            names = _fetchers(opts)
            if "run_tariff_refresh" not in names:
                # Nothing amain() starts reaches the feed. Anything ELSE that
                # does is a task this harness cannot drive, and it is a FAILURE
                # rather than a shrug - see the check below.
                return ctl, names
            task = asyncio.create_task(ctl.run_tariff_refresh())
            for _ in range(250):                      # ~5 s ceiling
                if ctl.tariff_at is not None:
                    break
                await asyncio.sleep(0.02)
            ctl.stopping = True
            stop.set()
            await asyncio.wait_for(task, timeout=5)
            return ctl, names
        ctl, names = asyncio.run(_go())
    finally:
        _price.https_get = real_get
    try:
        n_points = len(list(ctl.tariff_plan.read.points))
    except Exception:                                 # noqa: BLE001 - Absence
        n_points = 0
    return {"requests": seen, "started": names,
            "store_present": os.path.exists(store_path),
            "points": n_points,
            "history": ctl.price_history.words}


_tmpdirs_t10 = []
_on = derived(lambda: _drive_startup({"price_provider": "generic_dayahead"}))
_off = derived(lambda: _drive_startup({}))
_static_armed = derived(lambda: _drive_startup(
    {"arbitrage_enabled": True, "static_profile": _T10_PROFILE}))
print(f"    provider + trading OFF -> {len(_on['requests'])} request(s), "
      f"store written {_on['store_present']}, {_on['points']} point plan, "
      f"Price history row: {_on['history'][:70]!r}")
print(f"    no provider            -> {len(_off['requests'])} request(s), "
      f"store written {_off['store_present']}, "
      f"Price history row: {_off['history'][:40]!r}")

check(f"AC1, DRIVEN: a shipped-default install with a provider configured "
      f"actually ASKS THE FEED - {len(_on['requests'])} request(s), "
      f"{_on['requests'][:1]}. The soak measured ZERO here over 40 hours",
      len(_on["requests"]) >= 1)
check(f"...and a plan RESOLVES from it: {_on['points']} points, against the "
      f"{len(_CAPTURE['price'])} the capture publishes",
      _on["points"] == len(_CAPTURE["price"]))
check(f"...and /data/price_history.json is WRITTEN - present={_on['store_present']}. "
      f"The soak measured `present: false` on all 802 samples",
      _on["store_present"] is True)
check(f"...and the *Price history* status row POPULATES rather than reading "
      f"`empty` forever: {_on['history'][:60]!r}",
      isinstance(_on["history"], str)
      and not _on["history"].startswith("empty")
      and not _on["history"].startswith("unavailable"))
check(f"AC2, DRIVEN: `price_provider: off` issues NO request and creates no "
      f"store - {len(_off['requests'])} request(s), "
      f"present={_off['store_present']}",
      _off["requests"] == [] and _off["store_present"] is False)
check(f"...and neither does the one configuration that keeps its task without "
      f"a provider - trading armed on a static profile makes "
      f"{len(_static_armed['requests'])} request(s), because "
      f"`resolve_tariff()` answers a provider-off site out of the static "
      f"profile without opening a socket. NOBODY acquires a periodic HTTPS "
      f"call from this card",
      _static_armed["requests"] == []
      and "run_tariff_refresh" in _static_armed["started"])
check(f"...and no OTHER startup task can reach the feed - if one appears, this "
      f"harness cannot drive it and every driven row above stops covering the "
      f"startup path {_on['started']}",
      _on["started"] == ["run_tariff_refresh"])


# -- AC3, and it is the property this card is most dangerous to: ARBITRAGE -----
# -- OFF STILL MEANS NO TRADE, EVER. Graded by DRIVING `Controller.cycle()` ---
# -- with a provider plan resolved - the state TARIFF-10 newly creates - and ---
# -- moving exactly one option (habit 7). ------------------------------------
#
# ⚠️ THE BASELINE IS THE HALF THAT MAKES THIS A CHECK. "No trade" is satisfied
# by a fixture in which nothing could have traded anyway - a flat series, a SoC
# outside the bounds, a peak guard that never ran - and a check like that stays
# green however the gate is wired. So the SAME plan, the SAME SoC, the SAME P1
# history and the SAME threshold are run with `arbitrage_enabled: true`, and
# that run must COMMAND THE BATTERY. If the armed row stops trading, the
# unarmed row stops being evidence and both go red together.


def _t10_sample(when, net_w):
    """One P1 telegram through the shipped derivation."""
    imp, exp = split_signed(net_w)
    return make_sample("ha_signed", imp, exp, phases=1,
                       ingest_ts=when, ingest_mono=when.timestamp())


class _T10TradeHass(_T10Hass):
    def __init__(self):
        self.writes = []

    async def set_number(self, entity, value):
        self.writes.append(float(value))
        return True


def _t10_plan(now, cheap_now):
    """A 24 h FIRM series whose cheapest hour is now, or whose dearest is.

    Firm, because refusal 3 drops provisional hours from the trade set and a
    plan of carried hours cannot trade whatever the gate says - which would
    make the armed baseline below vacuous exactly as a flat series would.
    """
    top = now.replace(minute=0, second=0, microsecond=0)
    lo, hi = 0.05, 0.60
    return TariffPlan(
        PriceRead.series(
            [PricePoint(start=top + timedelta(hours=i),
                        eur_per_kwh=(lo if (i < 3) == cheap_now else hi),
                        provisional=False)
             for i in range(24)],
            PriceProduct.DAY_AHEAD, "generic_dayahead", now),
        None, "provider")


def _t10_trade_cycle(armed, cheap_now):
    """One real `Controller.cycle()` on a resolved provider plan."""
    tmp = tempfile.mkdtemp(prefix="tariff10t-")
    _tmpdirs_t10.append(tmp)
    opts = dict(_T10_BASE, meter_source="ha_dsmr", meter_entity="sensor.grid",
                heartbeat_s=10, price_provider="generic_dayahead",
                cycle_cost_threshold_eur=0.10, arbitrage_enabled=armed)

    async def _go():
        hass = _T10TradeHass()
        ctl = _main.Controller(opts, hass, _T10Store(), _T10Pub(),
                               stop=asyncio.Event(),
                               peak_db_path=os.path.join(tmp, "tariff.db"),
                               price_history_path=os.path.join(tmp, "ph.json"))
        ctl.auto = True
        ctl.dev_min, ctl.dev_max = -5000.0, 5000.0
        now = datetime.now(timezone.utc)
        # ⚠️ A WHOLE QUARTER OF TELEGRAMS, AND FIVE MINUTES IS NOT ENOUGH.
        # Without any, `p1_enabled` is on, `inputs_ok` is False and cycle()
        # returns before it reaches the strategy at all. With only five
        # minutes, whether the quarter is COVERED depends on where the wall
        # clock happens to sit inside it: run late in a quarter and the guard
        # is blind, engages conservatively, and `decide()` refuses for §9.1's
        # reason instead of trading - which would make the armed baseline
        # below vacuous on some runs and not others. Measured: the same four
        # rows came back `peak guard is holding` at one time of day and
        # `ARBITRAGE_CHARGE` at another. 900 s at the shipped 5 s telegram
        # cadence covers any quarter from any starting phase.
        for k in range(181):
            ctl.p1.submit(_t10_sample(now - timedelta(seconds=5 * (180 - k)),
                                      300.0))
        ctl.tariff_plan, ctl.tariff_at = _t10_plan(now, cheap_now), now
        await ctl.read_inputs()
        ctl.tariff_plan, ctl.tariff_at = _t10_plan(now, cheap_now), now
        await ctl.cycle()
        return ctl, hass
    ctl, hass = asyncio.run(_go())
    return {"arb": None if ctl.arb is None else ctl.arb.action,
            "target": ctl.target, "reason": ctl.reason,
            "writes": hass.writes,
            "guard": None if ctl.peak_guard is None else ctl.peak_guard.engaged}


_T10_TRADE = derived(lambda: {(a, c): _t10_trade_cycle(a, c)
                              for a in (False, True) for c in (True, False)})
for _key in sorted(_T10_TRADE, key=str):
    _r = _T10_TRADE[_key]
    print(f"    arbitrage_enabled={str(_key[0]):5s} cheapest hour is now="
          f"{str(_key[1]):5s} -> arb={str(_r['arb']):16s} "
          f"commanded {_r['target']:+.0f} W, reason {_r['reason'][:40]!r}")

# ⚠️ THE PRECONDITION, AND IT IS THE MASKING HAZARD IN THIS FIXTURE (habit 1).
# `decide()` refuses outright while the peak guard is engaged, §9.1 outranking
# §10.1 - so a fixture whose guard is holding answers "no trade" in EVERY row,
# the armed one included, and the AC3 rows below would grade the guard rather
# than the gate. It is asserted, not assumed.
check(f"the peak guard is CLEAR in all four rows, so what refuses a trade "
      f"below is the arbitrage gate and not §9.1's interlock "
      f"{ {k: _T10_TRADE[k]['guard'] for k in sorted(_T10_TRADE, key=str)} }",
      all(_T10_TRADE[k]["guard"] is False for k in _T10_TRADE))
check("the ARMED baseline actually trades on the plan TARIFF-10 now resolves - "
      "it charges in the cheap hour and discharges in the dear one. Without "
      "this row, 'trading off never trades' would be satisfied by a fixture "
      "that could not have traded either way",
      _T10_TRADE[(True, True)]["arb"] == "ARBITRAGE_CHARGE"
      and _T10_TRADE[(True, False)]["arb"] == "ARBITRAGE_DISCHARGE")
check(f"...and the site is actually COMMANDED there, which is the only "
      f"observable that moves energy: {_T10_TRADE[(True, True)]['target']:+.0f} W "
      f"on the cheap hour against "
      f"{_T10_TRADE[(True, False)]['target']:+.0f} W on the dear one",
      _T10_TRADE[(True, True)]["target"] < -1000.0
      and _T10_TRADE[(True, False)]["target"] > 1000.0)
check(f"AC3: with `arbitrage_enabled: false` and the SAME resolved provider "
      f"plan, the strategy is never consulted and NOTHING trades - "
      f"arb={_T10_TRADE[(False, True)]['arb']!r} / "
      f"{_T10_TRADE[(False, False)]['arb']!r} in both directions",
      _T10_TRADE[(False, True)]["arb"] is None
      and _T10_TRADE[(False, False)]["arb"] is None)
check(f"...and the command is the LOOP's, not a trade that merely went "
      f"unnamed: {_T10_TRADE[(False, True)]['target']:+.0f} W / "
      f"{_T10_TRADE[(False, False)]['target']:+.0f} W, reason "
      f"{_T10_TRADE[(False, True)]['reason'][:30]!r} - the same small "
      f"self-consumption figure on both plans, where the armed rows differ by "
      f"10 kW between them",
      all(abs(_T10_TRADE[(False, c)]["target"]) < 1000.0
          and "arbitrage" not in _T10_TRADE[(False, c)]["reason"]
          for c in (True, False)))
check("...and the trading gate is untouched by this card: it is still "
      "`arbitrage.enabled()` and it is still the only thing the price "
      "strategy is gated on - the four rows above differ in that option and "
      "in nothing else",
      _arb.enabled({"arbitrage_enabled": True})
      and not _arb.enabled({})
      and not _arb.enabled({"price_provider": "generic_dayahead"}))


# -- AC4: the four surfaces that DESCRIBE this, pinned as a PAIR with it ------
# ⚠️ THIS IS A TEXT CHECK AND IT IS WEAK ON PURPOSE (test_translations.py §3's
# argument). It cannot know a sentence is TRUE. What it CAN do is refuse to let
# the code and the prose move independently: the left side of each pairing is
# DRIVEN above, the right side is read off the file, and the two must agree. It
# shipped as False against True - the docs described Option A for as long as
# both existed - and an overclaim in a doc is a defect in this repository.
_T10_DOCS = derived(lambda: open(os.path.join(HERE, "DOCS.md"),
                                 encoding="utf-8").read())
_T10_CFG = derived(lambda: open(os.path.join(HERE, "config.yaml"),
                                encoding="utf-8").read())
_T10_TR = derived(lambda: {lang: open(os.path.join(HERE, "translations",
                                                   f"{lang}.yaml"),
                                      encoding="utf-8").read()
                           for lang in ("en", "nl")})
_T10_FETCHES = bool(_on["requests"])
_T10_DOCS_SAYS = "selecting a provider is what starts the fetch" \
    in _T10_DOCS.lower()
check(f"AC4: the BEHAVIOUR and DOCS.md agree about what a configured provider "
      f"does on a shipped-default install - it fetches = {_T10_FETCHES}, DOCS "
      f"says it fetches = {_T10_DOCS_SAYS}. These two are the pair that has to "
      f"move together",
      _T10_FETCHES == _T10_DOCS_SAYS)
# The claim that had rotted in the OTHER direction: OPT-04 shipped the
# consumer, so "nothing consumes the price series yet" had been false for a
# release before this card made the first half false too.
_T10_STALE = {"config.yaml": "consumes\n  # the series yet" in _T10_CFG
              or "nothing consumes the price series" in _T10_CFG,
              "en.yaml": "nothing consumes the price series yet"
              in _T10_TR["en"],
              "nl.yaml": "er is nog niets dat de prijsreeks gebruikt"
              in _T10_TR["nl"]}
check(f"...and no surface still says the price series has no consumer - untrue "
      f"since OPT-04, and it sat beside the claim this card is fixing in all "
      f"three files {[k for k, v in _T10_STALE.items() if v]}",
      not any(_T10_STALE.values()))
check("...and BOTH translations tell an operator which switch does which - "
      "the provider option says it is what starts the fetch, and the trading "
      "option says turning it off does not turn the feed off. One language "
      "updated and not the other is how this surface rots",
      "STARTS THE PRICE FETCH" in _T10_TR["en"]
      and "DOES NOT TURN THE PRICE FEED OFF" in _T10_TR["en"]
      and "HET OPHALEN VAN PRIJZEN START" in _T10_TR["nl"]
      and "ZET DE PRIJSFEED NIET UIT" in _T10_TR["nl"])


# --------------------------------------------------------------------------- #
print()
print("14. TARIFF-14 - the point whose OWN interval covers the instant")
# ⚠️ THE RESOLUTION IS THE WHOLE SECTION, AND ITS ABSENCE IS WHY THE DEFECT
# SHIPPED. Every price fixture in this repository before TARIFF-08 was the
# invented HOURLY row shape, and on an hourly series `s <= u < s + 1 h` and
# "until the next point's start" AGREE AT EVERY INSTANT - so the suite had
# been exercising the one resolution that cannot expose the bug. Section 8
# counts points on the 23- and 25-hour days and never asks `at()` for a
# quarter-hour. Belgium moved to a 15-minute MTU on 2025-10-01 and the live
# day-ahead feed returns 96 points at 900 s.

# ⚠️ BRUSSELS MIDNIGHT, NOT UTC MIDNIGHT, so that every wall-clock time in
# a check name below is the time the series really renders - "00:15" and
# "23:45" are what `canonical()` writes, not an offset away from it.
Q0 = datetime(2026, 1, 15, 0, 0, tzinfo=timezone(timedelta(hours=1)))
HOUR = timedelta(hours=1)


def spaced(gaps, first=0.10, step=0.01, provisional=None, start=Q0):
    """A series whose consecutive gaps are exactly `gaps`, in SECONDS.

    `len(points) == len(gaps) + 1`. Prices ascend by `step`, so a point can be
    named by VALUE as well as by identity and a sweep that compared only
    prices could still tell two points apart.
    """
    n = len(gaps) + 1
    flags = [False] * n if provisional is None else list(provisional)
    pts, t = [], start
    for i in range(n):
        pts.append(PricePoint(brussels(t), round(first + step * i, 6),
                              bool(flags[i])))
        if i < len(gaps):
            t += timedelta(seconds=gaps[i])
    return PriceRead.series(pts, PriceProduct.DAY_AHEAD, "tariff14", start)


def gaps_of(s):
    u = [p.start.astimezone(UTC) for p in s.points]
    return [int((b - a).total_seconds()) for a, b in zip(u, u[1:])]


# The three shapes the acceptance criteria name, plus the two degenerate ones.
Q = spaced([900] * 95)                      # the shipped 96-point Belgian day
H = spaced([3600] * 23)                     # what every older fixture carried
# ⚠️ FCST-04'S STEADY STATE, NOT AN EDGE CASE. Eight firm quarter-hours, then
# the carried HOURLY points the fill adds for the hours nobody published. The
# seam between them is a 900 s gap and the tail is 3600 s.
MIX = spaced([900] * 8 + [3600] * 3,
             provisional=[False] * 9 + [True] * 3)
# The same seam made the END of the series - the last point is the FIRST
# carried hourly one, and the gap behind it is 900 s. This is the fixture the
# tail decision is graded on.
MIX_TAIL = spaced([900] * 8, provisional=[False] * 8 + [True])
ONE = spaced([])

print("14a. the fixtures are not degenerate - asserted before anything else")
check(f"the quarter-hourly fixture is the live shape - 96 points, spacing "
      f"{sorted(set(gaps_of(Q)))} s",
      len(Q.points) == 96 and set(gaps_of(Q)) == {900})
check(f"the hourly fixture is 24 points at {sorted(set(gaps_of(H)))} s - and "
      f"it is a REGRESSION guard, not a witness: the shipped rule and this "
      f"one agree at every instant of it, which is exactly why nothing caught "
      f"this",
      len(H.points) == 24 and set(gaps_of(H)) == {3600})
check(f"...while the MIXED fixture really is mixed, spacing "
      f"{sorted(set(gaps_of(MIX)))} s - the pair the FCST-04 soak measured in "
      f"70 % of samples, so a window derived from ONE global spacing would be "
      f"wrong in the NORMAL steady state",
      set(gaps_of(MIX)) == {900, 3600}
      and set(gaps_of(MIX_TAIL)) == {900}
      and gaps_of(MIX)[-1] == 3600 and gaps_of(MIX_TAIL)[-1] == 900)
check("...and every point of every fixture carries a DISTINCT price, so 'the "
      "covering point' names one point and a wrong answer cannot coincide "
      "with the right one",
      all(len({p.eur_per_kwh for p in s.points}) == len(s.points)
          for s in (Q, H, MIX, MIX_TAIL)))
check("...and the mixed fixtures carry BOTH firm and carried points, which is "
      "what makes them FCST-04's output rather than a spacing exercise",
      all(any(p.provisional for p in s.points)
          and any(not p.provisional for p in s.points)
          for s in (MIX, MIX_TAIL)))

print()
print("14b. the reading the shipped scan got wrong, stated as literals")
# ⚠️ ASSERTED AGAINST TYPED-OUT INDICES AND PRICES, not against a window this
# file recomputed by price.py's own rule. The sweep below is exhaustive and
# the sweep is derived; these four are the anchors that keep it honest.
_q15 = derived(lambda: Q.at(Q0 + timedelta(minutes=15)))
check(f"at(00:15) on the 900 s series is the 00:15 point at 0.11 EUR/kWh - the "
      f"shipped scan answered the 00:00 point at 0.10, because the :00 "
      f"point's HOUR had not run out yet (saw {_q15!r})",
      _q15 is Q.points[1] and derived(lambda: _q15.eur_per_kwh) == 0.11)
_q195 = derived(lambda: Q.at(Q0 + timedelta(minutes=195)))
check(f"at(03:15) is the 03:15 point at 0.23 - the shipped scan answered "
      f"02:30 at 0.20 here, and that is the SECOND failure mode: from the "
      f"second hour on the first point still inside its own hour wins, so the "
      f"error COMPOUNDS with depth into the series (saw {_q195!r})",
      _q195 is Q.points[13] and derived(lambda: _q195.eur_per_kwh) == 0.23)
check("at(00:30) and at(00:45) are their own quarters too, so the answer is "
      "not merely 'one point later'",
      derived(lambda: Q.at(Q0 + timedelta(minutes=30))) is Q.points[2]
      and derived(lambda: Q.at(Q0 + timedelta(minutes=45))) is Q.points[3])
check("on the MIXED series the quarter-hourly half answers per QUARTER and "
      "the carried hourly half answers per HOUR, in one series - at(00:15) is "
      "the 00:15 point and at(02:30) is the 02:00 one",
      derived(lambda: MIX.at(Q0 + timedelta(minutes=15))) is MIX.points[1]
      and derived(lambda: MIX.at(Q0 + timedelta(minutes=150))) is MIX.points[8]
      and derived(lambda: MIX.points[8].start.astimezone(UTC)) ==
      Q0 + timedelta(hours=2))

print()
print("14c. the exhaustive sweep, at 900 s, 3600 s and mixed")


def windows(s):
    """`(start_utc, width)` per point, computed HERE from the gaps.

    Deliberately NOT read off `_Series._cover`: a check built from the table it
    grades compares a thing against itself. The rule is restated - a point runs
    to the next start, capped at an hour; the last point copies the width in
    front of it; a lone point gets an hour.
    """
    u = [p.start.astimezone(UTC) for p in s.points]
    out = []
    for i, t in enumerate(u):
        if i + 1 < len(u):
            g = u[i + 1] - t
        elif i:
            g = t - u[i - 1]
        else:
            g = HOUR
        out.append((t, g if g < HOUR else HOUR))
    return out


def sweep(s):
    """Every probe whose covering point is unambiguous, and what `at()` said."""
    wrong = []
    for i, (t, w) in enumerate(windows(s)):
        for off in (timedelta(0), w / 2, w - timedelta(seconds=1)):
            got = derived(lambda: s.at(t + off))
            if got is not s.points[i]:
                wrong.append(f"{(t + off).isoformat()} -> {got!r}, "
                             f"wanted {s.points[i].eur_per_kwh}")
    return wrong


for _name, _s in (("900 s", Q), ("3600 s", H), ("mixed 900/3600", MIX),
                  ("mixed, 900 s tail", MIX_TAIL)):
    _w = derived(lambda s=_s: sweep(s))
    check(f"{_name}: all {3 * len(_s.points)} probes inside the series return "
          f"the point whose OWN interval covers them "
          f"({_w if _w != [] else 'no misses'})", _w == [])

print()
print("14d. the one-hour cap, which is what keeps FCST-04's holes holes")
# ⚠️ NOT DECORATION. FCST-04 leaves an hour OUT when nothing published it and
# yesterday has no value either, and price_forecast.py's module note names
# `at()` raising Absence for it as the behaviour that sends the caller to the
# static profile (§11.4 for that hour). Uncapped, "until the next point's
# start" would stretch the point before a three-hour hole across the whole
# hole - filling holes with a stale price, which is the defect this module
# exists to refuse, introduced by the fix for a different one.
HOLE = PriceRead.series(
    [PricePoint(brussels(Q0 + timedelta(hours=i)), 0.10 + i / 100.0, False)
     for i in (0, 1, 4, 5)],
    PriceProduct.DAY_AHEAD, "tariff14", Q0)
_in_hole = [Q0 + timedelta(hours=2), Q0 + timedelta(hours=2, minutes=30),
            Q0 + timedelta(hours=3), Q0 + timedelta(hours=3, minutes=59)]
check(f"a three-hour hole stays a hole: all {len(_in_hole)} instants inside it "
      f"raise Absence, while the point BEFORE it still covers its own hour and "
      f"the point after it answers from its own start",
      all(raises(lambda u=u: HOLE.at(u), Absence) for u in _in_hole)
      and derived(lambda: HOLE.at(Q0 + timedelta(hours=1, minutes=59)))
      is HOLE.points[1]
      and derived(lambda: HOLE.at(Q0 + timedelta(hours=4))) is HOLE.points[2])
check(f"...and the cap is stated once, as MAX_POINT_SPAN = "
      f"{_price.MAX_POINT_SPAN}, rather than spelled into the lookup",
      _price.MAX_POINT_SPAN == HOUR)
# ⚠️ AND THE OTHER HALF OF THE HOLE, WHICH IS THE ONE §8.3 READS. FCST-04
# fills a whole hour nobody published with ONE carried point, so a
# quarter-hourly day can carry an hourly carried point in the middle of it.
# The instant inside that hour is covered by the CARRIED point - and the
# shipped scan answered the FIRM quarter 52 minutes earlier, because that was
# the first point still inside its own hour. `_act()`'s `here.provisional`
# guard and `price_forecast._walk()`'s `series.at(u).provisional` both read the
# flag off whatever that lookup returned, so the flag was FIRM on an hour
# nobody published.
SANDWICH = PriceRead.series(
    [PricePoint(brussels(Q0 + timedelta(minutes=15 * i)),
                round(0.10 + i / 1000.0, 6), False) for i in range(44)]
    + [PricePoint(brussels(Q0 + timedelta(hours=11)), 0.500, True)]
    + [PricePoint(brussels(Q0 + timedelta(hours=12, minutes=15 * i)),
                  round(0.20 + i / 1000.0, 6), False) for i in range(4)],
    PriceProduct.DAY_AHEAD, "tariff14", Q0)
_sand_t = Q0 + timedelta(hours=11, minutes=37)


def shipped_at(s, u):
    """The rule TARIFF-14 replaced, as the COUNTERFACTUAL only - nothing is
    graded against it. It is here so the fixture can be SHOWN to separate the
    two rules; a check whose data cannot reach the failure it names grades
    nothing, and on an hourly series these two agree at every instant."""
    uu = u.astimezone(UTC)
    for p in s.points:
        st = p.start.astimezone(UTC)
        if st <= uu < st + HOUR:
            return p
    return None


check("the sandwich fixture is FCST-04's real output - 44 firm quarters, ONE "
      "carried hourly point filling the hour nobody published, then firm "
      "quarters again",
      len(SANDWICH.points) == 49
      and SANDWICH.points[44].provisional is True
      and sum(1 for p in SANDWICH.points if p.provisional) == 1)
_sand_now = derived(lambda: SANDWICH.at(_sand_t))
_sand_old = derived(lambda: shipped_at(SANDWICH, _sand_t))
check(f"an instant inside the carried hour is covered by the CARRIED point, "
      f"so §8.3's flag is read off the hour it belongs to. The shipped scan "
      f"answered the FIRM quarter 52 minutes earlier "
      f"({derived(lambda: _sand_old.start.isoformat())}, provisional="
      f"{derived(lambda: _sand_old.provisional)}), which is a carried hour "
      f"wearing a published hour's flag",
      _sand_now is SANDWICH.points[44]
      and derived(lambda: _sand_now.provisional) is True
      and _sand_old is SANDWICH.points[43]
      and derived(lambda: _sand_old.provisional) is False)

print()
print("14e. the LAST point's window - decided and documented, not inherited")
# There is no next start and §9.3 says nothing about how wide the final
# interval is, so THE LAST POINT IS GIVEN THE SAME SPAN AS THE POINT BEFORE IT
# (one hour when it is the only point). The alternative - a flat one-hour tail
# - is defensible for an hourly series and wrong for the live one: it answers
# 00:30 TOMORROW with today's last price, which is reaching for a neighbour
# across a day boundary.
_ql = Q.points[-1].start.astimezone(UTC)
_hl = H.points[-1].start.astimezone(UTC)
_ml = MIX_TAIL.points[-1].start.astimezone(UTC)
check("the 900 s day's last point covers its own QUARTER: at(+14:59) answers "
      "and at(+15:00) raises, so today's 23:45 price cannot be returned for "
      "00:30 tomorrow",
      derived(lambda: Q.at(_ql + timedelta(minutes=14, seconds=59)))
      is Q.points[-1]
      and raises(lambda: Q.at(_ql + timedelta(minutes=15)), Absence))
check("the hourly series' last point still covers an HOUR, exactly as it "
      "shipped - no static or hourly site changes behaviour",
      derived(lambda: H.at(_hl + timedelta(minutes=59))) is H.points[-1]
      and raises(lambda: H.at(_hl + HOUR), Absence))
check("the MIXED series ending at the firm/carried seam takes the 900 s in "
      "front of it, so its last 45 minutes raise Absence. That is the SAFE "
      "direction and it is CHOSEN: §11.4 sends the caller to the static "
      "profile rather than trading on a span the series never demonstrated, "
      "and the point is carried, so §8.3 forbids irreversible action on it "
      "anyway",
      derived(lambda: MIX_TAIL.at(_ml + timedelta(minutes=14, seconds=59)))
      is MIX_TAIL.points[-1]
      and raises(lambda: MIX_TAIL.at(_ml + timedelta(minutes=15)), Absence)
      and MIX_TAIL.points[-1].provisional)
# ⚠️ HABIT 1. Without this the three checks above would all pass under a
# single global tail rule of 15 minutes, or of an hour, depending which two
# you read - so the claim that the tail FOLLOWS THE SERIES is asserted as its
# own check, at one instant where the three fixtures must disagree.
_t30 = [derived(lambda s=s, l=l: s.covers(l + timedelta(minutes=30)))
        for s, l in ((Q, _ql), (H, _hl), (MIX_TAIL, _ml))]
check(f"...and those tails are genuinely three readings of one rule and not "
      f"one constant: 30 minutes past its own last point the hourly series "
      f"answers and both 900 s ones do not {_t30}",
      _t30 == [False, True, False])
check("a series of ONE point has no gap in front of it to copy, so it gets an "
      "hour - the shipped behaviour, kept deliberately for the degenerate case",
      derived(lambda: ONE.at(Q0 + timedelta(minutes=59, seconds=59)))
      is ONE.points[0]
      and raises(lambda: ONE.at(Q0 + HOUR), Absence))

print()
print("14f. order is established by the lookup, not assumed of the caller")
# ⚠️ `at()` now reads a point's NEIGHBOURS, so it WOULD depend on sortedness.
# The index is sorted at construction so that it cannot. `_validate_raw()`
# refuses out-of-order and duplicate starts on the `load_prices()` path
# (section 9 above), `price_history._parse_points()` re-sorts what it reads off
# disk and FCST-04's fill emits ascending - but `PriceRead.series()` itself has
# never enforced any of it, and an assumption that holds today and is unstated
# is how this class of defect recurs.
_REV = PriceRead.series(tuple(reversed(Q.points)), PriceProduct.DAY_AHEAD,
                        "tariff14", Q0)
check("the reversed fixture really is reversed, so the check below is not "
      "grading two identical series",
      _REV.points[0] is Q.points[-1] and _REV.points[-1] is Q.points[0]
      and _REV.points != Q.points)
_rev_bad = derived(lambda: [p.start.isoformat() for p in Q.points
                            if _REV.at(p.start) is not p])
check(f"...and a series handed its 96 points in REVERSE order answers at() "
      f"identically at every one of them "
      f"({_rev_bad if _rev_bad != [] else 'no misses'})",
      _rev_bad == [])
check("...while the load path refuses an unsorted feed outright, so the two "
      "defences are stated rather than one being quietly relied on",
      raises(lambda: _price._validate_raw(
          [RawPoint(T0 + timedelta(minutes=15), 0.1, False),
           RawPoint(T0, 0.2, False)], _caps), SourceUnavailable))
check("a DUPLICATED start does not raise anywhere - not at construction and "
      "not on the 30 s control path. The later list entry takes the real span "
      "and the earlier one becomes unreachable through at(); load_prices() "
      "refuses the shape outright, and this says what happens to anyone who "
      "builds one by hand",
      derived(lambda: PriceRead.series(
          [PricePoint(brussels(Q0), 0.10, False),
           PricePoint(brussels(Q0), 0.20, False),
           PricePoint(brussels(Q0 + timedelta(minutes=15)), 0.30, False)],
          PriceProduct.DAY_AHEAD, "tariff14", Q0).at(Q0).eur_per_kwh) == 0.20)

print()
print("14g. covers() shared it; window() did not")
check("covers() shared the defect EXACTLY, because it has no window rule of "
      "its own - it asks at() - so the one change fixes both. On the 900 s "
      "day it is now True 14:59 past the last point and False 15:00 past it, "
      "where an hour-wide window said True for another 45 minutes",
      derived(lambda: Q.covers(_ql + timedelta(minutes=14, seconds=59))) is True
      and derived(lambda: Q.covers(_ql + timedelta(minutes=15))) is False
      and derived(lambda: Q.covers(_ql + timedelta(minutes=30))) is False
      and derived(lambda: Q.covers(Q0 + timedelta(minutes=15))) is True)
check("window() did NOT share it, and the reason is worth a check rather than "
      "a rediscovery: it grades a point's START against a half-open range and "
      "never asks how long a point lasts, so it carries no resolution "
      "assumption at all. 00:07:30 .. 00:22:30 holds exactly the 00:15 point "
      "- while at(00:07:30) is the 00:00 one, which is what makes it a "
      "DIFFERENT question and not a coverage query",
      derived(lambda: Q.window(Q0 + timedelta(minutes=7, seconds=30),
                               Q0 + timedelta(minutes=22, seconds=30)))
      == (Q.points[1],)
      and derived(lambda: Q.at(Q0 + timedelta(minutes=7, seconds=30)))
      is Q.points[0])

print()
print("14h. the 'no data is not free' contract, unchanged")
check("at() answers with a PricePoint or raises Absence - never None, at any "
      "resolution, so `series.at(t) or fallback` cannot become a price of zero",
      all(isinstance(derived(lambda s=s: s.at(s.points[0].start)), PricePoint)
          for s in (Q, H, MIX, MIX_TAIL, ONE))
      and raises(lambda: Q.at(Q0 - timedelta(seconds=1)), Absence)
      and raises(lambda: Q.at(Q0 + timedelta(days=2)), Absence))
_absence_words = _reason(lambda: Q.at(Q0 + timedelta(days=2)))
check(f"...and the refusal still names the range the series runs, so an "
      f"operator can tell 'outside the horizon' from 'into a hole'",
      Q.points[0].start.isoformat() in _absence_words
      and Q.points[-1].start.isoformat() in _absence_words)
check("...and it never invents a neighbour: the instant one second before the "
      "series starts is refused, not answered with the first point",
      raises(lambda: Q.at(Q0 - timedelta(seconds=1)), Absence)
      and raises(lambda: MIX.at(Q0 - timedelta(seconds=1)), Absence)
      and raises(lambda: H.at(Q0 - timedelta(seconds=1)), Absence))


# --------------------------------------------------------------------------- #
for _d in _tmpdirs_t10:
    shutil.rmtree(_d, ignore_errors=True)

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