"""PRICE-01 - the owner's price server as a price source, and its fallbacks.
`python3 test_price_service.py`

WHAT THIS FILE DEFENDS, in the order the card names it:

  1. `price_service` fetches from ONE fixed HTTPS host with a bearer token, and
     the price it returns is wholesale energy, so markup and taxes are added
     once, exactly as for `generic_dayahead`.
  2. The chain own server -> energy-charts -> stored prices: every way the
     server can fail steps down, every step down and back up is logged once,
     and a day with every source down still plans from what was kept.
  3. No price for now: one status line and one notification, once, gone when a
     price returns. English and Dutch.
  4. The token is in no log, no status row, no notification, no error text.
  5. The maintenance charge gate keeps the Home Assistant rule (current price
     at or below today's average), read from the provider. The same day is fed
     both ways and every 15-minute period must give the same answer.
  6. Modes plan on these prices.

THE SERVER IS A MOCK. This session holds no token, so it cannot ask the real
one for a price. `Server` below answers the layout the design's API section
gives (`unix_seconds`, `price`, `unit`, `zone`, `covers`, `source`,
`fetched_at`), over a transport with the same contract as `price.https_get`.
Section 11 drives the REAL `https_get` against a local aiohttp server, for the
parts a mock transport cannot show: the header on the wire, the status words,
the timeout, a refused connection.

THE PRICES ARE NOT A FLAT LINE. They are deterministic and uneven, with a
negative midday, so "at or below the average" separates quarters on both sides
and a rule that always says yes or always says no cannot pass.

⚠️ MUTATING A MODULE THIS FILE GRADES? Purge `__pycache__` first, or use
`tools/mutation_run.sh`. See MUTATION.md.
"""

import asyncio
import io
import json
import logging
import os
import shutil
import sys
import tempfile
from datetime import date, datetime, timedelta, timezone

from named_absence import nth
from app import main as _main
from app import messages, price as P, price_chain as C
from app.price import (
    PROVIDERS, PriceProduct, SourceUnavailable, brussels, brussels_offset,
    load_prices, scrub, status_words)
from app.price_history import PriceHistory
from app.static_tariff import resolve_tariff
from app.validate import validate_startup_config

fails = []


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


def run(coro):
    return asyncio.run(coro)


UTC = timezone.utc
HERE = os.path.dirname(os.path.abspath(__file__))
TOKEN = "tk-7Hq2-secret-9xZ4-PRICE01"
HOST = P.PriceService.HOST
EC_HOST = "api.energy-charts.info"
MARKUP, TAXES = 0.031, 0.057
_TMP = []


def tmpdir(prefix="price01-"):
    d = tempfile.mkdtemp(prefix=prefix)
    _TMP.append(d)
    return d


# --------------------------------------------------------------------------- #
# The mock server, and the day it serves
# --------------------------------------------------------------------------- #

def local_midnight(d: date) -> datetime:
    """Brussels midnight of `d` as a UTC instant. Midnight is never inside the
    02:00-03:00 change, so one of the two offsets is always right."""
    for off in (P.CET, P.CEST):
        t = datetime(d.year, d.month, d.day, tzinfo=off)
        if brussels_offset(t) == off:
            return t.astimezone(UTC)
    raise AssertionError(d)


def day_starts(d: date) -> list:
    """The UTC instants of every 15-minute period of Brussels day `d`: 96, or
    92 and 100 on the two clock-change days."""
    a, b = local_midnight(d), local_midnight(d + timedelta(days=1))
    out, t = [], a
    while t < b:
        out.append(t)
        t += timedelta(seconds=900)
    return out


def ws_price(d: date, i: int) -> float:
    """EUR/MWh for period `i` of day `d`. Uneven, two decimals, negative at
    midday on some days. Not a function of anything the code under test sees."""
    base = 40 + 0.9 * ((d.toordinal() * 131 + i * i * 7) % 120)
    if 44 <= i <= 58 and d.toordinal() % 2 == 0:
        base -= 80
    return round(base, 2)


def day_rows(d: date) -> list:
    return [(int(t.timestamp()), ws_price(d, i))
            for i, t in enumerate(day_starts(d))]


class Clock:
    def __init__(self, now):
        self.now = now

    def __call__(self):
        return self.now


class Server:
    """The price server, as the design's API section describes it.

    `published(now)` is the days it holds: today always, tomorrow from 13:00
    Brussels time if `tomorrow` is on. `mode` picks what it does to a request;
    every mode that is not "up" is a way a real server or the road to it can
    fail.
    """

    def __init__(self, clock, token=TOKEN):
        self.clock, self.token = clock, token
        self.mode = "up"
        self.tomorrow = True
        self.requests = []

    def days(self):
        today = brussels(self.clock()).date()
        return [today] + ([today + timedelta(days=1)] if self.tomorrow else [])

    def doc(self, days=None):
        days = self.days() if days is None else days
        rows = [r for d in days for r in day_rows(d)]
        return {"unix_seconds": [r[0] for r in rows],
                "price": [r[1] for r in rows],
                "unit": "EUR / MWh", "zone": "BE",
                "covers": [d.isoformat() for d in days],
                "source": {d.isoformat(): "entsoe" for d in days},
                "fetched_at": self.clock().strftime("%Y-%m-%dT%H:%M:%SZ")}

    def answer(self, headers):
        self.requests.append(dict(headers or {}))
        m = self.mode
        if (headers or {}).get("Authorization") != "Bearer " + self.token \
                and m == "up":
            raise SourceUnavailable(status_words(401))
        if m == "refused":
            # A hostile library text: it carries the header it was sent.
            raise ConnectionRefusedError(
                f"Cannot connect to host {HOST}:443 ssl:default "
                f"[Connect call failed] Authorization: Bearer {self.token}")
        if m == "timeout":
            raise asyncio.TimeoutError()
        if m in ("401", "403", "429", "500", "503"):
            raise SourceUnavailable(status_words(int(m)))
        if m == "bad_json":
            return b"<html>Bad gateway</html>"
        d = self.doc()
        if m == "no_arrays":
            del d["price"]
        elif m == "unequal":
            d["price"] = d["price"][:-3]
        elif m == "wrong_unit":
            d["unit"] = "EUR / kWh"
        elif m == "bad_covers":
            d["covers"] = "2026-10-14"
        elif m == "empty":
            d["unix_seconds"], d["price"] = [], []
        elif m == "stale":
            d = self.doc([brussels(self.clock()).date() - timedelta(days=3)])
        elif m == "uncovered":
            d["covers"] = []
        elif m == "partial_tomorrow":
            # Tomorrow's rows are there but the server does not call the day
            # complete, so it must not be used.
            d["covers"] = d["covers"][:1]
        return json.dumps(d).encode()


class EnergyCharts:
    """api.energy-charts.info: the columnar layout, for the dates asked for."""

    def __init__(self, clock):
        self.clock = clock
        self.mode = "up"
        self.tomorrow = True
        self.requests = []

    def answer(self, url):
        self.requests.append(url)
        if self.mode == "down":
            raise SourceUnavailable("energy-charts is not answering")
        if self.mode == "503":
            raise ConnectionError("503 Service Unavailable")
        q = dict(p.split("=") for p in url.split("?", 1)[1].split("&"))
        a, b = date.fromisoformat(q["start"]), date.fromisoformat(q["end"])
        today = brussels(self.clock()).date()
        out, d = [], a
        while d <= b:
            if d <= today or (d == today + timedelta(days=1)
                              and self.tomorrow):
                out += day_rows(d)
            d += timedelta(days=1)
        return json.dumps({"unix_seconds": [r[0] for r in out],
                           "price": [r[1] for r in out],
                           "unit": "EUR / MWh", "license_info": "x",
                           "deprecated": False}).encode()


class Net:
    """The transport. Same contract as `price.https_get`."""

    def __init__(self, clock):
        self.clock = clock
        self.server = Server(clock)
        self.ec = EnergyCharts(clock)
        self.calls = []

    async def __call__(self, url, timeout_s, headers=None):
        self.calls.append((url, dict(headers or {})))
        host = url.split("/")[2]
        if host == HOST:
            return self.server.answer(headers)
        if host == EC_HOST:
            return self.ec.answer(url)
        raise AssertionError(f"a request left for an undeclared host: {url}")

    def down(self):
        self.server.mode, self.ec.mode = "refused", "down"

    def up(self):
        self.server.mode, self.ec.mode = "up", "up"


# --------------------------------------------------------------------------- #
# A place to read logs, notifications and statuses from
# --------------------------------------------------------------------------- #

class Tap(logging.Handler):
    def __init__(self):
        super().__init__(logging.DEBUG)
        self.records = []

    def emit(self, record):
        self.records.append(record)

    def lines(self, minimum=logging.INFO):
        return [r.getMessage() for r in self.records
                if r.levelno >= minimum]

    def text(self):
        return "\n".join(f"{r.levelname} {r.name} {r.getMessage()}"
                         for r in self.records)


TAP = Tap()
_root = logging.getLogger()
_root.addHandler(TAP)
_root.setLevel(logging.DEBUG)


def mark():
    return len(TAP.records)


def since_mark(n, minimum=logging.INFO):
    return [r.getMessage() for r in TAP.records[n:] if r.levelno >= minimum]


def opts(**over):
    o = {"price_provider": "price_service", "price_service_token": TOKEN,
         "price_markup_eur_kwh": MARKUP, "price_taxes_eur_kwh": TAXES,
         "price_timeout_s": 5}
    o.update(over)
    return o


NOW = datetime(2026, 10, 14, 15, 7, tzinfo=UTC)         # 17:07 Brussels (CEST)


def make_ctx(now=NOW, **over):
    clock = Clock(now)
    net = Net(clock)
    hist = PriceHistory(os.path.join(tmpdir(), "price_history.json"))
    return type("Ctx", (), {"clock": clock, "net": net, "hist": hist,
                            "watch": C.PriceWatch(), "o": opts(**over)})()


def refresh(ctx, now=None):
    """One background refresh, exactly as `run_tariff_refresh` calls it."""
    if now is not None:
        ctx.clock.now = now
    now = ctx.clock.now
    return run(resolve_tariff(ctx.o, now, now + timedelta(hours=48),
                              transport=ctx.net, now=now, history=ctx.hist,
                              watch=ctx.watch))


def covers_now(plan, now):
    return plan.published is not None and plan.published.covers(now)


def all_in(ws, markup=MARKUP, taxes=TAXES):
    return round(ws / 1000 + markup + taxes, 6)


# =========================================================================== #
print("0. the mock is the server the design describes, and the day is uneven")
_ctx = make_ctx()
_d = _ctx.net.server.doc()
check("the answer has the design's keys",
      {"unix_seconds", "price", "unit", "zone", "covers", "source",
       "fetched_at"} <= set(_d))
check("...with one price per start, 900 s apart, today and tomorrow",
      len(_d["unix_seconds"]) == len(_d["price"]) == 192
      and {b - a for a, b in zip(_d["unix_seconds"],
                                 _d["unix_seconds"][1:])} == {900}
      and _d["covers"] == ["2026-10-14", "2026-10-15"])
check("the fixture is uneven and has a negative price, so 'at or below the "
      "average' separates periods on both sides",
      min(_d["price"]) < 0 < max(_d["price"])
      and 20 < sum(1 for p in _d["price"][:96]
                   if p <= sum(_d["price"][:96]) / 96) < 76)
check("clock-change days have 92 and 100 periods, not 96",
      (len(day_starts(date(2026, 3, 29))), len(day_starts(date(2026, 10, 25))))
      == (92, 100))

# =========================================================================== #
print("\n1. the provider: one fixed host, a bearer token, wholesale energy")
_ctx = make_ctx()
_n = mark()
_plan = refresh(_ctx)
check("the first step answers: the plan came from the price service",
      _plan.step == C.STEP_OWN and _plan.published.provider == "price_service"
      and _plan.from_feed)
_url, _hdr = _ctx.net.calls[0]
check("the request went to the one fixed HTTPS URL, and only to it",
      _url == "https://prices.kammenstraatha.duckdns.org/v1/dayahead/BE"
      and [u for u, _h in _ctx.net.calls if HOST not in u] == [])
check("...with the token as a bearer header", _hdr == {"Authorization":
                                                      "Bearer " + TOKEN})
check("...and the token is not in the URL", TOKEN not in _url)
check("the host is declared as the plugin's one egress host, so the framework "
      "refuses any other",
      P.PriceService.capabilities().egress_hosts == (HOST,))
def _refused(fn):
    try:
        fn()
    except SourceUnavailable:
        return True
    return False


check("...an http:// URL on the right host is refused",
      _refused(lambda: P.check_egress("http://" + HOST + "/v1/dayahead/BE",
                                      P.PriceService.capabilities())))
check("...and so is another host over https",
      _refused(lambda: P.check_egress("https://example.org/v1/dayahead/BE",
                                      P.PriceService.capabilities())))
_ctx2 = make_ctx(price_feed_url="http://evil.example/x")
refresh(_ctx2)
check("the operator's feed override is ignored by this provider: the request "
      "still went to the fixed URL",
      _ctx2.net.calls[0][0] == P.PriceService.URL)

# markup and taxes are added once, as for generic_dayahead
_ps = _plan.published
_gd = run(load_prices(opts(price_provider="generic_dayahead"), NOW,
                      NOW + timedelta(hours=48), transport=_ctx.net))
_ps_prices = [p.eur_per_kwh for p in _ps.points]
_want = [all_in(r[1]) for d in _ctx.net.server.days() for r in day_rows(d)]
check("the price is the wholesale figure plus markup and taxes, once: every "
      "point equals ws/1000 + markup + taxes",
      _ps_prices == _want)
check("...and a price_service series equals the generic_dayahead one for the "
      "same days, point for point (start and price)",
      [(p.start, p.eur_per_kwh) for p in _ps.points]
      == [(p.start, p.eur_per_kwh) for p in _gd.points])
check("the provider declares energy only, so the framework adds markup and "
      "taxes, and the product is day-ahead",
      P.PriceService.capabilities().adds
      == P.GenericDayAhead.capabilities().adds
      and P.PriceService.capabilities().product is PriceProduct.DAY_AHEAD)
check("the plugin cannot reach a framework key (register() would refuse it)",
      P.code_leaks(P.PriceService) == [])

_ctx3 = make_ctx(price_service_token="")
_p3 = refresh(_ctx3)
check("an empty token sends NO request to the server and falls to energy-charts",
      not any(HOST in u for u, _h in _ctx3.net.calls)
      and _p3.step == C.STEP_PUBLIC)
check("a token with a line break in it is never sent",
      not any(HOST in u for u, _h in
              (lambda c: (refresh(c), c.net.calls)[1])(
                  make_ctx(price_service_token="abc\r\nX-Evil: 1"))))

# =========================================================================== #
print("\n2. every way the server can fail steps down to energy-charts")
_FAILURES = {
    "refused":      "does not answer",
    "timeout":      "does not answer",
    "401":          "refused the token",
    "403":          "refused the token",
    "429":          "does not answer",
    "500":          "does not answer",
    "503":          "does not answer",
    "bad_json":     "does not answer",
    "no_arrays":    "does not answer",
    "unequal":      "does not answer",
    "wrong_unit":   "does not answer",
    "bad_covers":   "does not answer",
    "empty":        "does not answer",
    "stale":        "does not answer",
    "uncovered":    "does not answer",
}
for _mode, _words in _FAILURES.items():
    _c = make_ctx()
    _c.net.server.mode = _mode
    _n = mark()
    _p = refresh(_c)
    _warn = since_mark(_n, logging.WARNING)
    check(f"server {_mode}: prices come from energy-charts, firm, covering now",
          _p.step == C.STEP_PUBLIC and _p.from_feed
          and _p.published.provider == "generic_dayahead"
          and covers_now(_p, _c.clock.now))
    check(f"...one WARNING says why, plainly ({_warn})",
          len(_warn) == 1 and _words in _warn[0]
          and "energy-charts" in _warn[0])
    check(f"...and the energy-charts request never carried the token",
          all(TOKEN not in json.dumps(h)
              for u, h in _c.net.calls if EC_HOST in u))
    _n2 = mark()
    refresh(_c, _c.clock.now + timedelta(minutes=30))
    check(f"...and the next refresh of the same failure logs nothing again",
          since_mark(_n2, logging.INFO) == [])

_c = make_ctx(price_feed_url="https://api.energy-charts.info/price?bzn=BE")
_c.net.server.mode = "refused"
_p = refresh(_c)
_ec_urls = [u for u, _h in _c.net.calls if EC_HOST in u]
check("the operator's feed override is not used for the energy-charts step: it "
      "is asked for its own feed with the date range, and answers",
      _p.step == C.STEP_PUBLIC and len(_ec_urls) == 1
      and "start=" in _ec_urls[0] and "end=" in _ec_urls[0])

# partial day: tomorrow not published yet (before ~13:00 the server holds today)
_c = make_ctx(now=datetime(2026, 10, 14, 6, 0, tzinfo=UTC))
_c.net.server.tomorrow = False
_n = mark()
_p = refresh(_c)
check("tomorrow missing: the server is still the source, no warning",
      _p.step == C.STEP_OWN and since_mark(_n, logging.WARNING) == [])
_days = {brussels(x.start).date() for x in _p.published.points}
check("...the series holds today only, and the plan carries the rest from "
      "yesterday's saved day as estimated, not firm",
      _days == {date(2026, 10, 14)})
_c = make_ctx()
_c.net.server.mode = "partial_tomorrow"
_p = refresh(_c)
check("a day the server does not list as complete is not used, even if its "
      "rows are there",
      _p.step == C.STEP_OWN
      and {brussels(x.start).date() for x in _p.published.points}
      == {date(2026, 10, 14)})

# =========================================================================== #
print("\n3. both sources down: the prices the history kept")
_c = make_ctx()
refresh(_c)                                          # a fetch that worked
_snap = open(_c.hist.path, "rb").read()
_st0 = os.stat(_c.hist.path)
_c.net.down()
_n = mark()
_p = refresh(_c, NOW + timedelta(minutes=30))
_warn = since_mark(_n, logging.WARNING)
check("the plan comes from the saved prices, is firm, and covers now",
      _p.step == C.STEP_STORED and _p.from_feed
      and _p.published.provider == "price_history"
      and covers_now(_p, _c.clock.now))
check(f"...and one WARNING says so plainly ({_warn})",
      _warn == ["No price source answers, so the add-on uses the prices it "
                "saved."])
check("...with the reasons of each step on a DEBUG line, token-free",
      any("Price sources said" in r.getMessage() and r.levelno == logging.DEBUG
          for r in TAP.records[_n:])
      and all(TOKEN not in r.getMessage() for r in TAP.records[_n:]))
_st1 = os.stat(_c.hist.path)
check("...and the saved file was not written again from a read of itself (the "
      "file is replaced on every save, so the same inode and time mean no save)",
      open(_c.hist.path, "rb").read() == _snap
      and (_st1.st_ino, _st1.st_mtime_ns) == (_st0.st_ino, _st0.st_mtime_ns))
_stored_prices = [p.eur_per_kwh for p in _p.published.points]
check("the saved prices are the ones fetched, not carried guesses",
      all(not p.provisional for p in _p.published.points)
      and _stored_prices == _want)
_n = mark()
refresh(_c, NOW + timedelta(minutes=60))
check("a second refresh in the same state logs nothing",
      since_mark(_n, logging.INFO) == [])

# nothing stored, everything down: no price
_c = make_ctx()
_c.net.down()
_n = mark()
_p = refresh(_c)
check("with nothing saved and every source down there is no plan from a feed",
      _p.step == C.STEP_NONE and not _p.from_feed and _p.published is None)
check("...the chain says so in one WARNING",
      [r.getMessage() for r in TAP.records[_n:]
       if r.name == "goodwe.price.chain" and r.levelno >= logging.WARNING]
      == ["No price source answers, and no saved price covers now."])
check("...and what each step said is kept for DEBUG, without the token",
      len(_p.reasons) == 3 and all(TOKEN not in r for r in _p.reasons)
      and "price_service" in _p.reasons[0]
      and "generic_dayahead" in _p.reasons[1]
      and "price_history" in _p.reasons[2])

# =========================================================================== #
print("\n4. back up the chain, one line per step")
_c = make_ctx()
_c.net.down()
refresh(_c)
_n = mark()
_c.net.ec.mode = "up"
refresh(_c, NOW + timedelta(minutes=30))
check("energy-charts back: INFO says so once",
      since_mark(_n, logging.INFO)
      == ["Prices come from energy-charts again."])
_n = mark()
_c.net.server.mode = "up"
refresh(_c, NOW + timedelta(minutes=60))
check("the server back: INFO says so once",
      since_mark(_n, logging.INFO) == ["The price service answers again."])
_n = mark()
refresh(_c, NOW + timedelta(minutes=90))
check("and nothing more while it stays up", since_mark(_n, logging.INFO) == [])
_c = make_ctx()
refresh(_c)
_c.net.down()
refresh(_c, NOW + timedelta(minutes=30))
_n = mark()
_c.net.server.mode = "up"
refresh(_c, NOW + timedelta(minutes=60))
check("stored -> server directly is one 'answers again' line",
      since_mark(_n, logging.INFO) == ["The price service answers again."])
_c = make_ctx()
_n = mark()
_c.net.server.mode = "401"
refresh(_c)
check("a start-up with the server refusing is one WARNING, not an INFO then a "
      "WARNING",
      since_mark(_n, logging.INFO)
      == ["The price service refused the token, so prices come from "
          "energy-charts."])

_w = C.PriceWatch()
_w.note(C.STEP_NONE)
_n = mark()
_w.note(C.STEP_STORED)
check("coming back from no price to the saved prices is one INFO, not a "
      "second alarm",
      since_mark(_n, logging.INFO)
      == ["The add-on uses the prices it saved again."])

# =========================================================================== #
print("\n5. a full day with every source down")
# A good fetch at 13:30 Brussels with tomorrow in it; then nothing answers for
# 24 hours. Every 30-minute refresh must still produce a firm plan that covers
# now, until the saved day itself runs out at midnight after tomorrow.
_start = datetime(2026, 10, 14, 11, 30, tzinfo=UTC)            # 13:30 CEST
_c = make_ctx(now=_start)
refresh(_c)
_c.net.down()
_bad = []
_t = _start
_gate_nones = 0
while _t < _start + timedelta(hours=24):
    _t += timedelta(minutes=30)
    _p = refresh(_c, _t)
    if not (_p.from_feed and _p.step == C.STEP_STORED
            and covers_now(_p, _t)):
        _bad.append(_t.isoformat())
    if not _p.firm_points():
        _bad.append("no firm point at " + _t.isoformat())
check(f"48 refreshes over 24 h, all from the saved day, all firm, all covering "
      f"now ({_bad[:3]})", not _bad)
_g = C.gate_for(_p.published, _t)
check("...and the charge gate still gives an answer 24 h later",
      _g.covered and _g.words.startswith("Prices from saved prices"))
_end = local = local_midnight(date(2026, 10, 16))
_p = refresh(_c, _end - timedelta(minutes=15))
check("the last saved period still covers", covers_now(_p, _end
                                                      - timedelta(minutes=15)))
_p = refresh(_c, _end)
check("and at the first moment after the saved days there is no price, and "
      "the series says since when",
      _p.step == C.STEP_NONE and C.gate_for(
          _c.hist.stored(_end).match(series=lambda r: r, no_data=None), _end
      ).since == _end)

# =========================================================================== #
print("\n6. no price for now: one status line, one notification, cleared")
OPTS_C = {"setpoint_entity": "number.sp", "max_w": 5000,
          "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
          "meter_source": "off", "meter_entity": "sensor.grid",
          "maintenance_enabled": True,
          "maintenance_charge_w": 2000, "maintenance_discharge_w": 2000}


class FakeHass:
    def __init__(self):
        self.calls = []
        self.values = {"sensor.grid": 0.0, "sensor.soc": 55.0,
                       "sensor.batt": 0.0}

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

    async def number(self, entity, invert=False):
        return self.values.get(entity)

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

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


class FakeStore:
    data = {}

    def set(self, *a):
        pass

    def get_time(self, key):
        return None


class FakePub:
    enabled = False

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

    def close(self):
        pass


class FakeDT(datetime):
    """`datetime` with a clock the test moves."""
    _clock = None

    @classmethod
    def now(cls, tz=None):
        t = cls._clock.now
        return t.astimezone(tz) if tz else t.replace(tzinfo=None)


def controller(clock, o, tmp=None):
    tmp = tmp or tmpdir()
    stop = asyncio.Event()
    ctl = _main.Controller(dict(OPTS_C, **o), FakeHass(), FakeStore(),
                           FakePub(), stop=stop,
                           peak_db_path=os.path.join(tmp, "tariff.db"),
                           price_history_path=os.path.join(tmp, "ph.json"))
    return ctl


_REAL_DT, _REAL_GET = _main.datetime, P.https_get
_real_resolve = _main.resolve_tariff


class Wired:
    """A controller whose clock and network are the test's."""

    def __init__(self, now, **over):
        self.clock = Clock(now)
        FakeDT._clock = self.clock
        self.net = Net(self.clock)
        self.ctl = controller(self.clock, opts(**over))
        self.refreshes = 0

    def __enter__(self):
        _main.datetime = FakeDT
        P.https_get = self.net
        W = self

        async def counting(*a, **k):
            r = await _real_resolve(*a, **k)
            W.refreshes += 1
            return r
        _main.resolve_tariff = counting
        return self

    def __exit__(self, *exc):
        _main.datetime, P.https_get = _REAL_DT, _REAL_GET
        _main.resolve_tariff = _real_resolve

    async def refresh(self, now=None):
        if now is not None:
            self.clock.now = now
        c = self.ctl
        n = self.refreshes
        c.stopping = False
        c.stop_event.clear()
        task = asyncio.create_task(c.run_tariff_refresh())
        for _ in range(500):
            if self.refreshes > n:
                break
            await asyncio.sleep(0.01)
        await asyncio.sleep(0.02)
        c.stopping = True
        c.stop_event.set()
        await asyncio.wait_for(task, timeout=5)

    async def tick(self, now=None):
        """One control tick as `run_control` makes it: read inputs, then the
        cycle, which is where the price-gap check is called from."""
        if now is not None:
            self.clock.now = now
        self.ctl.auto = True
        await self.ctl.read_inputs()
        await self.ctl.cycle()

    def all_notes(self):
        return [(s, d) for dom, s, d in self.ctl.hass.calls
                if dom == "persistent_notification"]

    def notes(self):
        """Without the dismiss of the old fixed-tariff card that every start
        makes (section 14 looks at it through `all_notes`)."""
        return [x for x in self.all_notes()
                if x != ("dismiss", {"notification_id": "goodwe_fixed_tariff_unused"})]


async def _scenario_gap():
    out = {}
    with Wired(NOW) as W:
        c = W.ctl
        out["rows_before"] = dict(c.status_payload()["rows"])
        await W.tick()
        out["before_first_refresh"] = (c.price_gap, list(W.notes()))
        W.net.down()
        await W.refresh()
        for k in range(6):                      # six ticks inside the gap
            await W.tick(NOW + timedelta(seconds=30 * k))
        out["gap"] = (c.price_gap, c.price_gap_since, list(W.notes()))
        out["rows_gap"] = dict(c.status_payload()["rows"])
        out["payload_gap"] = json.dumps(c.status_payload(), default=str)
        out["cheap_gap"] = (c.cheap, c._price_gated,
                            c._charge_gate_open(c.inverters[0]))
        c.inverters[0].batt = -2000.0            # a pack really taking charge
        out["open_while_charging"] = c._charge_gate_open(c.inverters[0])
        c.inverters[0].batt = None
        # the prices come back
        W.net.up()
        await W.refresh(NOW + timedelta(minutes=30))
        for k in range(4):
            await W.tick(NOW + timedelta(minutes=30, seconds=30 * k))
        out["back"] = (c.price_gap, list(W.notes()))
        out["rows_back"] = dict(c.status_payload()["rows"])
        out["payload_back"] = json.dumps(c.status_payload(), default=str)
        out["words_back"] = c.tariff_words
    return out


_S = run(_scenario_gap())
check("before the first refresh has finished there is no gap and no "
      "notification, so a start-up is not an alarm",
      _S["before_first_refresh"] == (False, []) and
      "Prices" not in _S["rows_before"])
_gap, _since, _notes = _S["gap"]
check("every source down with nothing saved: a gap, since nothing ever "
      "published", _gap is True and _since is None)
check("...exactly ONE notification over six ticks, with the id the docs name",
      len(_notes) == 1 and _notes[0][0] == "create"
      and _notes[0][1]["notification_id"] == "goodwe_no_prices")
check("...in English and Dutch, plain and short",
      "No prices have come in yet" in _notes[0][1]["message"]
      and "Prijzen ontbreken. Er zijn nog geen prijzen binnen"
      in _notes[0][1]["message"]
      and _notes[0][1]["title"] == "No prices")
check("...the status row says it once, in the card's words",
      _S["rows_gap"].get("Prices") == "No prices yet, maintenance waits.")
check("...Cheap window is no, and the charge gate is shut",
      _S["cheap_gap"] == (False, True, False))
check("...but a pack that is really taking charge is not stopped by it",
      _S["open_while_charging"] is True)
_gap2, _notes2 = _S["back"]
check("a price returns: the gap ends, the same notification is taken down "
      "once, and no second one was raised",
      _gap2 is False and len(_notes2) == 2 and _notes2[1][0] == "dismiss"
      and _notes2[1][1] == {"notification_id": "goodwe_no_prices"})
check("...and the row goes with it", "Prices" not in _S["rows_back"])
check("...and the Tariff row names the source and the rule",
      _S["words_back"].startswith("Prices from the price service,"))

# the refresh task stops: after 90 minutes the plan is retired and the gate
# shuts, with the moment it was retired as 'since'
async def _scenario_dead_task():
    start = datetime(2026, 10, 14, 11, 30, tzinfo=UTC)
    with Wired(start) as W:
        c = W.ctl
        await W.refresh()
        await W.tick(start + timedelta(minutes=80))
        before = (c.price_gap, c.cheap is not None, len(W.notes()))
        await W.tick(start + timedelta(minutes=95))
        await W.tick(start + timedelta(minutes=96))
        mid = (c.price_gap, c.price_gap_since, list(W.notes()),
               c._charge_gate_open(c.inverters[0]))
        await W.refresh(start + timedelta(minutes=100))
        await W.tick(start + timedelta(minutes=101))
        return before, mid, (c.price_gap, list(W.notes()))
_b, _m, _a = run(_scenario_dead_task())
check("a plan 80 minutes old is still current: no gap, no notification",
      _b == (False, True, 0))
check("at 95 minutes with no refresh the plan is retired: a gap, since the "
      "moment it was retired (90 minutes after the fetch), one notification, "
      "gate shut",
      _m[0] is True and _m[1] == datetime(2026, 10, 14, 11, 30, tzinfo=UTC)
      + timedelta(seconds=_main.TARIFF_MAX_AGE_S)
      and len(_m[2]) == 1 and _m[3] is False)
check("...and the next refresh that works takes it down again",
      _a[0] is False and [n[0] for n in _a[1]] == ["create", "dismiss"])

# a gap that begins mid-day, with a price history that ends: 'since' is shown
async def _scenario_since():
    start = datetime(2026, 10, 14, 11, 30, tzinfo=UTC)
    with Wired(start) as W:
        c = W.ctl
        await W.refresh()                         # a good fetch
        W.net.down()
        end = local_midnight(date(2026, 10, 16))  # the saved days run out
        for k in range(3):
            await W.refresh(end - timedelta(minutes=30 * (2 - k)))
        await W.tick(end + timedelta(minutes=5))
        gap = (c.price_gap, c.price_gap_since)
        row = dict(c.status_payload()["rows"])
        notes = list(W.notes())
        # the Home screen
        from app import homeview
        plan = homeview.build_plan(c, W.clock.now, False, True)
        return gap, row, notes, plan["lines"], end
_gap, _row, _notes, _lines, _end = run(_scenario_since())
check("when the saved days run out, the gap begins at that moment",
      _gap == (True, _end))
check("...the row reads 'No prices since 00:00, maintenance waits.' (Brussels "
      "time, the card's words)",
      _row.get("Prices") == "No prices since 00:00, maintenance waits.")
check("...the one notification carries the same moment, in English and Dutch",
      len(_notes) == 1
      and "No prices since 00:00, so the monthly battery cycle waits"
      in _notes[0][1]["message"]
      and "Prijzen ontbreken. Sinds 00:00 zijn er geen prijzen, dus de "
      "maandelijkse batterijcyclus wacht" in _notes[0][1]["message"])
check("...and the Home screen says it once (English)",
      _lines.count("No prices since 00:00, maintenance waits.") == 1)
from app import uitext as _uitext                                  # noqa: E402
_nl = _uitext.words("nl")["home_words"]
check("...with the Dutch words 'Geen prijzen sinds {at}, onderhoud wacht.'",
      _nl["no_prices_since"] == "Geen prijzen sinds {at}, onderhoud wacht."
      and _nl["no_prices_yet"] == "Nog geen prijzen, onderhoud wacht."
      and _nl["no_prices_since_day"]
      == "Geen prijzen sinds {day} {month} om {at}, onderhoud wacht.")
_a = brussels(datetime(2026, 10, 13, 19, 0, tzinfo=UTC))
_b = brussels(datetime(2026, 10, 14, 9, 0, tzinfo=UTC))
check("an earlier day is spelled with its date, not only a time, in each "
      "language's own words",
      messages.since_text(_a, _b) == "13 Oct at 21:00"
      and messages.since_text(_a, _b, "nl") == "13 okt om 21:00"
      and messages.since_text(_b, _b) == "11:00")
from app import homeview as _hv                                    # noqa: E402
_prev = _hv._LANG[0]
try:
    _hv._LANG[0] = "nl"
    _nl_line = _hv.line_no_prices(datetime(2026, 10, 13, 19, 0, tzinfo=UTC),
                                  datetime(2026, 10, 14, 9, 0, tzinfo=UTC))
    _hv._LANG[0] = "en"
    _en_line = _hv.line_no_prices(datetime(2026, 10, 13, 19, 0, tzinfo=UTC),
                                  datetime(2026, 10, 14, 9, 0, tzinfo=UTC))
finally:
    _hv._LANG[0] = _prev
check("...and the Home line for an earlier day is in the page's language "
      f"({_en_line!r}, {_nl_line!r})",
      _en_line == "No prices since 13 October at 21:00, maintenance waits."
      and _nl_line == "Geen prijzen sinds 13 oktober om 21:00, onderhoud wacht.")

# =========================================================================== #
print("\n7. the token is in no log, status row, notification or error text")
_everything = [TAP.text(),
               json.dumps(_S, default=str), json.dumps(
                   [_row, _notes, _lines], default=str)]
check("across every scenario above the token never appeared in any log record "
      "at any level",
      all(TOKEN not in r.getMessage() and TOKEN not in str(r.args)
          and TOKEN not in (r.exc_text or "") for r in TAP.records))
check("...nor in a status row, a notification or a home line",
      all(TOKEN not in blob for blob in _everything[1:]))
check("...nor anywhere in the whole status payload, during the gap or after it",
      len(_S["payload_gap"]) > 1000 and len(_S["payload_back"]) > 1000
      and TOKEN not in _S["payload_gap"] and TOKEN not in _S["payload_back"])
check("...nor in the reason a failed read carries, even when the library's own "
      "error text holds the header it was sent",
      TOKEN not in (lambda c: (c.net.server.__setattr__("mode", "refused"),
                               c.net.ec.__setattr__("mode", "down"),
                               refresh(c).read.reason)[2])(make_ctx()))
_o = opts()
check("scrub() takes the value out of any text, and leaves other text alone",
      scrub(f"x {TOKEN} y", _o) == "x *** y"
      and scrub("no secret here", _o) == "no secret here"
      and scrub(f"x {TOKEN}", {"price_service_token": ""}) == f"x {TOKEN}")


async def _hostile(url, timeout_s, headers=None):
    # A transport that fails with the full request in its text, by any means.
    raise RuntimeError(f"boom {headers} {url} {TOKEN}")


_r = run(load_prices(opts(), NOW, NOW + timedelta(hours=48),
                     transport=_hostile))
check("an exception from the transport that carries the token reaches no "
      "reason text", _r.kind == "no_data" and TOKEN not in _r.reason)


def _fetch_error(transport):
    """What the PLUGIN itself raises, before the framework's scrub sees it."""
    try:
        run(P.PriceService().fetch(P.PluginOptions(opts()), NOW,
                                   NOW + timedelta(hours=48), transport))
    except SourceUnavailable as e:
        return str(e)
    return ""


_own = _fetch_error(_hostile)
check("the plugin's own error text has no token in it even before the "
      "framework scrubs (defence in depth: two layers, each graded alone)",
      _own.startswith("the price service did not answer (RuntimeError)")
      and TOKEN not in _own)


async def _hostile2(url, timeout_s, headers=None):
    raise SourceUnavailable(f"bad: {headers}")


check("a SourceUnavailable built from the headers is scrubbed too",
      TOKEN not in run(load_prices(opts(), NOW, NOW + timedelta(hours=48),
                                   transport=_hostile2)).reason)
from app import uilayout as _L, uipage as _UP                       # noqa: E402
check("the settings page never draws the token (it is a hidden, secret key), "
      "and it can keep the provider but not choose it",
      "price_service_token" in _L.HIDDEN
      and _L.field("price_service_token") is None
      and "price_service" in _UP.ONLY_IF_HELD)

# =========================================================================== #
print("\n8. the maintenance charge gate: the Home Assistant rule, from the "
      "provider's prices")


def ha_state_now(rows, t):
    """sensor.price_service_now, as price_service.yaml computes it."""
    u = [r[0] for r in rows]
    p = [r[1] for r in rows]
    step = (u[1] - u[0]) if len(u) > 1 else 900
    v = None
    for i in range(len(u)):
        if u[i] <= t < u[i] + step:
            v = p[i]
    if not (len(u) > 0 and u[0] <= t < u[-1] + 900):
        return None
    return None if v is None else round(v / 1000, 5)


def ha_state_avg(rows, day_start, day_end):
    """sensor.price_service_avg, as price_service.yaml computes it."""
    s = n = 0
    for ts, pr in rows:
        if day_start <= ts < day_end:
            s += pr
            n += 1
    return round(s / n / 1000, 5) if n else None


def ha_cheap(rows, t, today):
    """The add-on's old rule: `now_p <= avg_p`, None means not cheap."""
    now_p = ha_state_now(rows, t)
    a = int(local_midnight(today).timestamp())
    b = int(local_midnight(today + timedelta(days=1)).timestamp())
    avg_p = ha_state_avg(rows, a, b)
    return now_p is not None and avg_p is not None and now_p <= avg_p


def published_series(day_list, markup=MARKUP, taxes=TAXES):
    """The all-in series the provider would publish for these days, built by
    the real framework from the mock's rows."""
    c = make_ctx(now=local_midnight(day_list[0]) + timedelta(hours=10),
                 price_markup_eur_kwh=markup, price_taxes_eur_kwh=taxes)
    c.net.server.days = lambda: day_list
    p = refresh(c)
    assert p.step == C.STEP_OWN, p.step
    return p.published, [r for d in day_list for r in day_rows(d)]


_days = [date(2026, 10, 14), date(2026, 10, 15), date(2026, 3, 29),
         date(2026, 10, 25), date(2026, 7, 1), date(2026, 1, 2)]
_diffs, _checked, _cheap_yes = [], 0, 0
for _d in _days:
    _ser, _rows = published_series([_d, _d + timedelta(days=1)])
    for _t in day_starts(_d):
        _want = ha_cheap(_rows, int(_t.timestamp()), _d)
        for dt in (0, 450, 899):            # first, middle and last second
            _tt = _t + timedelta(seconds=dt)
            _got = C.gate_for(_ser, _tt).cheap
            _checked += 1
            if _got != _want:
                _diffs.append((_d.isoformat(), _tt.isoformat(), _want, _got))
        _cheap_yes += _want
check(f"{_checked} instants over {len(_days)} days (96-period days, the two "
      f"clock-change days, winter, summer): the provider's gate equals "
      f"now_p <= avg_p on the same prices, every time ({_diffs[:2]})",
      not _diffs and _checked > 1000)
check("...and the days really separate: a fifth to four fifths of periods are "
      "cheap, so neither 'always' nor 'never' passes",
      0.2 < _cheap_yes / (_checked / 3) < 0.8)
_diffs = []
for _mk, _tx in ((0.0, 0.0), (0.03, 0.11736), (0.04321, 0.09876),
                 (0.0, 0.2)):
    _ser, _rows = published_series([date(2026, 10, 14), date(2026, 10, 15)],
                                   _mk, _tx)
    for _t in day_starts(date(2026, 10, 14)):
        if C.gate_for(_ser, _t).cheap != ha_cheap(_rows, int(_t.timestamp()),
                                                  date(2026, 10, 14)):
            _diffs.append((_mk, _tx, _t.isoformat()))
check(f"...for markup and taxes of up to five decimals, with the wholesale "
      f"sensor unchanged: the same answer ({_diffs[:2]})", not _diffs)

# the same day through the REAL controller, both ways
async def _both_ways(day):
    rows = [r for d in (day, day + timedelta(days=1)) for r in day_rows(d)]
    out = []
    t0 = local_midnight(day)
    with Wired(t0 + timedelta(hours=10), price_markup_eur_kwh=MARKUP,
               price_taxes_eur_kwh=TAXES) as W:
        W.net.server.days = lambda: [day, day + timedelta(days=1)]
        W.net.ec.tomorrow = True
        W.clock.now = t0 + timedelta(hours=10)
        await W.refresh()
        ps = W.ctl
        ha_hass = FakeHass()
        FakeDT._clock = W.clock
        ha = controller(W.clock, {"price_provider": "off",
                                  "price_now_entity": "sensor.now",
                                  "price_avg_entity": "sensor.avg"})
        ha.hass = ha_hass
        a, b = (int(t0.timestamp()),
                int(local_midnight(day + timedelta(days=1)).timestamp()))
        last = None
        for t in day_starts(day):
            W.clock.now = t + timedelta(seconds=100)
            # The background refresh runs every 30 minutes; a plan older than
            # 90 minutes is retired, which is what keeps this gate shut when
            # the refresh task has died.
            if last is None or (t - last) >= timedelta(minutes=30):
                await W.refresh(W.clock.now)
                last = t
            ha_hass.values["sensor.now"] = ha_state_now(
                rows, int(W.clock.now.timestamp()))
            ha_hass.values["sensor.avg"] = ha_state_avg(rows, a, b)
            await ps.read_inputs()
            await ha.read_inputs()
            out.append((ha.cheap, ps.cheap,
                        ha._charge_gate_open(ha.inverters[0]),
                        ps._charge_gate_open(ps.inverters[0]),
                        ha._price_gated, ps._price_gated))
    return out


_rows_both = run(_both_ways(date(2026, 10, 14)))
check(f"the same day fed through the Home Assistant pair and through "
      f"price_service: {len(_rows_both)} periods, the cheap flag and the "
      f"charge gate agree in every one",
      len(_rows_both) == 96
      and all(a == b and c == d and e == f
              for a, b, c, d, e, f in _rows_both))
check("...with both kinds of period present (cheap and not cheap)",
      {r[0] for r in _rows_both} == {True, False})
check("...and an open gate whenever it is cheap, a shut one when it is not "
      "(no pack is charging)",
      all(r[3] == r[1] for r in _rows_both))

# what the old rule did when the price was unavailable, and what this does
_ser, _rows = published_series([date(2026, 10, 14)])
_late = local_midnight(date(2026, 10, 15)) + timedelta(minutes=1)
check("with no price for now both rules say not cheap (the HA sensor was "
      "'unavailable'), and the gate stays shut",
      C.gate_for(_ser, _late).cheap is False
      and ha_cheap(_rows, int(_late.timestamp()), date(2026, 10, 15)) is False)
check("it is NOT TariffPlan.cheap_now(): on a price exactly at the average "
      "this gate says cheap where cheap_now (strictly below) says no",
      C.gate_for(
          P.PriceRead.series(
              [P.PricePoint(local_midnight(date(2026, 10, 14))
                            + timedelta(minutes=15 * i), v, False)
               for i, v in enumerate((0.10, 0.20, 0.30))],
              PriceProduct.DAY_AHEAD, "price_service", NOW),
          local_midnight(date(2026, 10, 14)) + timedelta(minutes=16)).cheap
      is True)

# The Home Assistant average was rounded to 5 decimals before it was compared,
# so a quarter that sits ABOVE the exact average but AT the rounded one was
# cheap. Real days almost never land on that, so it is built here: three
# periods at 0.20000 and five at 0.20001 average 0.20000625, which rounds to
# 0.20001, and a period at 0.20001 is therefore cheap by the old rule.
_t0 = local_midnight(date(2026, 10, 14))
_vals = [0.20000] * 3 + [0.20001] * 5
_toy = P.PriceRead.series(
    [P.PricePoint(_t0 + timedelta(minutes=15 * i), v, False)
     for i, v in enumerate(_vals)], PriceProduct.DAY_AHEAD, "price_service",
    NOW)
_toy_rows = [(int((_t0 + timedelta(minutes=15 * i)).timestamp()),
              round(v * 1000, 2)) for i, v in enumerate(_vals)]
_at = lambda i: _t0 + timedelta(minutes=15 * i + 1)                 # noqa: E731
check("a period above the exact average but at the 5-decimal rounded one is "
      "cheap, as the Home Assistant pair said: the gate and the pair agree on "
      "all eight periods of a day built to sit on that edge, and the edge "
      "period is cheap",
      [C.gate_for(_toy, _at(i)).cheap for i in range(8)]
      == [ha_cheap(_toy_rows, int(_at(i).timestamp()), date(2026, 10, 14))
          for i in range(8)]
      and C.gate_for(_toy, _at(7)).cheap is True)

# =========================================================================== #
print("\n9. battery modes plan on the provider's prices")
async def _modes():
    day = date(2026, 10, 14)
    t = local_midnight(day) + timedelta(hours=8)
    with Wired(t, battery_modes_enabled=True) as W:
        await W.refresh()
        c = W.ctl
        plan = c.fresh_tariff_plan(t)
        return (plan is not None and plan.has_prices, plan.origin,
                plan.n_firm > 150,
                [p.provider for p in [plan.read]][0], c.modes_on)


_m = run(_modes())
check("the standing plan is a feed plan of firm price_service prices covering "
      "the two days, and battery modes are on",
      _m[0] and _m[1] == "provider" and _m[2] and _m[3] == "price_service"
      and _m[4] is True)

async def _modes_plan():
    day = date(2026, 10, 14)
    t = local_midnight(day) + timedelta(hours=8)
    with Wired(t, battery_modes_enabled=True, arbitrage_enabled=True) as W:
        await W.refresh()
        c = W.ctl
        plan = c.fresh_tariff_plan(t)
        verdict = plan.arbitrage(
            peak_guard_active=False, min_spread_eur_kwh=0.05,
            uncertainty=frozenset())
        return verdict.match(allowed=lambda v: ("allowed", v.spread_eur_kwh),
                             refused=lambda v: ("refused", v.reason))


_v = run(_modes_plan())
check(f"...and the arbitrage verdict on them is a real one over firm prices "
      f"{_v}", _v[0] == "allowed" and _v[1] > 0.1)

async def _runs_for(provider):
    """The kept price-run plan of a battery in cheap-hours mode, built by the
    shipped controller from a plan that came through `provider`."""
    from app import packmode
    t = local_midnight(date(2026, 10, 14)) + timedelta(hours=8)
    over = {"battery_modes_enabled": True, "capacity_kwh": 10}
    if provider != "price_service":
        over["price_provider"] = provider
    with Wired(t, **over) as W:
        await W.refresh()
        c = W.ctl
        c.store = FakeStore()
        c.store.data = {packmode.unit_key("inverter-a", packmode.MODE_KEY):
                        packmode.CHEAP_HOURS}
        c.inverters[0].soc, c.inverters[0].capacity_kwh = 40.0, 10.0
        return c._price_runs(t, c._mode_packs(), 20, 95)


_ra, _rb = run(_runs_for("price_service")), run(_runs_for("generic_dayahead"))
check("a battery in cheap-hours mode gets a real price-run plan from "
      "price_service prices (a cheap run is chosen)",
      _ra is not None and _ra.cheap is not None)
check("...the same plan, run for run, that generic_dayahead gives on the same "
      "prices: the modes read these prices as they read those",
      _ra == _rb)

# =========================================================================== #
print("\n10. the options: validation, defaults, the old pair")
def problems(**over):
    o = {"price_provider": "price_service", "price_service_token": TOKEN}
    o.update(over)
    base = dict(OPTS_C)
    base.update(o)
    return validate_startup_config(base, None, None)


check("price_service with a token starts", problems() == [])
_e = problems(price_service_token="")
check("...without a token it is refused, naming both settings by label and key "
      f"({_e})", len(_e) == 1 and "“Dynamic price source” (price_provider)"
      in _e[0] and "“Price service token” (price_service_token)" in _e[0]
      and TOKEN not in _e[0])
# Long enough to pass the length rule, so only the space can refuse it.
_SPACED = "tk-0123456789 abcdefgh"
_e = problems(price_service_token=_SPACED)
check("...a token with a space in it is refused for the space, and the token "
      "is not echoed",
      len(_e) == 1 and "space" in _e[0] and _SPACED not in _e[0])
_e = problems(price_now_entity="sensor.a", price_avg_entity="sensor.b")
check(f"the existing conflict refusal stands: the Home Assistant pair beside "
      f"this provider is refused ({len(_e)} problem)",
      any("would win over it" in x for x in _e))
check("a static_profile beside it is accepted (it is the fallback for "
      "trading)", problems(static_profile=json.dumps({"periods": [
          {"name": "all", "days": "all", "from": "00:00", "to": "24:00",
           "eur_per_kwh": 0.25}]})) == [])
check("the shipped default token is empty and the option is a password field",
      P.PRICE_OPTION_DEFAULTS["price_service_token"] == ""
      and "price_service_token: password?" in open(
          os.path.join(HERE, "config.yaml")).read())
check("config.yaml offers the new provider id",
      "price_service" in open(os.path.join(HERE, "config.yaml")).read()
      .split("price_provider: list(")[1].split(")")[0])
check("only this provider asks for the chain; the others take the road they "
      "always took",
      [pid for pid in sorted(PROVIDERS)
       if C.uses_chain({"price_provider": pid})] == ["price_service"]
      and not C.uses_chain({"price_provider": "off"}))

# =========================================================================== #
print("\n11. the real transport, against a local server")
async def _wire():
    from aiohttp import web
    seen = []
    hold = {"mode": "ok"}

    async def handler(request):
        seen.append(dict(request.headers))
        if hold["mode"] == "slow":
            await asyncio.sleep(3)
        if hold["mode"] in ("401", "403", "429", "503", "418"):
            return web.Response(status=int(hold["mode"]), text="nope " + TOKEN)
        if request.headers.get("Authorization") != "Bearer " + TOKEN:
            return web.Response(status=401, text="bad token")
        return web.Response(body=b'{"ok": 1}')
    elsewhere = []

    async def other(request):
        elsewhere.append(dict(request.headers))
        return web.Response(body=b'{"ok": 2}')

    async def moved(request):
        raise web.HTTPFound("/elsewhere")
    app = web.Application()
    app.router.add_get("/v1/dayahead/BE", handler)
    app.router.add_get("/moved", moved)
    app.router.add_get("/elsewhere", other)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    url = f"http://127.0.0.1:{port}/v1/dayahead/BE"
    res = {}
    hdr = {"Authorization": "Bearer " + TOKEN}
    res["ok"] = await P.https_get(url, 3, headers=hdr)
    res["seen"] = nth(seen, -1)
    for st in ("401", "403", "429", "503", "418"):
        hold["mode"] = st
        try:
            await P.https_get(url, 3, headers=hdr)
            res[st] = "no error"
        except SourceUnavailable as e:
            res[st] = str(e)
        except Exception as e:                                # noqa: BLE001
            res[st] = "LEAK " + repr(e)
    try:
        await P.https_get(url.replace("/v1/dayahead/BE", "/moved"), 3,
                          headers=hdr)
        res["302"] = "no error"
    except SourceUnavailable as e:
        res["302"] = str(e)
    res["followed"] = len(elsewhere)
    hold["mode"] = "slow"
    try:
        await P.https_get(url, 0.4, headers=hdr)
        res["slow"] = "no error"
    except SourceUnavailable as e:
        res["slow"] = str(e)
    except Exception as e:                                    # noqa: BLE001
        res["slow"] = "LEAK " + repr(e)
    await runner.cleanup()
    try:
        await P.https_get(url, 1, headers=hdr)
        res["closed"] = "no error"
    except SourceUnavailable as e:
        res["closed"] = str(e)
    except Exception as e:                                    # noqa: BLE001
        res["closed"] = "LEAK " + repr(e)
    # without headers it is the old transport: a status raises aiohttp's error
    return res


_w = run(_wire())
check("the token went on the wire as a bearer header, and the body came back",
      _w["seen"].get("Authorization") == "Bearer " + TOKEN
      and _w["ok"] == b'{"ok": 1}')
check(f"401 and 403 are 'refused the token', with the status and no body "
      f"({_w['401']!r})",
      _w["401"] == "the price service refused the token (401)"
      and _w["403"] == "the price service refused the token (403)")
check(f"429, 5xx and other statuses have their own plain words "
      f"({_w['429']!r}, {_w['503']!r}, {_w['418']!r})",
      _w["429"] == "the price service says too many requests (429)"
      and _w["503"] == "the price service has an error (503)"
      and _w["418"] == "the price service answered 418")
check(f"a redirect is not followed, so the token cannot be taken to another "
      f"place: it is reported as a status and the target was never asked "
      f"({_w['302']!r}, requests to the target: {_w['followed']})",
      _w["302"] == "the price service answered 302" and _w["followed"] == 0)
check(f"a timeout and a refused connection are SourceUnavailable with the "
      f"exception's NAME only ({_w['slow']!r}, {_w['closed']!r})",
      _w["slow"].startswith("the price service did not answer (")
      and _w["closed"].startswith("the price service did not answer ("))
check("...and nothing the server sent back, token echo included, reached any "
      "message", all(TOKEN not in v for k, v in _w.items()
                     if isinstance(v, str)))

# =========================================================================== #
print("\n12. restart in the middle of an outage, and the shipped logger names")
_c = make_ctx()
refresh(_c)
_c.net.down()
_hist2 = PriceHistory(_c.hist.path)                     # a restart: new objects
_c2 = type("C", (), {"clock": _c.clock, "net": _c.net, "hist": _hist2,
                     "watch": C.PriceWatch(), "o": _c.o})()
_n = mark()
_p = refresh(_c2, NOW + timedelta(hours=1))
check("a restart with every source down plans from what the file kept",
      _p.step == C.STEP_STORED and covers_now(_p, _c2.clock.now))
check("...and says so once, at WARNING",
      since_mark(_n, logging.WARNING)
      == ["No price source answers, so the add-on uses the prices it saved."])
check("the chain logs under goodwe.price.chain",
      all(r.name == "goodwe.price.chain" for r in TAP.records[_n:]
          if "price source" in r.getMessage() or "price service"
          in r.getMessage()))

# =========================================================================== #
print("\n13. round 2: the no-price words only with the maintenance charge on (F3)")
async def _scenario_maint(on):
    """Every source down, six ticks, then the prices come back. `on` is the
    maintenance switch."""
    from app import homeview
    with Wired(NOW, maintenance_enabled=on) as W:
        c = W.ctl
        W.net.down()
        await W.refresh()
        for k in range(6):
            await W.tick(NOW + timedelta(seconds=30 * k))
        mid = {"gap": c.price_gap, "notes": list(W.notes()),
               "row": dict(c.status_payload()["rows"]).get("Prices"),
               "home": [x for x in homeview.build_plan(
                   c, W.clock.now, False, True)["lines"]
                   if "No prices" in x],
               "gate": c._charge_gate_open(c.inverters[0]),
               "cheap": c.cheap}
        W.net.up()
        await W.refresh(NOW + timedelta(minutes=30))
        await W.tick(NOW + timedelta(minutes=30))
        return mid, list(W.notes())


_on_mid, _on_all = run(_scenario_maint(True))
_off_mid, _off_all = run(_scenario_maint(False))
check("maintenance on: the row, the Home line and ONE notification appear",
      _on_mid["gap"] is True and len(_on_mid["notes"]) == 1
      and _on_mid["row"] == "No prices yet, maintenance waits."
      and _on_mid["home"] == ["No prices yet, maintenance waits."])
check("...and the notification is taken down once when prices return",
      [n[0] for n in _on_all] == ["create", "dismiss"])
check("maintenance off: no row, no Home line, no notification, at any tick",
      _off_mid["gap"] is False and _off_mid["notes"] == []
      and _off_mid["row"] is None and _off_mid["home"] == []
      and _off_all == [])
check("...and the gate itself does not change with the switch: still shut "
      "with no price",
      _off_mid["gate"] is False and _on_mid["gate"] is False
      and _off_mid["cheap"] is False)


async def _scenario_switched_off():
    """A notification raised with the cycle on is cleared when it is off."""
    with Wired(NOW) as W:
        c = W.ctl
        W.net.down()
        await W.refresh()
        await W.tick(NOW)
        raised = list(W.notes())
        c.o["maintenance_enabled"] = False
        await W.tick(NOW + timedelta(seconds=30))
        await W.tick(NOW + timedelta(seconds=60))
        return raised, list(W.notes()), c.price_gap


_r, _n, _g = run(_scenario_switched_off())
check("a notification raised while the cycle was on is cleared, once, when "
      "it is off",
      len(_r) == 1 and [x[0] for x in _n] == ["create", "dismiss"]
      and _g is False)


async def _scenario_key_missing():
    """No `maintenance_enabled` key at all: its shipped default (off) holds."""
    with Wired(NOW) as W:
        c = W.ctl
        c.o.pop("maintenance_enabled", None)
        W.net.down()
        await W.refresh()
        for k in range(3):
            await W.tick(NOW + timedelta(seconds=30 * k))
        return c.price_gap, list(W.notes())


_g2, _n2 = run(_scenario_key_missing())
check("with no maintenance_enabled key its default (off) holds: no no-price "
      "words, no notification", _g2 is False and _n2 == [])

async def _modes_no_price():
    """Battery modes on, maintenance off, no price anywhere: what happens."""
    from app import packmode, homeview
    with Wired(NOW, battery_modes_enabled=True, capacity_kwh=10,
               maintenance_enabled=False) as W:
        c = W.ctl
        W.net.down()
        await W.refresh()
        c.store = FakeStore()
        c.store.data = {packmode.unit_key("inverter-a", packmode.MODE_KEY):
                        packmode.CHEAP_HOURS}
        c.inverters[0].soc, c.inverters[0].capacity_kwh = 40.0, 10.0
        await W.tick(NOW)
        runs = c._price_runs(NOW, c._mode_packs(), 20, 95)
        plan = homeview.build_plan(c, NOW, False, True)["lines"]
        return runs, c.price_gap, list(W.notes()), plan


_mr, _mg, _mn, _ml = run(_modes_no_price())
check("battery modes with no price and maintenance off: no price run is "
      "made (so no battery holds or trades on price), and nothing about "
      "'maintenance waits' is shown or notified",
      _mr is None and _mg is False and _mn == []
      and not [x for x in _ml if "maintenance waits" in x])
print("   Home plan lines in that state:", _ml)

# =========================================================================== #
print("\n14. round 3: a fixed tariff beside price_service is a log line and a "
      "settings note, no notification (G1, S05)")
_PROFILE = json.dumps({"periods": [
    {"name": "all", "days": "all", "from": "00:00", "to": "24:00",
     "eur_per_kwh": 0.25}]})
_LINE = "Fixed tariff is only used if no prices are available."
_NOTE = ("With the price service, the fixed tariff is only used if no prices "
         "are available.")
_OLD_ID = "goodwe_fixed_tariff_unused"


async def _scenario_fixed(**over):
    with Wired(NOW, **over) as W:
        before = mark()
        await W.refresh()
        for k in range(4):
            await W.tick(NOW + timedelta(seconds=30 * k))
        said = [m for m in since_mark(before, logging.INFO) if "only used if no prices" in m]
        return said, W.all_notes(), W.ctl


def _settings_notes(**over):
    base = dict(OPTS_C)
    base.update(price_provider="price_service", price_service_token=TOKEN)
    base.update(over)
    out = []
    validate_startup_config(base, None, None, notes=out)
    return [n.text for n in out if n.keys == ("static_profile",)
            and "fixed tariff" in n.text.lower()]


_said, _fn, _fc = run(_scenario_fixed(static_profile=_PROFILE))
check("price_service + a set profile: the INFO line appears once over four "
      "ticks, and the setting is not refused",
      _said == [_LINE] and problems(static_profile=_PROFILE) == [])
check("...no notification is raised, only the old id is dismissed",
      [x for x in _fn if x[1].get("notification_id") == _OLD_ID]
      == [("dismiss", {"notification_id": _OLD_ID})]
      and not [x for x in _fn if x[0] == "create"])
for _label, _kw in (("no profile", {}),
                    ("an empty profile", {"static_profile": ""}),
                    ("a spaces-only profile", {"static_profile": "   "}),
                    ("a line-break-only profile", {"static_profile": "\n \t"}),
                    ("another provider", {"price_provider": "generic_dayahead",
                                          "static_profile": _PROFILE})):
    _s, _n, _c = run(_scenario_fixed(**_kw))
    check(f"{_label}: no log line", _s == [])
    check(f"{_label}: the old id is still dismissed at start (harmless)",
          [x for x in _n if x[1].get("notification_id") == _OLD_ID]
          == [("dismiss", {"notification_id": _OLD_ID})])
_off_s, _, _ = run(_scenario_fixed(price_provider="off",
                                   static_profile=_PROFILE))
check("provider off with a profile: no log line", _off_s == [])


async def _scenario_fixed_level_and_raise():
    """The line's level, and a dismiss that raises: the cycle goes on."""
    with Wired(NOW, static_profile=_PROFILE) as W:
        real = W.ctl.hass.call

        async def call(domain, service, data):
            if service == "dismiss" and data.get("notification_id") == _OLD_ID:
                raise RuntimeError("Home Assistant said no")
            return await real(domain, service, data)
        W.ctl.hass.call = call
        before = mark()
        await W.refresh()
        try:
            for k in range(2):
                await W.tick(NOW + timedelta(seconds=30 * k))
            done = True
        except Exception as e:  # noqa: BLE001 - the check reports it
            done = repr(e)
        mine = [(r.levelno, r.getMessage()) for r in TAP.records[before:]
                if "only used if no prices" in r.getMessage()]
        return done, mine


_done, _lvl = run(_scenario_fixed_level_and_raise())
check(f"the start line is INFO, not a warning ({_lvl})",
      _lvl == [(logging.INFO, _LINE)])
check(f"a dismiss that raises does not stop the cycle ({_done})", _done is True)
check("the log line is the same words as the note's second half",
      _NOTE.endswith(_LINE[0].lower() + _LINE[1:]))

check("settings page: price_service + a set profile shows the one note, "
      "beside static_profile", _settings_notes(static_profile=_PROFILE) == [_NOTE])
check("...not with no, empty, spaces-only or line-break-only profile",
      all(_settings_notes(**kw) == [] for kw in (
          {}, {"static_profile": ""}, {"static_profile": "   "},
          {"static_profile": "\n \t"})))
check("...not with another provider",
      _settings_notes(price_provider="generic_dayahead",
                      static_profile=_PROFILE) == []
      and _settings_notes(price_provider="off",
                          static_profile=_PROFILE) == [])
_nn = []
validate_startup_config(dict(OPTS_C, price_provider="price_service",
                             price_service_token=TOKEN,
                             static_profile=_PROFILE), None, None, notes=_nn)
_NOTE_NL = ("Met de prijsdienst wordt het vaste tarief alleen gebruikt als er "
            "geen prijzen zijn.")
_en = _UP.anchors([], _nn, {"price_provider": "price_service"},
                     ["price_provider", "static_profile"], "en")
_nl = _UP.anchors([], _nn, {"price_provider": "price_service"},
                     ["price_provider", "static_profile"], "nl")
check("the page puts the note beside the fixed-tariff field, in English and "
      f"Dutch ({_en})",
      [m["text"] for m in _en] == [_NOTE] and [m["text"] for m in _nl]
      == [_NOTE_NL] and _en[0]["tier"] == "info" and _en[0]["field"]
      and _en[0]["field"] == _nl[0]["field"])
check("...and the note names no refusal: the options are still accepted",
      problems(static_profile=_PROFILE) == [])


async def _scenario_profile_roles(profile):
    """Prices up: what the cheap window and the gate say. Then every source
    down: what the plan falls back to."""
    out = {}
    over = {"static_profile": profile} if profile else {}
    with Wired(NOW, **over) as W:
        c = W.ctl
        await W.refresh()
        await W.tick()
        out["cheap"], out["gate"] = c.cheap, c._charge_gate_open(c.inverters[0])
        out["words"] = c.tariff_words
        out["origin_up"] = c.tariff_plan.origin
        W.net.down()
        later = NOW + timedelta(days=5)          # saved prices run out
        await W.refresh(later)
        await W.tick(later)
        out["origin_down"] = c.tariff_plan.origin
        out["gate_down"] = c._charge_gate_open(c.inverters[0])
        out["points"] = c.tariff_plan.read.match(
            series=lambda r: [pt.eur_per_kwh for pt in r.points][:2],
            no_data=lambda r: [])
    return out


_with = run(_scenario_profile_roles(_PROFILE))
_without = run(_scenario_profile_roles(None))
check("with prices up, the cheap window and the gate are the same with or "
      "without a fixed tariff set",
      (_with["cheap"], _with["gate"], _with["words"])
      == (_without["cheap"], _without["gate"], _without["words"])
      and _with["origin_up"] == "provider")
check("with every source down, the plan falls back to the fixed tariff "
      f"({_with['origin_down']}, {_with['points']}) and without it there is none",
      _with["origin_down"] == "fallback" and _with["points"]
      and set(_with["points"]) == {0.25}
      and _without["origin_down"] == "none")
check("...and the gate stays shut on that fallback (it is not a firm price)",
      _with["gate_down"] is False)

# =========================================================================== #
print("\n15. round 2: a token shorter than 16 characters is refused (F5)")
_short = problems(price_service_token="a" * 15)
check(f"15 characters: refused with one short plain sentence ({_short})",
      _short == ["“Price service token” (price_service_token) is too short. "
                 "Paste the whole token."])
check("...16 characters start", problems(price_service_token="a" * 16) == [])
check("...the refusal does not echo the value",
      "a" * 15 not in "".join(_short))

# =========================================================================== #
print("\n16. round 2: the guards the reviewer's mutations R01 and R10 got past")
_t0 = local_midnight(date(2026, 10, 14))
_now0 = _t0 + timedelta(minutes=31)                       # inside period 2
_pts = [P.PricePoint(_t0 + timedelta(minutes=15 * i), v, i == 2)
        for i, v in enumerate((0.10, 0.20, 0.05, 0.40))]
_prov = P.PriceRead.series(_pts, PriceProduct.DAY_AHEAD, "price_service", NOW)
_g = C.gate_for(_prov, _now0)
check("R01: if the quarter holding now is an estimate it is not priced as "
      "firm, even though it is cheap: the gate is shut and says 'no price'",
      _g.cheap is False and _g.covered is False
      and _g.words == "No price for now")
_firm = P.PriceRead.series(
    [P.PricePoint(p.start, p.eur_per_kwh, False) for p in _pts],
    PriceProduct.DAY_AHEAD, "price_service", NOW)
check("...the same series with every quarter firm opens (it is cheap), so "
      "only the estimate flag shut it",
      C.gate_for(_firm, _now0).cheap is True
      and C.gate_for(_firm, _now0).covered is True)
_prov2 = P.PriceRead.series(
    [P.PricePoint(_t0 + timedelta(minutes=15 * i), v, i == 3)
     for i, v in enumerate((0.10, 0.15, 0.05, 0.40))],
    PriceProduct.DAY_AHEAD, "price_service", NOW)
check("an estimated quarter elsewhere in the day stays out of today's "
      "average: 0.15 against 0.10, 0.15, 0.05 is above their average (0.10), "
      "where with the estimated 0.40 counted it would be cheap (0.175)",
      C.gate_for(_prov2, _t0 + timedelta(minutes=16)).cheap is False)


async def _plain_wire():
    from aiohttp import web, ClientResponseError
    seen, elsewhere = [], []

    async def ok(request):
        seen.append(dict(request.headers))
        return web.Response(body=b'{"ok": 1}')

    async def moved(request):
        raise web.HTTPFound("/elsewhere")

    async def other(request):
        elsewhere.append(1)
        return web.Response(body=b'{"ok": 2}')

    async def missing(request):
        return web.Response(status=404, text="no")

    async def broken(request):
        return web.Response(status=503, text="no")
    app = web.Application()
    for path, fn in (("/ok", ok), ("/moved", moved), ("/elsewhere", other),
                     ("/missing", missing), ("/broken", broken)):
        app.router.add_get(path, fn)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    base = f"http://127.0.0.1:{site._server.sockets[0].getsockname()[1]}"
    res = {}
    try:
        res["ok"] = await P.https_get(base + "/ok", 3)
        res["moved"] = await P.https_get(base + "/moved", 3)
        for name in ("missing", "broken"):
            try:
                await P.https_get(base + "/" + name, 3)
                res[name] = "no error"
            except ClientResponseError as e:
                res[name] = e.status
            except Exception as e:                            # noqa: BLE001
                res[name] = "other " + type(e).__name__
    finally:
        await runner.cleanup()
    res["headers"] = seen[0] if seen else {}
    res["followed"] = len(elsewhere)
    return res


_pw = run(_plain_wire())
check("R10: the plain transport the other providers use returns the body "
      "of a 200", _pw["ok"] == b'{"ok": 1}')
check("...follows a redirect, as it always did (the price service's own "
      "path does not)",
      _pw["moved"] == b'{"ok": 2}' and _pw["followed"] == 1)
check("...raises the library's own error for a 4xx and a 5xx, not the price "
      "service's words",
      _pw["missing"] == 404 and _pw["broken"] == 503)
check("...and sends no Authorization header of its own",
      "Authorization" not in _pw["headers"])

for d in _TMP:
    shutil.rmtree(d, ignore_errors=True)

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