goodwe-addon/goodwe_controller/test_messages.py

"""UX-08: user-facing text follows STYLE.md.   RUN: python3 -B test_messages.py

1. Every notification in `app/messages.py` obeys the rule. The test renders
   each one with realistic values and checks the title length, the number of
   sentences and words for its severity, the banned terms, the entity ids and
   the Details line.
2. No notification is worded anywhere else. An AST scan of `app/` fails any
   `persistent_notification` call whose payload is not built by
   `app/messages.py`.
3. The status page. Every row and banner in `app/messages.py` obeys the rule
   for its colour: a row one sentence of at most 15 words, a red row up to 3
   sentences and 30 words, a banner at most 12 words.
   THE RATCHET for the kinds not rewritten yet. Every status-row and banner
   literal left in `app/`, every validator refusal, every WARNING+ log line and every option
   name and description in `translations/en.yaml` must either obey its limit
   or appear in `ux08_allowlist.txt`. A new string that breaks the rule and is
   not listed goes red. So does a listed entry that no longer matches a
   failing string, so the list can only shrink: remove the line when you
   rewrite the string.
4. DOCS.md has a section for every notification, under "## Notifications".
5. Mutations: each rule is shown to catch a spoiled input (habit 1 of
   MUTATION.md).

Run `python3 -B test_messages.py --violations` to print what part 3 sees.
"""

import ast
import hashlib
import json
import os
import re
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE / "tools"))

from app import messages                                          # noqa: E402
import translations_lint as TL                                    # noqa: E402

fails = []


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


# --- the rule (STYLE.md sections 1 and 4): tools/style_rules.py, shared with
# test_ui_text.py since UX-13 P5.
from style_rules import (BANNED, ENTITY, PLACEHOLDER, SHOUT, SNAKE,  # noqa: E402,F401
                         sentences, text_problems, title_problems, words)


def body_problems(body, severity):
    out = []
    n_s, n_w = sentences(body), len(words(body))
    if severity == messages.ALARM:
        if n_s > 3:
            out.append(f"alarm body over 3 sentences ({n_s})")
        if n_w > 70:
            out.append(f"alarm body over 70 words ({n_w})")
        long = [s for s in re.split(r"(?<=[.!?])\s+", body) if len(words(s)) > 30]
        if long:
            out.append(f"sentence over 30 words ({len(words(long[0]))})")
    else:
        if n_s > 1:
            out.append(f"{severity} body over 1 sentence ({n_s})")
        if n_w > 25:
            out.append(f"{severity} body over 25 words ({n_w})")
    if ";" in body:
        out.append("semicolon: split it into sentences or drop a clause")
    return out + text_problems(body, keys=False, entities=False)


def notice_problems(n):
    out = []
    if n.severity not in (messages.ALARM, messages.WARNING, messages.INFO,
                          messages.RESOLVED):
        out.append(f"unknown severity {n.severity!r}")
    out += title_problems(n.title) + body_problems(n.body, n.severity)
    if not ends_in_details(n.data()["message"], n.docs):
        out.append("message does not end in its Details line")
    return out


def ends_in_details(message, docs):
    return message.endswith("\n\n" + messages.details(docs))


# ============================================================================== #
print("1. every notification obeys the rule")
SAMPLES = messages.samples()
for n in SAMPLES:
    p = notice_problems(n)
    check(f"{n.severity:8} {n.title!r}: {len(words(n.body))} words, "
          f"{sentences(n.body)} sentence(s){' - ' + '; '.join(p) if p else ''}",
          not p)
ids = {v for k, v in vars(messages).items() if k.startswith("ID_")}
seen = {re.sub(r"inverter_b$", "{slug}", n.id) for n in SAMPLES}
check(f"the samples cover every notification id in the catalogue "
      f"(missing: {sorted(ids - seen)})", ids <= seen)
check("the payload carries exactly the three keys Home Assistant accepts",
      all(set(n.data()) == {"notification_id", "title", "message"}
          for n in SAMPLES))
# The Details link (UX-09). Home Assistant 2026.2+ serves add-on pages at
# /config/app/<slug>/<tab> and 2026.5 dropped /hassio/addon/<slug>/<tab>, so
# the path follows the version the Supervisor reports; unknown means current.
_LINE = ("Details: [add-on documentation](/config/app/local_goodwe_controller/"
         "documentation), section “Battery is not keeping up” under "
         "Notifications.")
os.environ["HOSTNAME"] = "3f2a9c1b7d4e"          # a container id, as on TEST-23
messages.set_site("local_goodwe_controller", "2026.9.1")
check("with the Supervisor's slug the Details line is exactly the current-HA "
      "link plus the section by name, even when the hostname is a container id",
      SAMPLES[0].data()["message"].endswith("\n\n" + _LINE))
_link = {v: messages.docs_url("local_goodwe_controller", v)
         for v in ("2026.1.3", "2026.2.0", "2026.4.2", "2026.10.0b1",
                   "2025.12.5", None, "rubbish")}
check(f"Home Assistant 2026.1 and older get the old /hassio path, 2026.2 and "
      f"newer and an unknown version the /config/app path ({_link})",
      _link == {
          "2026.1.3": "/hassio/addon/local_goodwe_controller/documentation",
          "2025.12.5": "/hassio/addon/local_goodwe_controller/documentation",
          "2026.2.0": "/config/app/local_goodwe_controller/documentation",
          "2026.4.2": "/config/app/local_goodwe_controller/documentation",
          "2026.10.0b1": "/config/app/local_goodwe_controller/documentation",
          None: "/config/app/local_goodwe_controller/documentation",
          "rubbish": "/config/app/local_goodwe_controller/documentation"})
messages.set_site("../../etc", "2026.9.1")
check("a slug that is not this add-on's shape is not put in a link",
      "](" not in SAMPLES[0].data()["message"])
messages.set_site(None, None)
os.environ["HOSTNAME"] = "abcd1234-goodwe-controller"
check("without the Supervisor's answer the slug comes from a Supervisor-style "
      "hostname",
      "(/config/app/abcd1234_goodwe_controller/documentation)"
      in SAMPLES[0].data()["message"])
os.environ["HOSTNAME"] = "some-laptop"
check("...and anywhere else it names the section without a guessed link",
      SAMPLES[0].data()["message"].endswith(
          "\n\nDetails: add-on documentation, section “Battery is not keeping "
          "up” under Notifications."))
del os.environ["HOSTNAME"]

# Where the slug and the version come from: the Supervisor, read once at start.
import asyncio                                                    # noqa: E402
from app.main import docs_link_facts                              # noqa: E402


class _Resp:
    def __init__(self, body):
        self.body = body

    async def __aenter__(self):
        if isinstance(self.body, Exception):
            raise self.body
        return self

    async def __aexit__(self, *a):
        return False

    async def json(self, content_type=None):
        return self.body


class _Session:
    def __init__(self, answers):
        self.answers, self.asked, self.timeouts = answers, [], []

    def get(self, url, **kw):
        self.asked.append((url, kw.get("headers", {}).get("Authorization")))
        self.timeouts.append(kw.get("timeout"))
        return _Resp(self.answers[url])


_ok = _Session({
    "http://supervisor/addons/self/info": {"result": "ok", "data": {
        "slug": "local_goodwe_controller", "version": "0.12.0"}},
    "http://supervisor/info": {"result": "ok", "data": {
        "homeassistant": "2026.9.1", "supervisor": "2026.08.0"}}})
_down = _Session({"http://supervisor/addons/self/info": OSError("refused"),
                  "http://supervisor/info": ValueError("not json")})
check("at start the add-on asks the Supervisor for its slug and HA's version, "
      "with its own token",
      asyncio.run(docs_link_facts(_ok, "tok")) == {
          "slug": "local_goodwe_controller", "ha_version": "2026.9.1"}
      and all(a == "Bearer tok" for _u, a in _ok.asked))
check("...and a Supervisor that does not answer costs the link, never the "
      "start",
      asyncio.run(docs_link_facts(_down, "tok")) == {
          "slug": None, "ha_version": None})
check("...and each question has a short timeout, so a Supervisor that hangs "
      f"delays control start by seconds, not minutes ({_ok.timeouts})",
      len(_ok.timeouts) == 2
      and all(isinstance(t, (int, float)) and 0 < t <= 3 for t in _ok.timeouts))

# Shadow mode (UX-09): nothing the add-on decides reaches the inverter, so no
# text may say a command was sent or that the battery is giving power.
_c1 = [messages.Clamp(None, 7412, 5000, "rating")]
_c2 = [messages.Clamp("inverter-a", -3000, -2500, "device"),
       messages.Clamp("inverter-b", 6000, 4600, "max_w+device")]
_live = [messages.output_clamp(c).body for c in (_c1, _c2, ())]
_shad = [messages.output_clamp(c, shadow=True).body for c in (_c1, _c2, ())]
check("live, the over-rating alarm says what was sent and that control runs",
      " and sent 5 000 W" in _live[0] and " and sent 2 500 W" in _live[1]
      and all("Control is still running" in b or "control is still running"
              in b for b in _live))
check("in shadow mode it says what would have been sent, and that the "
      "inverters stay at 0 W",
      "would have sent 5 000 W" in _shad[0]
      and "would have sent 2 500 W" in _shad[1]
      and "would have been cut back" in _shad[2]
      and all(" sent" not in b.replace("would have sent", "") for b in _shad)
      and all("0 W" in b and "hadow mode" in b for b in _shad))
_row_live = messages.row_power_too_high(_c1, 3)
_row_shad = messages.row_power_too_high(_c1, 3, shadow=True)
check("the over-rating row says 'sent' live and 'would send' in shadow mode",
      "sent 5 000 W" in _row_live and "would" not in _row_live
      and "would send 5 000 W" in _row_shad and " sent" not in _row_shad
      and "shadow mode" in _row_shad)
_sat = messages.peak_saturated(5000, shadow=True).body
check("in shadow mode the peak alarm does not claim the battery is giving "
      "power, and keeps its action",
      "giving all it can" not in _sat and "gives nothing" in _sat
      and "Switch off large loads" in _sat
      and "giving all it can" in messages.peak_saturated(5000).body)
check("shadow mode changes no notification's id or severity",
      all(messages.output_clamp(c, shadow=True).id == messages.ID_OUTPUT_CLAMP
          and messages.output_clamp(c, shadow=True).severity == messages.ALARM
          for c in (_c1, _c2, ()))
      and messages.peak_saturated(1, shadow=True).id == messages.ID_PEAK_SATURATED
      and messages.peak_saturated(1, shadow=True).severity == messages.ALARM)
check("the output-clamp alarm quotes asked before sent",
      messages.output_clamp([messages.Clamp(None, 7412, 5000, "rating")]).body
      .index("7 412 W") < messages.output_clamp(
          [messages.Clamp(None, 7412, 5000, "rating")]).body.index("5 000 W"))


# ============================================================================== #
print("\n2. no notification is worded outside app/messages.py")
APP = HERE / "app"


# UX-08 2d: a refusal is counted as it is SHOWN. A placeholder that names a
# setting is expanded to the label and key `messages.setting()` prints, so
# `{S('max_w')}` counts as “Inverter rating” (max_w), three words, not one.
SETTING_CALLS = ("S", "_S", "setting", "messages.setting")
CONSTANTS = {}          # module-level NAME = "text", per file, for keys


def _key(node):
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    if isinstance(node, ast.Name):
        return CONSTANTS.get(node.id)
    return None


def expand_setting(node):
    """The shown text of a placeholder that names a setting, else None."""
    if isinstance(node, ast.Call) and node.args:
        name = ast.unparse(node.func)
        if name in SETTING_CALLS:
            key = _key(node.args[0])
            if key not in messages.LABELS:
                return None
            unit = node.args[1] if len(node.args) > 1 else None
            plain = unit is None or (isinstance(unit, ast.Constant)
                                     and not unit.value)
            return messages.setting(key, "" if plain else "inverter-b")
        if name == "_labels" and isinstance(node.args[0], (ast.List, ast.Tuple)):
            keys = [_key(e) for e in node.args[0].elts]
            if all(k in messages.LABELS for k in keys):
                return " and ".join(messages.setting(k) for k in keys)
    if (isinstance(node, ast.Subscript)
            and ast.unparse(node.value) == "messages.LABELS"):
        key = _key(node.slice)
        return messages.LABELS.get(key) if key else None
    return None


def render(node, expand=False):
    """A string expression as text, placeholders as {expr}; None otherwise.
    `expand` shows setting placeholders as the page shows them."""
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    if isinstance(node, ast.JoinedStr):
        out = []
        for v in node.values:
            if isinstance(v, ast.Constant):
                out.append(v.value)
                continue
            shown = expand_setting(v.value) if expand else None
            out.append(shown if shown is not None
                       else "{" + ast.unparse(v.value) + "}")
        return "".join(out)
    if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
        a, b = render(node.left, expand), render(node.right, expand)
        if a is None and b is None:
            return None
        return (a if a is not None else "{…}") + (b if b is not None else "{…}")
    if isinstance(node, ast.IfExp):
        a, b = render(node.body, expand), render(node.orelse, expand)
        if a is None and b is None:
            return None
        return (a or "") + " / " + (b or "")
    return None


def notification_calls(tree):
    for node in ast.walk(tree):
        if (isinstance(node, ast.Call) and node.args
                and ast.unparse(node.func).endswith("call")
                and render(node.args[0]) == "persistent_notification"):
            yield node


bad_calls, n_calls = [], 0
for f in sorted(APP.glob("*.py")):
    for call in notification_calls(ast.parse(f.read_text(encoding="utf-8"))):
        n_calls += 1
        payload = call.args[2] if len(call.args) > 2 else None
        ok = (isinstance(payload, ast.Call)
              and isinstance(payload.func, ast.Attribute)
              and payload.func.attr == "data"
              and "messages." in ast.unparse(payload.func.value))
        if not ok:
            bad_calls.append(f"{f.name}:{call.lineno}")
check(f"the scan found the add-on's notification calls ({n_calls})", n_calls >= 7)
check(f"every one is built by app/messages.py (not: {bad_calls})", not bad_calls)


# ============================================================================== #
print("\n3. the ratchet: other user-facing text obeys its limit or is listed")

LIMITS = {                      # kind: (words, keys ok, entity ids ok)
    "row": (15, False, False),
    "banner": (12, False, False),
    "refusal": (30, True, True),
    "log": (25, True, True),
    "option-name": (6, False, False),
    "option-desc": (30, True, True),
    "note": (25, True, True),           # UX-13: a settings-page warning
    "ui-name": (6, False, False),       # UX-13: app/ui_text, a name or choice
    "ui-desc": (25, False, False),      # UX-13: app/ui_text, one-line help
}
LOG_CALL = re.compile(r"(?:^|\.)(?:_LOG|log|logger|LOG)\.(warning|error|critical|exception)$")


def strings_in(node, expand=False):
    """Top-level string expressions inside `node` (not their parts)."""
    t = render(node, expand)
    if t is not None:
        yield node, t
        return
    for child in ast.iter_child_nodes(node):
        yield from strings_in(child, expand)


def _module_constants(tree):
    """Module-level `NAME = "text"` assignments."""
    return {t.id: n.value.value for n in tree.body
            if isinstance(n, ast.Assign) and isinstance(n.value, ast.Constant)
            and isinstance(n.value.value, str)
            for t in n.targets if isinstance(t, ast.Name)}


_MODCONST = {f.stem: _module_constants(ast.parse(f.read_text(encoding="utf-8")))
             for f in sorted(APP.glob("*.py"))}


def collect():
    """(kind, where, text) for every user-facing literal of a ratcheted kind."""
    out = []
    for f in sorted(APP.glob("*.py")):
        if f.name == "messages.py":
            continue
        tree = ast.parse(f.read_text(encoding="utf-8"))
        # The file's own constants, and those it imports from another app
        # module (`from .validate import TAXES_PROFILE_OPTION`), so a key
        # named by a constant from elsewhere is expanded too.
        CONSTANTS.clear()
        CONSTANTS.update(_module_constants(tree))
        for n in tree.body:
            if isinstance(n, ast.ImportFrom) and n.level == 1 and n.module:
                theirs = _MODCONST.get(n.module, {})
                CONSTANTS.update({a.asname or a.name: theirs[a.name]
                                  for a in n.names if a.name in theirs})
        for node in ast.walk(tree):
            if isinstance(node, ast.Dict):
                for k, v in zip(node.keys, node.values):
                    key = render(k) if k is not None else None
                    if key in ("text", "banner"):
                        for n, t in strings_in(v):
                            out.append(("row" if key == "text" else "banner",
                                        f"{f.name}:{n.lineno}", t))
            elif isinstance(node, ast.Assign) and f.name == "main.py":
                tgt = ast.unparse(node.targets[0]).replace(" ", "").strip("()")
                if tgt in ("level,banner", "banner"):
                    for n, t in strings_in(node.value):
                        if len(words(t)) > 1:
                            out.append(("banner", f"{f.name}:{n.lineno}", t))
            elif isinstance(node, ast.Call):
                name = ast.unparse(node.func)
                if LOG_CALL.search(name) or name.endswith("._moan"):
                    arg = node.args[1 if name.endswith("._moan") else 0] \
                        if node.args else None
                    if arg is not None:
                        for n, t in strings_in(arg):
                            out.append(("log", f"{f.name}:{n.lineno}", t))
                elif name.endswith("_advise") and node.args:
                    for n, t in strings_in(node.args[-1], expand=True):
                        if len(words(t)) > 1:
                            out.append(("note", f"{f.name}:{n.lineno}", t))
                elif re.search(r"problems\.(append|extend)$", name):
                    for a in node.args:
                        for n, t in strings_in(a, expand=True):
                            if len(words(t)) > 1:
                                out.append(("refusal", f"{f.name}:{n.lineno}", t))
    # UX-08 2d: the Dutch option texts are held to the same length.
    for lang in ("en", "nl"):
        tr = TL.parse_translation_file(HERE / "translations" / f"{lang}.yaml")

        def walk(conf, path):
            for key, v in (conf or {}).items():
                if not isinstance(v, dict):
                    continue
                if isinstance(v.get("name"), str):
                    out.append(("option-name", f"{lang}.yaml:{path}{key}",
                                v["name"].split("·")[-1].strip()))
                if isinstance(v.get("description"), str):
                    out.append(("option-desc", f"{lang}.yaml:{path}{key}",
                                v["description"]))
                if isinstance(v.get("fields"), dict):
                    walk(v["fields"], f"{path}{key}.")
        walk(tr.get("configuration"), "")
    # UX-13: the new page's own words, held to the option texts' rule. Home's
    # sentences (`home_words`) are templates, checked rendered in part 8.
    for f in sorted((APP / "ui_text").glob("*.json")):
        def walk_ui(node, path):
            for key, v in node.items():
                if not path and key == "home_words":
                    continue
                if isinstance(v, dict):
                    walk_ui(v, f"{path}{key}.")
                elif isinstance(v, str):
                    out.append(("ui-desc" if key == "description" else
                                "ui-name", f"{f.name}:{path}{key}", v))
        walk_ui(json.loads(f.read_text(encoding="utf-8")), "")
    return out


def problems_of(kind, text):
    lim, keys, ents = LIMITS[kind]
    p = []
    if len(words(text)) > lim:
        p.append(f"over {lim} words ({len(words(text))})")
    if ";" in text:
        p.append("semicolon")
    if kind in ("note", "ui-desc") and sentences(text) > 1:
        p.append(f"over 1 sentence ({sentences(text)})")
    return p + text_problems(text, keys=keys, entities=ents)


def key_of(kind, text):
    return f"{kind} {hashlib.sha1(text.encode('utf-8')).hexdigest()[:12]}"


ALLOW_FILE = HERE / "ux08_allowlist.txt"
# The allowlist may never grow. The test demands the list be exactly this long,
# so deleting lines means lowering this number in the same change, and adding
# one means raising it where a reviewer sees it.
ALLOWLIST_MAX = 0


def read_allowlist(path):
    keys = set()
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.split("#", 1)[0].strip()
        if line:
            keys.add(" ".join(line.split()[:2]))
    return keys


found = collect()
violating = {}
for kind, where, text in found:
    p = problems_of(kind, text)
    if p:
        violating.setdefault(key_of(kind, text), (kind, where, text, p))
if "--violations" in sys.argv:
    for k, (kind, where, text, p) in sorted(violating.items(),
                                             key=lambda kv: kv[1][:2]):
        print(f"{k}  # {where} {'; '.join(p)} :: {text[:70]!r}")
    sys.exit(0)

STATUS = messages.status_samples()


def row_problems(colour, text):
    """STYLE.md section 1 for a status row (by colour) or a banner (None)."""
    n_s, n_w = sentences(text), len(words(text))
    out = []
    if colour is None:
        if n_w > 12:
            out.append(f"banner over 12 words ({n_w})")
    elif colour == messages.RED:
        if n_s > 3:
            out.append(f"red row over 3 sentences ({n_s})")
        if n_w > 30:
            out.append(f"red row over 30 words ({n_w})")
    else:
        if n_s > 1:
            out.append(f"{colour} row over 1 sentence ({n_s})")
        if n_w > 15:
            out.append(f"{colour} row over 15 words ({n_w})")
    if ";" in text:
        out.append("semicolon: split it into sentences or drop a clause")
    return out + text_problems(text, keys=False, entities=False)


for fn, colour, text in STATUS:
    p = row_problems(colour, text)
    check(f"{colour or 'banner':6} {text!r}{' - ' + '; '.join(p) if p else ''}",
          not p)
_catalogue = {k for k, v in vars(messages).items()
              if callable(v) and k.startswith(("row_", "banner_"))}
_missing = sorted(_catalogue - {fn for fn, _c, _t in STATUS})
check(f"the samples cover every row and banner in the catalogue "
      f"(missing: {_missing})", not _missing)
# A red row with no banner of its own heads the page as "Not ready - <its first
# sentence>". The three trips have their own branch in status_payload(), and
# main.py gives the other two a "banner" key (both are sampled above).
_own_banner = {"row_direction_trip", "row_site_direction_trip",
               "row_circulating_trip", "row_peak_reserve", "row_power_too_high"}
_long = [t for fn, c, t in STATUS if c == messages.RED and fn not in _own_banner
         and len(words(messages.banner_not_ready({"text": t}))) > 12]
# The hint under the buttons (UX-08 2d): the last event in plain words, one
# per key of messages.HINTS, held to the rule for a green row.
for key, text in messages.hint_samples():
    p = row_problems(messages.GREEN, text)
    check(f"hint   {text!r}{' - ' + '; '.join(p) if p else ''}", not p)
_calls, _bare = 0, []
for f in sorted(APP.glob("*.py")):
    for node in ast.walk(ast.parse(f.read_text(encoding="utf-8"))):
        if (isinstance(node, ast.Call)
                and ast.unparse(node.func).endswith("log_event")):
            _calls += 1
            h = next((k.value for k in node.keywords if k.arg == "hint"), None)
            if not (isinstance(h, ast.Call)
                    and ast.unparse(h.func) in ("messages.hint", "hint")):
                _bare.append(f"{f.name}:{node.lineno}")
check(f"every event the add-on records gives the page its plain words, from "
      f"messages.hint() ({_calls} calls, without: {_bare})",
      _calls >= 40 and not _bare)
_raw = ("OUTPUT_CLAMP", "BUG SIGNAL", "FAILSAFE", "SIGN_DISAGREEMENT",
        "ALLOCATOR", "UNMET", "ARBITER", "PEAK GUARD", "PEAK_GUARD")
check("no hint carries the raw event codes the page used to show",
      not [t for _k, t in messages.hint_samples()
           if any(r in t for r in _raw)])

check(f"every other red row makes a 'Not ready' banner of at most 12 words "
      f"(too long: {_long})", not _long)

# A latch row must say how it clears: that is the one action it exists for.
_clears = {
    "row_direction_trip": "A restart does not clear this",
    "row_site_direction_trip": "Only starting control clears this",
    "row_circulating_trip": "Only starting control clears this",
    "row_not_delivering": "only stopping and starting control clears this",
}
_unclear = [fn for fn, _c, t in STATUS if fn in _clears and _clears[fn] not in t]
check(f"every latch row says how it clears (missing: {_unclear})",
      not _unclear and {fn for fn, _c, _t in STATUS} >= set(_clears))

allow = read_allowlist(ALLOW_FILE)
by_kind = {}
for kind, *_ in found:
    by_kind[kind] = by_kind.get(kind, 0) + 1
for _fn, colour, _t in STATUS:
    kind = "row" if colour else "banner"
    by_kind[kind] = by_kind.get(kind, 0) + 1
check(f"the scan found every ratcheted kind ({by_kind})",
      all(by_kind.get(k, 0) >= 5 for k in LIMITS))
new = sorted(k for k in violating if k not in allow)
for k in new:
    kind, where, text, p = violating[k]
    print(f"       NEW  {where}: {'; '.join(p)} :: {text[:90]!r}\n"
          f"            (rewrite it to STYLE.md; listing it needs a reason)")
check(f"no new string breaks the rule ({len(new)} new)", not new)
stale = sorted(k for k in allow if k not in violating)
for k in stale:
    print(f"       STALE {k} - that string now obeys the rule or is gone: "
          f"delete its line from ux08_allowlist.txt")
check(f"no allowlist entry is stale, so the list only shrinks "
      f"({len(stale)} stale)", not stale)
check(f"the allowlist is exactly its cap ({len(allow)} entries, cap "
      f"{ALLOWLIST_MAX}): lower the cap when you delete lines",
      len(allow) == ALLOWLIST_MAX)

# A refusal names a setting by its Configuration-tab label (UX-08 2c). The
# catalogue's labels must be the en.yaml names without group prefix and unit.
_TR = TL.parse_translation_file(HERE / "translations" / "en.yaml")["configuration"]


def _label_stems(key):
    names = [c[key]["name"] for c in (_TR, _TR["inverters"]["fields"])
             if isinstance(c.get(key), dict)]
    out = set()
    for n in names:
        n = n.split("·")[-1].strip()
        while re.search(r"\s*\([^()]*\)$", n):
            n = re.sub(r"\s*\([^()]*\)$", "", n)
        out.add(n)
    return out


_drift = sorted(f"{k}: {v!r}" for k, v in messages.LABELS.items()
                if v not in _label_stems(k))
check(f"every label a refusal quotes is the setting's name in en.yaml "
      f"(drifted: {_drift})", not _drift)
check("...and a refusal quotes the label and then the key, so the config "
      "panel still files it under that setting",
      messages.setting("max_w", "inverter-b")
      == "“Inverter rating” for inverter-b (inverters.max_w)")
# UX-08 2d: the refusals as the validator SHOWS them. The scan above expands
# setting labels in a template, but a reason that carries a parser's error
# (`{e}`) is only as long as that error, so the long ones found by the 2c test
# are driven for real here and held to the same 30 words.
from app import custom_formula as _cf                             # noqa: E402
from app.validate import validate_startup_config                 # noqa: E402

_BASE = {"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}
_CARD = _cf._LEGS["price_formula"].example.replace("15.55", "15.05")
_TABLE = ('{"periods":[{"name":"day","from":"07:00","to":"22:00","days":'
          '"mon-fri","eur_per_kwh":0.1},{"name":"night","from":"22:00","to":'
          '"07:00","days":"mon-fri","eur_per_kwh":0.08},{"name":"night",'
          '"from":"00:00","to":"24:00","days":"sat,sun","eur_per_kwh":0.08}]}')
_ROSTER = {k: v for k, v in _BASE.items()
           if k not in ("setpoint_entity", "soc_entity", "batt_entity",
                        "max_w")}
_ROSTER["inverters"] = [
    dict(name=n, setpoint_entity=f"number.sp_{n[-1]}", max_w=w,
         soc_entity=f"sensor.soc_{n[-1]}", batt_entity=f"sensor.batt_{n[-1]}",
         bms_soc_floor=10)
    for n, w in (("inverter-a", 2000), ("inverter-b", 200))]
_DRIVEN = {
    "formula card": dict(_BASE, price_provider="custom_formula",
                         price_formula=_CARD),
    "formula not valid": dict(_BASE, price_provider="custom_formula",
                              price_formula='{"index": "epex_be"} x'),
    "injection card": dict(_BASE, price_export_formula=_CARD),
    "roster thresholds": dict(_ROSTER, maintenance_charge_w=100,
                              maintenance_discharge_w=100),
    "levies table and flat": dict(_BASE, price_taxes_profile=_TABLE,
                                  price_taxes_eur_kwh=0.1198),
    "feed and entities": dict(_BASE, price_provider="generic_dayahead",
                              price_now_entity="sensor.now",
                              price_avg_entity="sensor.avg"),
    "fixed tariff and entities": dict(_BASE, static_profile=_TABLE,
                                      price_now_entity="sensor.now",
                                      price_avg_entity="sensor.avg"),
    "fixed tariff not valid": dict(_BASE, static_profile='{"periods": []} x'),
}
_LEAKS = ("c_eur_per_kwh", "vat_pct", "its own card", "line 1 column",
          "None", "JSON", "one you do not use")
_shown, _long, _leaky = [], [], []
for _name, _o in _DRIVEN.items():
    _ps = validate_startup_config(_o, -5000, 5000)
    _shown.append((_name, len(_ps)))
    _long += [f"{_name}: {len(words(t))} words: {t}" for t in _ps
              if len(words(t)) > 30]
    _leaky += [f"{_name}: {t}" for t in _ps if any(x in t for x in _LEAKS)]
check(f"every driven case is refused ({_shown})",
      all(n >= 1 for _n, n in _shown)
      and dict(_shown)["roster thresholds"] >= 2)
check(f"...and every refusal, as SHOWN, is at most 30 words (too long: "
      f"{_long})", not _long)
check(f"...and shows no parser text, file units or Python's None (found: "
      f"{_leaky})", not _leaky)

# The config panel files a reason beside every setting whose key it names, but
# `inverters.max_w` names a setting of one inverter in the list, not the flat
# `max_w` (UX-08 2d, from the 2c test).
from app.configui import problems_by_field                        # noqa: E402
_by, _un = problems_by_field(
    ["“Inverter rating” for inverter-b (inverters.max_w) is 200 W.",
     "“Inverter rating” (max_w) is 0."], ["max_w", "inverters"])
check("a roster refusal shows beside `inverters` only, and a flat one beside "
      f"`max_w` ({ {k: len(v) for k, v in _by.items()} })",
      _by.get("inverters") == [
          "“Inverter rating” for inverter-b (inverters.max_w) is 200 W."]
      and _by.get("max_w") == ["“Inverter rating” (max_w) is 0."]
      and not _un)

print(f"CI: UX-08 allowlist {len(allow)} entries "
      + ", ".join(f"{k} {sum(1 for x in allow if x.startswith(k + ' '))}"
                  for k in LIMITS))


# ============================================================================== #
print("\n4. DOCS.md holds the detail of every notification")
DOCS = (HERE / "DOCS.md").read_text(encoding="utf-8")
part = DOCS.split("\n## Notifications\n", 1)
check("DOCS.md has a '## Notifications' section", len(part) == 2)
section = part[-1].split("\n## ", 1)[0] if len(part) == 2 else ""
heads = set(re.findall(r"^### (.+)$", section, re.M))
missing = sorted({n.docs for n in SAMPLES} - heads)
check(f"...with a heading for every Details line (missing: {missing})", not missing)
check("DOCS.md explains the status page rows under "
      f"'## {messages.DOCS_STATUS}'", f"\n## {messages.DOCS_STATUS}\n" in DOCS)


# ============================================================================== #
print("\n5. mutations: each rule catches a spoiled input")
N = messages.Notice
check("a six-word title is refused",
      bool(title_problems("The meter reading may be stuck")))
check("a GoodWe prefix is refused", bool(title_problems("GoodWe: meter stuck")))
check("a second sentence in a warning is refused",
      bool(body_problems("The battery lags. Check it.", messages.WARNING)))
check("a semicolon run-on is refused, in an alarm too",
      bool(body_problems("It stopped; check it.", messages.ALARM)))
check("a fourth sentence in an alarm is refused",
      bool(body_problems("One. Two. Three. Four.", messages.ALARM)))
check("an internal name is refused", bool(body_problems(
    "The integrator reached its bound.", messages.WARNING)))
check("a card id is refused", bool(body_problems(
    "See SAFETY-25.", messages.WARNING)))
check("shouting is refused", bool(body_problems(
    "Control is STILL RUNNING.", messages.WARNING)))
check("an entity id in an alarm is refused", bool(body_problems(
    "Check sensor.battery_power now.", messages.ALARM)))
check("...but allowed in a refusal", not problems_of(
    "refusal", "Check sensor.battery_power in Developer Tools."))
check("a semicolon in a refusal or an option text is refused",
      bool(problems_of("refusal", "It is wrong; fix it."))
      and bool(problems_of("option-desc", "It is wrong; fix it.")))
check("a label that drifted from en.yaml is caught",
      "Inverter power" not in _label_stems("max_w"))
check("an option key in an alarm is refused", bool(body_problems(
    "Lower peak_soc_floor.", messages.ALARM)))
check("a message that lost its Details line is refused",
      ends_in_details(N("x", messages.WARNING, "Meter stuck", "It is.")
                      .data()["message"], "Meter stuck")
      and not ends_in_details("It is.", "Meter stuck"))
_fake = "  x = {'text': 'THE METER IS FROZEN AND NOTHING WORKS AT ALL'}\n"
_n = [t for node in ast.walk(ast.parse(_fake.strip()))
      if isinstance(node, ast.Dict) for _, t in strings_in(node.values[0])]
check("a second sentence in an amber row is refused",
      bool(row_problems(messages.AMBER, "Meter stuck. Check it.")))
check("a fourth sentence in a red row is refused",
      bool(row_problems(messages.RED, "One. Two. Three. Four.")))
check("a 16-word green row is refused",
      bool(row_problems(messages.GREEN, " ".join(["word"] * 16))))
check("a 13-word banner is refused",
      bool(row_problems(None, " ".join(["word"] * 13))))
check("an entity id in a status row is refused",
      bool(row_problems(messages.GREEN, "Battery level: sensor.soc = 54")))
check("a semicolon in a status row is refused",
      bool(row_problems(messages.RED, "It stopped; check it.")))
check("a new shouting status row is seen by the scan and is not listed",
      _n and problems_of("row", _n[0])
      and key_of("row", _n[0]) not in allow)
_bad = ast.parse("await self.hass.call('persistent_notification', 'create', "
                 "{'title': 'x', 'message': 'y'})", mode="exec")
check("a hand-written notification payload is seen by part 2",
      any(not (isinstance(c.args[2], ast.Call)) for c in notification_calls(_bad)))


# ============================================================================== #
print("\n6. log lines as they are built at run time (UX-10)")
# Part 3 reads a WARNING+ line's own literal and counts each placeholder as one
# word. So a reason text built elsewhere and passed in as `%s` was never seen:
# the 2d test found "…clamped to +2000 W; … This is a BUG SIGNAL … Cut back 1
# times" in the log while part 3 was green, because `loop_clamp_note()` built
# it. This part closes that gap in two steps.
#  a. Every dynamic part of every WARNING+ line in `app/` is classified here,
#     as data (a number, a name, a path, an exception's own text) or as a text
#     the add-on builds. A new dynamic part that is in neither list fails, so a
#     new builder cannot slip past.
#  b. Every built text is produced by the real code and checked. It is the
#     evidence behind an alarm, so it gets the red-alarm shape: at most 3
#     sentences of at most 30 words, 70 in all. It keeps its reason code. The
#     whole line, as logged, may carry no semicolon, banned term, capitals,
#     "1 times", "(s)" or a doubled full stop.
from datetime import datetime as _dt, timedelta as _td, timezone as _tz  # noqa: E402

from app import clamp as _clamp, freshness as _fresh, peakguard as _pg   # noqa: E402
from app import store as _store                                         # noqa: E402
from app.circulation import CirculationConfig, CirculationWatchdog      # noqa: E402
from app.control import Tuning, compute                                 # noqa: E402
from app.delivery import (DeliveryConfig, DeliveryWatchdog, UnitDelivery,  # noqa: E402
                          delinquent)
from app.freeze import FreezeWatchdog, FreezeWatchdogConfig             # noqa: E402
from app.price import PricePoint, PriceProduct, PriceRead               # noqa: E402
from app.static_tariff import TariffPlan                                # noqa: E402
from app.tariff import Projection                                       # noqa: E402
from app.watchdog import (AGAINST_METER, AGAINST_OWN_POWER, SignWatchdog,  # noqa: E402
                          SignWatchdogConfig)

_T0 = _dt(2026, 9, 27, 9, 0, tzinfo=_tz.utc)
PLURAL = re.compile(r"\b1 [a-z]{2,}s\b(?<!\b1 [a-z]ss)|\w\(s\)")
COUNT_PLURAL = re.compile(r"%d [a-z]{2,}s\b")


def dynamic_parts(tree, fname):
    """(key, call) for every dynamic part of a WARNING+ log call in `tree`."""
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        name = ast.unparse(node.func)
        moan = name.endswith("._moan")
        if not (LOG_CALL.search(name) or moan):
            continue
        i = 1 if moan else 0
        if len(node.args) <= i:
            continue
        fmt = node.args[i]
        parts = [ast.unparse(a) for a in node.args[i + 1:]]
        if isinstance(fmt, ast.JoinedStr):
            parts += [ast.unparse(v.value) for v in fmt.values
                      if isinstance(v, ast.FormattedValue)]
        elif not (isinstance(fmt, ast.Constant) and isinstance(fmt.value, str)):
            parts.append(ast.unparse(fmt))
        for part in parts:
            yield f"{fname}:{part}", node


# a. DATA: shown as it is, never worded by the add-on. An exception's text is
# the library's (hass.py words the Home Assistant ones, UX-10 item 3).
DATA = {
    # hass.py: the _moan() wrapper and what its callers pass
    "hass.py:msg", "hass.py:*args", "hass.py:entity_id", "hass.py:err",
    "hass.py:obj", "hass.py:raw", "hass.py:domain", "hass.py:service",
    "hass.py:resp.status", "hass.py:said", "hass.py:entity", "hass.py:what",
    # UX-13 P2: the entity list's wait, and an error's text or else its name
    "hass.py:STATES_TIMEOUT_S", "hass.py:str(err) or type(err).__name__",
    # main.py
    "main.py:'cancelled' if task.cancelled() else repr(exc) if exc else 'returned'",
    "main.py:EXIT_CONTROL_DIED", "main.py:first.name", "main.py:OPTIONS_PATH",
    "main.py:err", "main.py:', '.join(sorted(stranded))",
    "main.py:sorted(on_roster)", "main.py:inv.name", "main.py:topic_s",
    "main.py:more",
    "main.py:inv.batt_fresh.bound.release_s", "main.py:inv.batt_fresh.bound.move_w",
    "main.py:inv.batt_fresh.bound.unchanged_s", "main.py:inv.batt_fresh.held",
    "main.py:value", "main.py:float(value)",
    "main.py:entity", "main.py:WRITE_BUDGET_S",
    "main.py:_e",
    "main.py:messages.times(self.clamp_wd.engagements)",
    "main.py:messages.when(brussels(self.tariff_at))",
    "main.py:messages.duration(TARIFF_MAX_AGE_S)",
    "main.py:messages.count(len(targets), 'inverter')",
    "main.py:messages.seconds(self.p1.max_age_s)",
    # maintenance.py, mqtt.py, p1.py
    "maintenance.py:phase", "mqtt.py:err", "p1.py:err", "p1.py:self.host",
    "p1.py:self.topic", "p1.py:poll_s", "p1.py:ingest.max_age_s",
    "p1.py:backoff", "p1.py:self.url", "p1.py:str(err) or type(err).__name__",
    "p1.py:self.poll_s",
    # price.py
    "price.py:doc.get('deprecated')", "price.py:len(dropped)",
    "price.py:messages.count(len(times), 'time')", "price.py:plugin_id",
    "price.py:override",
    "price.py:', '.join((str(i) for i in dropped[:5])) + (' and more' if len(dropped) > 5 else '')",
    "price.py:*(d.isoformat() for d in brussels_date_range(since, until))",
    "price.py:pid",
    "price.py:'/'.join(sorted((c.value for c in caps.sources - {Component.ENERGY})))",
    "price.py:' and '.join(ignored) + (' is' if len(ignored) == 1 else ' are')",
    "price.py:getattr(cls, 'id', cls)", "price.py:e", "price.py:type(e).__name__",
    # price_history.py, selfconfig.py, static_tariff.py, store.py, tariff.py
    "price_history.py:self.path", "price_history.py:day_words(want)",
    "price_history.py:messages.count(len(points), 'price')",
    "price_history.py:err",
    "selfconfig.py:messages.count(len(problems), 'problem')",
    "selfconfig.py:', '.join(keys) or 'none this page also edited'",
    "static_tariff.py:e", "store.py:err", "store.py:key", "store.py:raw",
    "tariff.py:self.path", "tariff.py:err", "tariff.py:start", "tariff.py:month",
    # validate.py: option values and keys (the refusals are part 3's)
    "validate.py:integ_w", "validate.py:max_w", "validate.py:disch_w",
    "validate.py:maint_where", "validate.py:maint_cap", "validate.py:pid",
    "validate.py:i", "validate.py:TAXES_PROFILE_OPTION", "validate.py:negative",
    "validate.py:EXPORT_FORMULA_OPTION", "validate.py:xspec.passthrough",
    "validate.py:xspec.offset_eur_kwh", "validate.py:'=' * 72",
    "validate.py:'-' * 72",
    "web.py:err",
}


# b. BUILT TEXT: each producer, run for real. A reason that is only ever
# stored and restored (a saved safety stop) is produced by the same builder.
def _sign(against, cmd=1500.0, got=-1400.0):
    wd = SignWatchdog(SignWatchdogConfig(min_w=250.0, hold_s=60.0),
                      against=against)
    for s in (0, 61):
        wd.evaluate(cmd, got, True, _T0 + _td(seconds=s))
    return wd.reason


def _sign_texts():
    return [_sign(AGAINST_METER), _sign(AGAINST_OWN_POWER, -1100.0, 2000.0)]


def _circ_texts():
    wd = CirculationWatchdog(CirculationConfig(min_w=250.0, hold_s=60.0))
    for s, w in ((0, 5200.0), (30, 6000.0), (61, 5800.0)):
        wd.evaluate(w, _T0 + _td(seconds=s))
    return [wd.reason]


def _delivery_texts():
    units = [UnitDelivery("inverter-a", 1000.0, 990.0),
             UnitDelivery("inverter-b", 2000.0, 0.0)]
    sentence = delinquent(units, 200.0)["inverter-b"]
    wd = DeliveryWatchdog(DeliveryConfig(hold_s=120.0))
    for s in (0, 121):
        wd.evaluate(sentence, _T0 + _td(seconds=s))
    return [wd.reason]


def _freeze_texts():
    out = []
    for dev in (None, 4700.0):
        wd = FreezeWatchdog(FreezeWatchdogConfig(hold_s=600.0))
        wd.evaluate(3600.0, 4480.0 if dev else 5000.0, 5000.0, _T0,
                    dev_min=-dev if dev else None, dev_max=dev)
        out.append(wd.reason)
    return out


def _stale_texts():
    one = _fresh.Freshness(_fresh.BOUNDS["batt"])
    for t in range(0, 40):
        one.submit(-1200.0, -1200.0 if t < 1 else 1300.0, float(t))
    two = _fresh.Freshness(_fresh.BOUNDS["batt"])
    for t in range(0, 40):
        two.submit(-1200.0, -1200.0 if t < 1 else 1300.0, float(t))
    for t in range(40, 80):
        two.submit(-1200.0, 0.0, float(t))
    for t in range(80, 130):
        two.submit(-1200.0, 1300.0, float(t))
    return [one.reason, two.reason]


def _clamp_texts():
    over = compute(prev_w=1000.0, grid_w=9000.0, actual_w=1000.0,
                   tuning=Tuning(max_w=5000.0, slew_w=1000.0, gain=0.6,
                                 integrator_max_w=9000.0), i_w=8000.0)
    at = compute(prev_w=1000.0, grid_w=9000.0, actual_w=1000.0,
                 tuning=Tuning(max_w=5000.0, slew_w=1000.0, gain=0.6),
                 i_w=4500.0)
    write = [(n, _clamp.clamp_to_rating(w, 5000.0, -2500.0, 2500.0))
             for n, w in (("inverter-a", 3000.0), ("inverter-b", -7412.0))]
    # One inverter is not named (main.py passes None), several are.
    return [_clamp.loop_clamp_note(over), _clamp.loop_clamp_note(at),
            _clamp.write_clamp_note([(None, write[0][1])]),
            _clamp.write_clamp_note(write[:1]), _clamp.write_clamp_note(write)]


def _peak_texts():
    proj = Projection(projected_w=4338.0, elapsed_w=4338.0, elapsed_s=799.0,
                      covered_frac=1.0, complete=False, block_start=_T0,
                      held_w=0.0)
    kw = dict(cap_w=3500.0, margin_w=0.0, reading=None, soc_floor=30.0, max_hold_s=30.0)
    sat = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=80.0, **kw)
    low = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=20.0, **kw)
    unread = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=None, **kw)
    return [sat.reason, low.reason, unread.reason]


def _migration_texts():
    out = []
    for moved, kept, lost, wrong in (
            (2, 0, 0, 0), (2, 1, 1, 0), (2, 0, 0, 1), (0, 2, 0, 0),
            (0, 2, 1, 0)):
        rep = _store.MigrationReport()
        rep.ran = moved > 0 or lost > 0
        rep.from_version = 1
        rep.roster = ["inverter-a"]
        rep.moved = {f"unit/inverter-a/k{i}": True for i in range(moved)}
        rep.kept = [f"unit/inverter-a/old{i}" for i in range(kept)]
        for i in range(lost):
            note = f"unit/inverter-a/k{i} was written as True and reads back as None"
            rep.problems.append(note)
            rep.lost.append(note)
        for i in range(wrong):
            note = (f"unit/garage/k{i} was migrated into namespace 'garage', "
                    f"which is not in the inverter list")
            rep.problems.append(note)
            rep.misdirected.append(note)
        out.append(rep.explain())
    return out


def _tariff_texts():
    hours = [_T0 + _td(minutes=15 * i) for i in range(8)]
    pts = [PricePoint(h, 0.2, i >= 4) for i, h in enumerate(hours)]
    read = PriceRead.series(pts, PriceProduct.DAY_AHEAD, "x", _T0)
    return [TariffPlan(read, None, "provider").words,
            TariffPlan(PriceRead.no_data("the price feed timed out"), None,
                       "none").words]


def _static(spec):
    """Templates for a family of builders that is not one function: the first
    argument of every call to `spec`, or every value assigned to `spec`, in
    the named file (placeholders count as one word, as in part 3)."""
    fname, kind, name = spec
    out = []
    for f in ([APP / fname] if fname != "*" else sorted(APP.glob("*.py"))):
        tree = ast.parse(f.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if kind == "call" and isinstance(node, ast.Call) and node.args \
                    and ast.unparse(node.func).split(".")[-1] == name:
                out += [t for _n, t in strings_in(node.args[0])]
            elif kind == "assign" and isinstance(node, ast.Assign) \
                    and ast.unparse(node.targets[0]) == name:
                out += [t for _n, t in strings_in(node.value)]
            elif kind == "return" and isinstance(node, ast.FunctionDef) \
                    and node.name == name:
                out += [t for r in ast.walk(node) if isinstance(r, ast.Return)
                        and r.value is not None for _n, t in strings_in(r.value)]
    return [t for t in out if len(words(t)) > 1]


# UX-11: a WARNING+ line quotes a safety reason by its code only. The reason,
# with its figures, is on a DEBUG line beside it and in the action log.
SAFETY_REASONS = {
    "sign": _sign_texts, "circulation": _circ_texts,
    "delivery": _delivery_texts, "clamp": _clamp_texts, "peak": _peak_texts,
    "freeze": _freeze_texts, "stale": _stale_texts,
}


def _coded(maker):
    """The codes a WARNING+ line shows for the reasons `maker` builds, plus
    the fallback for a restored stop with no reason saved."""
    return lambda: ([messages.reason_code(t) for t in maker()]
                    + [messages.reason_code("")])


TEXT = {
    "main.py:messages.reason_code(first.sign_wd.reason)": _coded(_sign_texts),
    "main.py:messages.reason_code(inv.sign_wd.reason)": _coded(_sign_texts),
    "main.py:messages.reason_code(self.site_sign_wd.reason)":
        _coded(_sign_texts),
    "main.py:messages.reason_code(self.circ_wd.reason)": _coded(_circ_texts),
    "main.py:messages.reason_code(inv.delivery_wd.reason)":
        _coded(_delivery_texts),
    "main.py:messages.reason_code(self.clamp_wd.reason)": _coded(_clamp_texts),
    "main.py:messages.reason_code(g.reason)": _coded(_peak_texts),
    "main.py:messages.reason_code(self.freeze_wd.reason)":
        _coded(_freeze_texts),
    "main.py:messages.reason_code(inv.batt_fresh.reason)": _coded(_stale_texts),
    "main.py:rep.explain()": _migration_texts,
    "store.py:rep.explain()": _migration_texts,
    "main.py:plan.words": _tariff_texts,
    # OPT-11 P2: why an MQTT command was ignored.
    "main.py:why": lambda: list(messages.PACK_REFUSED.values()),
    # Families: every builder of the text, as a template.
    "main.py:refusal": lambda: (_static(("main.py", "return", "write_refusal"))
                                + _static(("allocator.py", "return", "authorises"))),
    "main.py:' / '.join(resolution.scope_errors)":
        lambda: _static(("arbiter.py", "return", "scope_violations"))
        + _static(("arbiter.py", "call", "append")),
    "price_forecast.py:gate.reason": lambda: _static(("*", "call", "block")),
    "static_tariff.py:read.reason": lambda: _static(("*", "call", "no_data")),
    "price_history.py:self.last_error":
        lambda: _static(("price_history.py", "assign", "self.last_error")),
    "price_history.py:bad": lambda: _static(("price_history.py", "return",
                                             "_parse_points")),
    "maintenance.py:msg": lambda: _static(("maintenance.py", "assign", "msg")),
    "validate.py:aged": lambda: _static(("*", "return", "stale_card")),
    "validate.py:xaged": lambda: _static(("*", "return", "stale_card")),
    # Refusals: part 3 holds them to the refusal rule already.
    "selfconfig.py:problems[0]": None,
    "validate.py:problem": None,
    "p1.py:problem": None,
}


# UX-11: a WARNING+ line, as logged with its built text filled in, is at most
# 25 words (STYLE.md). These families are not safety reasons and were not
# rewritten by UX-11. Each is listed with why, the list may only shrink
# (LONG_LINE_CAP), and a listed key that no longer runs long fails.
LONG_LINE_OK = {
    "main.py:plan.words": "tariff plan words, TARIFF-11 pins them, follow-up",
    "static_tariff.py:read.reason": "price feed reasons, follow-up",
    "main.py:rep.explain()": "state migration report, names saved keys",
    "store.py:rep.explain()": "state migration report, names saved keys",
    "main.py:refusal": "write refusals, a software fault only",
    "main.py:' / '.join(resolution.scope_errors)":
        "claim scope errors, a software fault only",
    "maintenance.py:msg": "maintenance gate reasons, follow-up",
    "price_forecast.py:gate.reason": "trading gate reasons, follow-up",
    "price_history.py:self.last_error": "price history errors, follow-up",
}
LONG_LINE_CAP = 9
LINE_WORDS = 25


def reason_problems(text):
    """STYLE.md's red-alarm shape, for a text a log line quotes."""
    out = []
    n_s, n_w = sentences(text), len(words(text))
    if n_s > 3:
        out.append(f"over 3 sentences ({n_s})")
    if n_w > 70:
        out.append(f"over 70 words ({n_w})")
    long = [x for x in re.split(r"(?<=[.!?])\s+", text) if len(words(x)) > 30]
    if long:
        out.append(f"a sentence over 30 words ({len(words(long[0]))})")
    return out + line_problems(text)


def logged_shape(line):
    """UX-10 review: the line as logged, wrapper and reason together, has the
    red-alarm shape too. The reason alone passing is not enough: the 2d-style
    "title: reason. tail. tail." line read as 4 sentences with a 3-sentence
    reason inside it."""
    out = []
    if sentences(line) > 3:
        out.append(f"over 3 sentences ({sentences(line)})")
    long = [x for x in re.split(r"(?<=[.!?])\s+", line) if len(words(x)) > 30]
    if long:
        out.append(f"a sentence over 30 words ({len(words(long[0]))})")
    return out


def line_problems(text):
    """What no WARNING+ line may carry, whatever its length."""
    out = []
    if ";" in text:
        out.append("semicolon")
    if re.search(r"(?<!\.)\.\.(?!\.)|\.{4,}|\.:|:\s*\.", text):
        out.append("a doubled full stop")
    if PLURAL.search(text):
        out.append("plural: " + PLURAL.search(text).group(0))
    return out + text_problems(text, keys=True, entities=True)


def _fill(template, fmt_node, key_part, sample):
    """The line as logged: `sample` in the text slot, `X` in the others."""
    if isinstance(fmt_node, ast.JoinedStr):
        return template.replace("{" + key_part + "}", sample)
    return template  # %-style: filled by the caller


def scan_log_lines(files, text=None, data=None, long_ok=None, long_seen=None):
    """-> (unclassified keys, stale keys, [(key, sample, problems)])."""
    text = TEXT if text is None else text
    data = DATA if data is None else data
    long_ok = LONG_LINE_OK if long_ok is None else long_ok
    long_seen = set() if long_seen is None else long_seen
    seen, bad = set(), []
    for fname, tree in files:
        for key, call in dynamic_parts(tree, fname):
            seen.add(key)
            if key in data or key not in text:
                continue
            maker = text[key]
            if maker is None:
                continue
            i = 1 if ast.unparse(call.func).endswith("._moan") else 0
            fmt = call.args[i]
            template = render(fmt) or ""
            args = [ast.unparse(a) for a in call.args[i + 1:]]
            part = key.split(":", 1)[1]
            for sample in maker():
                p = reason_problems(sample)
                if isinstance(fmt, ast.JoinedStr):
                    line = template.replace("{" + part + "}", sample)
                else:
                    slots = iter(["%s" if a != part else "\0" for a in args])
                    line = re.sub(r"%[-+ #0-9.]*[sdrfgx]",
                                  lambda m: next(slots, m.group(0)), template)
                    line = PLACEHOLDER.sub("X", line).replace("\0", sample)
                p += [f"as logged: {x}" for x in line_problems(line)
                      + logged_shape(line) if x not in p]
                if len(words(line)) > LINE_WORDS:
                    long_seen.add(key)
                    if key not in long_ok:
                        p.append(f"as logged: over {LINE_WORDS} words "
                                 f"({len(words(line))})")
                if p:
                    bad.append((key, sample, p))
    unknown = sorted(seen - set(data) - set(text))
    stale = sorted((set(data) | set(text)) - seen)
    return unknown, stale, bad


_FILES = [(f.name, ast.parse(f.read_text(encoding="utf-8")))
          for f in sorted(APP.glob("*.py"))]
_long_seen = set()
_unknown, _stale, _bad = scan_log_lines(_FILES, long_seen=_long_seen)
for k, sample, p in _bad:
    print(f"       BAD  {k}: {'; '.join(p)} :: {sample!r}")
check(f"every dynamic part of a WARNING+ line is classified as data or as a "
      f"built text (unclassified: {_unknown})", not _unknown)
check(f"...and no classification is left over from a line that is gone "
      f"(stale: {_stale})", not _stale)
_n_texts = sum(len(m()) for m in TEXT.values() if m is not None)
check(f"UX-11: no WARNING+ line runs over {LINE_WORDS} words as logged, except "
      f"the {len(LONG_LINE_OK)} listed families, and none of them is a safety "
      f"reason (cap {LONG_LINE_CAP})",
      len(LONG_LINE_OK) <= LONG_LINE_CAP
      and not any("reason_code" in k
                  or TEXT.get(k) in SAFETY_REASONS.values()
                  for k in LONG_LINE_OK))
check(f"...and every listed family still runs long, so the list only shrinks "
      f"(no longer long: {sorted(set(LONG_LINE_OK) - _long_seen)})",
      set(LONG_LINE_OK) <= _long_seen)


def _code_lines_without_evidence(files):
    """WARNING+ calls quoting `messages.reason_code(X)` whose next statement
    is not a DEBUG call that logs X itself (UX-11: nothing is lost)."""
    out, n = [], 0
    for fname, tree in files:
        for node in ast.walk(tree):
            for body in (getattr(node, f, None) for f in ("body", "orelse")):
                if not isinstance(body, list):
                    continue
                for i, st in enumerate(body):
                    call = st.value if isinstance(st, ast.Expr) else None
                    if not (isinstance(call, ast.Call)
                            and LOG_CALL.search(ast.unparse(call.func))):
                        continue
                    coded = [a.args[0] for a in call.args
                             if isinstance(a, ast.Call) and a.args
                             and ast.unparse(a.func) == "messages.reason_code"]
                    if not coded:
                        continue
                    n += 1
                    nxt = body[i + 1] if i + 1 < len(body) else None
                    dbg = (nxt.value if isinstance(nxt, ast.Expr)
                           and isinstance(nxt.value, ast.Call)
                           and ast.unparse(nxt.value.func).endswith(".debug")
                           else None)
                    want = ast.unparse(coded[0])
                    if dbg is None or not any(
                            ast.unparse(a) == want
                            or ast.unparse(a).startswith(want + " or ")
                            for a in dbg.args):
                        out.append(f"{fname}:{call.lineno}")
    return n, out


_n_coded, _no_evidence = _code_lines_without_evidence(_FILES)
check(f"every WARNING+ line that quotes a reason code ({_n_coded}) has the "
      f"reason itself on a DEBUG line right after it (missing: {_no_evidence})",
      _n_coded >= 14 and not _no_evidence)
_n_m, _m_bad = _code_lines_without_evidence([("x.py", ast.parse(
    "_LOG.error('Stop: %s.', messages.reason_code(wd.reason))\n"
    "_LOG.debug('Something else.')\n"))])
check("...and a line whose DEBUG evidence was dropped is caught",
      _n_m == 1 and _m_bad == ["x.py:1"])
_evidence_bad = [(n, t, reason_problems(t)) for n, m in SAFETY_REASONS.items()
                 for t in m() if reason_problems(t)]
check(f"the safety reasons behind those lines, on the DEBUG line and in the "
      f"action log, keep the red-alarm shape ({_evidence_bad})",
      not _evidence_bad)
_codes = {n: [messages.reason_code(t) for t in m()]
          for n, m in SAFETY_REASONS.items()}
check(f"...and every one carries a code the line can quote ({_codes})",
      all(re.fullmatch(r"[A-Z]+(?:_[A-Z]+)+", c)
          and c not in (messages.NO_REASON_SAVED, messages.NO_REASON_CODE)
          for cs in _codes.values() for c in cs))
check(f"every built text, run for real, obeys the rule, and so does the line "
      f"that quotes it ({_n_texts} texts, {len(_bad)} bad)", not _bad)
_counts = [f"{f}:{n.lineno}" for f, t in _FILES for n in ast.walk(t)
           if isinstance(n, ast.Call) and (LOG_CALL.search(ast.unparse(n.func))
                                           or ast.unparse(n.func).endswith("._moan"))
           for a in n.args[:2] if isinstance(a, ast.Constant)
           and isinstance(a.value, str) and COUNT_PLURAL.search(a.value)]
check(f"no WARNING+ line puts a plural noun after a bare count, which reads "
      f"'1 times' (use messages.times() and friends; found: {_counts})",
      not _counts)

# UX-10 review: the two clamp lines the owner quoted, as logged for one
# inverter, fit the red shape the tight way: at most 3 sentences and 30 words
# in all, code and numbers kept. The real literals from main.py, filled with
# the real notes.
def _clamp_line(opening):
    for n in ast.walk(dict(_FILES)["main.py"]):
        if isinstance(n, ast.Call) and LOG_CALL.search(ast.unparse(n.func)) \
                and n.args and (render(n.args[0]) or "").startswith(opening):
            return n.args[0].value
    return None


_over, _at, _w1, _w1n, _w2 = _clamp_texts()
_clamp_lines = {
    "error, loop": (_clamp_line("Power request too high"), _over),
    "error, write": (_clamp_line("Power request too high"), _w1),
    "warning, loop": (_clamp_line("The battery gives less than asked"), _at),
}
_clamp_shown = {k: (f % (messages.reason_code(r), messages.times(1))
                    if f else None)
                for k, (f, r) in _clamp_lines.items()}
for k, v in _clamp_shown.items():
    print(f"       {k}: {v!r} ({len(words(v or ''))} words)")
check("UX-11: the clamp lines are at most 25 words and keep OUTPUT_CLAMP and "
      "the count, and the numbers are in the reason the DEBUG line beside "
      "them logs",
      all(v and len(words(v)) <= LINE_WORDS
          and "OUTPUT_CLAMP" in v and "1 time so far" in v
          and "+9000" not in v for v in _clamp_shown.values())
      and "software fault" in _clamp_shown["error, loop"]
      and all(x in _over for x in ("+9000 W", "+5000 W", "+2000 W")))

# Mutations: the scan catches what the 2d scan missed.
_OLD_CLAMP = ("OUTPUT_CLAMP: the loop computed +2148 W, over the 2000 W rating, "
              "and it was clamped to +2000 W; the loop's command this cycle was "
              "+440 W. This is a BUG SIGNAL, not a limit being used: something "
              "upstream computed a setpoint the actuator cannot deliver.")
_mut_text = dict(TEXT, **{"main.py:messages.reason_code(self.clamp_wd.reason)":
                          lambda: [_OLD_CLAMP]})
_u, _s, _b = scan_log_lines([f for f in _FILES if f[0] == "main.py"],
                            text=_mut_text)
check("an injected bad reason (the 2d clamp text) is caught, on both clamp "
      f"lines ({sorted({k for k, _t, _p in _b if 'clamp' in k})})",
      [k for k, _t, _p in _b].count(
          "main.py:messages.reason_code(self.clamp_wd.reason)") == 2)
# UX-11: the line that quoted the whole STALE_READING reason (60 words on
# TEST-23) is caught by the length rule alone.
_old_stale = dict(TEXT, **{"main.py:messages.reason_code(inv.batt_fresh.reason)":
                           _stale_texts})
_u, _s, _b = scan_log_lines([f for f in _FILES if f[0] == "main.py"],
                            text=_old_stale)
check("...and the UX-10 stale line, which quoted the whole reason, is caught "
      f"for its length ({[p for k, _t, p in _b if 'batt_fresh' in k][:1]})",
      any("batt_fresh" in k and any(f"over {LINE_WORDS} words" in x for x in p)
          for k, _t, p in _b))
_new_site = ast.parse("_LOG.warning('Something broke: %s', self.new_wd.reason)")
_u, _s, _b = scan_log_lines([("main.py", _new_site)])
check(f"...and a new line quoting a text nobody classified is caught ({_u})",
      _u == ["main.py:self.new_wd.reason"])
check("...and '1 times', '(s)', a semicolon and a doubled full stop in a "
      "logged line are caught",
      all(line_problems(t) for t in ("Cut back 1 times so far.",
                                     "2 malformed point row(s) dropped.",
                                     "It failed; retrying.",
                                     "Cut back to +2000 W.. Only this.",
                                     "index 0, 1, .... Those are left out.",
                                     "moved to one part: . Problems")))
check("...and so is the 2d-style line, a 2-sentence reason inside a title and "
      "two tails, which reads as 4 sentences although the reason passes",
      logged_shape("Power request too high: OUTPUT_CLAMP: cut to +5000 W. The "
                   "command was +1000 W. Control is still running. Cut back "
                   "1 time so far.")
      and not reason_problems("OUTPUT_CLAMP: cut to +5000 W. The command was "
                              "+1000 W"))
check("...while '1 s' and '1 time' are fine",
      not line_problems("Held for 1 s, cut back 1 time so far."))
# INFO lines are not held to the length rule, but the 2d test found "§8.3",
# ISO timestamps and "keys ARE" in them. Their literals may carry none.
INFO_JARGON = re.compile(r"§\s?\d|isoformat|\b(?:ARE|NOT|AND)\b")
_info_bad = [f"{f}:{n.lineno}" for f, t in _FILES for n in ast.walk(t)
             if isinstance(n, ast.Call) and n.args
             and re.search(r"(?:^|\.)(?:_LOG|log|logger|LOG)\.info$",
                           ast.unparse(n.func))
             for _n, s in strings_in(n.args[0])
             if INFO_JARGON.search(ast.unparse(n.args[0])) or SHOUT.search(s)]
check(f"no INFO line carries a section number, an ISO timestamp or capitals "
      f"(found: {_info_bad})", not _info_bad)
check("...and one that did is caught",
      bool(INFO_JARGON.search('"§8.3 price history: %s"'))
      and bool(SHOUT.search("the flat keys ARE NOT that pack's history")))


# ============================================================================== #
print("\n7. log levels, log noise and the display bugs (UX-10)")
import logging                                                    # noqa: E402

from app import hass as _hass, p1 as _p1, validate as _validate, web as _web  # noqa: E402
from app.arbiter import Claim                                     # noqa: E402
from app.main import Controller, why_key                          # noqa: E402
from app import control as _control                              # noqa: E402
from app import main as _main                                    # noqa: E402


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

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


def _captured(logger_name, fn):
    rec, lg = _Rec(), logging.getLogger(logger_name)
    old = lg.level
    lg.addHandler(rec)
    lg.setLevel(logging.DEBUG)
    try:
        fn()
    finally:
        lg.removeHandler(rec)
        lg.setLevel(old)
    return rec.records


def _level_of(fname, words_in_line):
    """The level of the log call in `fname` whose literal holds the words."""
    tree = ast.parse((APP / fname).read_text(encoding="utf-8"))
    for node in ast.walk(tree):
        if isinstance(node, ast.Call) and node.args \
                and re.search(r"_LOG\.\w+$", ast.unparse(node.func)):
            text = render(node.args[0]) or ""
            if words_in_line in text:
                return ast.unparse(node.func).rsplit(".", 1)[-1]
    return None


check("the shadow line of every cycle is DEBUG, not INFO in capitals "
      f"({_level_of('main.py', 'Shadow mode: would send')})",
      _level_of("main.py", "Shadow mode: would send") == "debug"
      and _level_of("main.py", "SHADOW MODE: not commanding") is None)
check("'Shadow mode is on' and 'is off', which follow a setting or a button, "
      f"are INFO ({_level_of('main.py', 'Shadow mode is on')}, "
      f"{_level_of('main.py', 'Shadow mode is off')})",
      _level_of("main.py", "Shadow mode is on") == "info"
      and _level_of("main.py", "Shadow mode is off") == "info")
_access = {}
for _lvl in (logging.INFO, logging.DEBUG):
    logging.getLogger("goodwe.web").setLevel(_lvl)
    _access[logging.getLevelName(_lvl)] = _web.access_log()
logging.getLogger("goodwe.web").setLevel(logging.NOTSET)
_start_src = " ".join(
    ast.unparse(n) for n in ast.walk(ast.parse(
        (APP / "web.py").read_text(encoding="utf-8")))
    if isinstance(n, ast.AsyncFunctionDef) and n.name == "start")
check(f"the page's own polls are not logged at INFO: aiohttp's access log is "
      f"off unless the log level is debug ({_access})",
      _access["INFO"] is None
      and isinstance(_access["DEBUG"], logging.Logger)
      and "access_log=access_log()" in _start_src)
_validate.WARN_ONCE.seen.clear()        # part 3 ran the validator already
_twice = _captured("goodwe.config", lambda: [
    _validate.validate_startup_config(
        dict(_BASE, price_provider="generic_dayahead", static_profile=""),
        -5000, 5000) for _ in range(2)])
_sp = [r for r in _twice if "static_profile is empty" in r.getMessage()]
check(f"a settings warning is logged once per start, though the check runs "
      f"twice ({len(_sp)} line(s))", len(_sp) == 1
      and _sp[0].levelno == logging.WARNING)


def _outage():
    o = _hass.Outage()
    o.down("api", "error 502", "read sensor.x: 502, message='Bad Gateway'")
    o.down("api", "error 502", "read sensor.y: 502")
    o.down("websocket", "the meter connection closed",
           "Received message 8:1000 is not str")
    o.up("api")
    o.up("websocket")
    o.down("api", "no connection", "Cannot connect to host supervisor:80")


_out = _captured("goodwe.hass", _outage)
_warn = [r.getMessage() for r in _out if r.levelno == logging.WARNING]
_info = [r.getMessage() for r in _out if r.levelno == logging.INFO]
check(f"a Home Assistant restart is one plain WARNING line, and one INFO line "
      f"when both connections are back ({_warn}, {_info})",
      _warn[:1] == ["Home Assistant is not answering (error 502), perhaps it "
                    "is restarting. The add-on keeps trying."]
      and _info == ["Home Assistant answers again."] and len(_warn) == 2)
check("...and the library's own text is kept, at DEBUG",
      any("Received message 8:1000" in r.getMessage() for r in _out
          if r.levelno == logging.DEBUG)
      and not any("8:1000" in m or "Bad Gateway" in m for m in _warn + _info))


class _Resp400:
    def __init__(self, body):
        self.status, self.reason, self._body = 400, "Bad Request", body

    async def __aenter__(self):
        return self

    async def __aexit__(self, *a):
        return False

    async def text(self):
        return self._body


class _Session400:
    def __init__(self, body):
        self.body = body

    def post(self, *a, **k):
        return _Resp400(self.body)


def _refused(body):
    ha = _hass.HomeAssistant(_Session400(body), token="t")
    ok = []
    recs = _captured("goodwe.hass", lambda: ok.append(asyncio.run(
        ha.call("number", "set_value", {"entity_id": "number.sp", "value": 1}))))
    return (ok[0] if ok else None), [r.getMessage() for r in recs
                                     if r.levelno == logging.WARNING]


def _one(res):
    """(write result, the one WARNING line) or a named absence that fails."""
    ok, lines = res
    return ok, (lines[0] if len(lines) == 1 else f"<{len(lines)} lines>")


_bare_ok, _bare = _one(_refused("400: Bad Request"))
_full_ok, _full = _one(_refused(
    '{"message": "Value 6000 for number.sp is outside valid range"}'))
check(f"a write Home Assistant refuses while it loads is one plain WARNING, "
      f"still a failed write, without the empty '400: Bad Request' ({_bare!r})",
      _bare_ok is False
      and _bare.startswith("Home Assistant refused number.set_value for "
                           "number.sp (error 400, no details).")
      and "Bad Request" not in _bare)
check(f"...while a body that says more is still quoted ({_full!r})",
      _full_ok is False and "outside valid range" in _full)


class _WS:
    def __init__(self, kind):
        self.kind = kind

    async def receive(self):
        return type("M", (), {"type": self.kind, "data": "{}"})()


def _closed_ws():
    try:
        asyncio.run(_p1._receive_json(_WS(_p1.aiohttp.WSMsgType.CLOSE)))
    except ConnectionResetError as e:
        return str(e)
    return None


check(f"a socket Home Assistant closed is a plain ConnectionResetError, which "
      f"the reconnect loop reports through that one line ({_closed_ws()!r})",
      (_closed_ws() or "").startswith("Home Assistant closed the connection"))

# "Why: Other". Every winner and every safety reason the code can produce has
# words. The keys are found in the code: each `Claim.set(source, ..., reason)`
# through main.why_key(), plus main.py's own refusal to split ("unsplit").
_SOURCES = {}
for _f in sorted(APP.glob("*.py")):
    _tree = ast.parse(_f.read_text(encoding="utf-8"))
    _consts = _module_constants(_tree)
    for _n in ast.walk(_tree):
        if isinstance(_n, ast.Call) and ast.unparse(_n.func) == "Claim.set" \
                and _n.args:
            _src = _n.args[0]
            _src = (_src.value if isinstance(_src, ast.Constant)
                    else _consts.get(getattr(_src, "id", ""), None))
            _rsn = next((k.value for k in _n.keywords if k.arg == "reason"),
                        _n.args[3] if len(_n.args) > 3 else None)
            _rsn = _rsn.value if isinstance(_rsn, ast.Constant) else ""
            _SOURCES.setdefault((_src, _rsn), f"{_f.name}:{_n.lineno}")
_keys = {why_key(s, [Claim.set(s, 0, 0.0, r)]) for s, r in _SOURCES}
_keys |= {why_key("failsafe", []),
          why_key("failsafe", [Claim.set("loop", 0, 1.0, "tracking")])}
# SAFETY-52: the loop's claim reason is computed, so the scan cannot see it.
_keys |= {why_key("loop", [Claim.set("loop", 0, 0.0, _r)])
          for _r in (_control.REFUSED, _control.PROBING)}
check("...including the loop's two battery-refuses reasons, which get words of "
      "their own", {_control.REFUSED, _control.PROBING} <= _keys)
check(f"the scan found the claims ({len(_SOURCES)}), including the latched "
      f"per-inverter stop the 2d test saw as 'Other'",
      ("failsafe", "sign disagreement") in _SOURCES and len(_SOURCES) >= 8)
_missing_why = sorted(k for k in _keys if k not in messages.WHY_WINNER)
check(f"every winner and safety reason has words in messages.WHY_WINNER "
      f"(missing: {_missing_why}, keys: {sorted(_keys)})", not _missing_why)
check("...so the latched stop reads as a safety stop, not 'Other' "
      f"({messages.why(why_key('failsafe', [Claim.set('failsafe', 0, 0.0, 'sign disagreement')]), 0)!r})",
      messages.why(why_key("failsafe", [Claim.set("failsafe", 0, 0.0,
                                                  "sign disagreement")]), 0)
      == "Safety stop, power went the wrong way: 0 W")
_without = {k: v for k, v in messages.WHY_WINNER.items()
            if k != "sign disagreement"}
check("...and a table that lost that entry is caught",
      any(k not in _without for k in _keys))


class _H:
    async def number(self, *a, **k):
        return None

    async def limits(self, *a):
        return (-2000.0, 2000.0, 1.0)

    async def call(self, *a, **k):
        return True

    async def set_number(self, *a, **k):
        return True


class _S:
    data = {}

    def set(self, *a):
        pass

    def get_time(self, *a):
        return None


class _P:
    enabled = False

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


_c = Controller({"setpoint_entity": "number.sp", "meter_source": "off",
                 "max_w": 2000, "stale_input_s": 15}, _H(), _S(), _P(),
                stop=asyncio.Event())
_c.p1_enabled = True
_c.p1 = _p1.P1Ingest(max_age_s=30.0)


def _meter_row():
    """The meter row, or a named absence that fails every check below."""
    rows = [r for r in _c.checks() if r.get("id") == "meter"]
    return rows[0] if len(rows) == 1 else {
        "ok": None, "warn": None, "text": f"<{len(rows)} meter rows>"}


_r0 = _meter_row()
_c.p1.started_mono -= 31.0
_r1 = _meter_row()
_c.p1.submit(_p1.make_sample(_p1.SOURCE_HA, 500.0, 0.0, phases=1))
_r2 = _meter_row()
check(f"right after a start, before the first meter reading and inside its "
      f"limit, the meter row is amber and waits, not red at 0 s ({_r0})",
      _r0["warn"] and not _r0["ok"]
      and _r0["text"] == "Meter: waiting for the first reading, limit 30 "
                         "seconds")
check(f"...past the limit with still no reading it is red, as before ({_r1})",
      not _r1["ok"] and not _r1["warn"]
      and _r1["text"].startswith("Meter: no reading for 31 seconds"))
check(f"...and a reading makes it green ({_r2['text']!r})",
      _r2["ok"] and _r2["text"].startswith("Meter: 500 W"))

_c.auto = True
_c.p1_enabled = False
asyncio.run(_c.cycle())
_w0 = _c.why_words
_c._why_waiting_since -= _td(seconds=7)
_w7 = _c.why_words
check(f"'A reading is missing for N seconds' counts while control waits, "
      f"instead of the age the last cycle saw ({_w0!r} -> {_w7!r})",
      _w0 == "A reading is missing for 0 seconds, control waits"
      and _w7 == "A reading is missing for 7 seconds, control waits")
_c.why_words = "Keeping the grid near zero: 0 W"
check("...and once a cycle decides again, the line is that decision",
      _c.why_words == "Keeping the grid near zero: 0 W")

# UX-11 item 3: past the hold, the Why line says which reading is missing,
# not "A reading is missing: 0 W". Driven through cycle() for real.
_c.inputs_bad_since = _dt.now(_tz.utc) - _td(seconds=20)
asyncio.run(_c.cycle())
check(f"past the hold, the Why line names the missing reading "
      f"({_c.why_words!r})",
      _c.why_words == "No usable meter reading for over 15 seconds: 0 W")

# UX-11 item 2: the red meter row says the log shows why. A meter that goes
# quiet logs one WARNING per episode, not one per cycle, and INFO when back.
_c.p1_enabled = True
_c.p1 = _p1.P1Ingest(max_age_s=30.0)


def _meter_at(age_s):
    import time as _time
    _c.p1.submit(_p1.make_sample(_p1.SOURCE_HA, 500.0, 0.0, phases=1,
                                 ingest_mono=_time.monotonic() - age_s))


def _quiet_episodes():
    # Two quiet episodes close together, then a third 10 minutes on.
    for age in (1, 40, 45, 50, 1, 2, 60, 65, 1):
        _meter_at(age)
        asyncio.run(_c.read_inputs())
    _c._meter_quiet_warned -= _main.METER_QUIET_REPEAT_S
    for age in (40, 45, 1):
        _meter_at(age)
        asyncio.run(_c.read_inputs())


_mq = [r for r in _captured("goodwe", _quiet_episodes)
       if "meter" in r.getMessage().lower()]
_mq_warn = [r.getMessage() for r in _mq if r.levelno == logging.WARNING]
_mq_info = [r.getMessage() for r in _mq if r.levelno == logging.INFO]
_mq_dbg = [r.getMessage() for r in _mq if r.levelno == logging.DEBUG]
check(f"a meter that goes quiet logs one WARNING per episode, not per cycle, "
      f"and one INFO when readings return, and a flapping meter at most one "
      f"pair per {_main.METER_QUIET_REPEAT_S:.0f} s ({_mq_warn}, {_mq_info}, "
      f"{_mq_dbg})",
      len(_mq_warn) == 2 and _mq_info == ["Meter readings are back."] * 2
      and len(_mq_dbg) == 2
      and _mq_warn[0] == "No meter reading for over 30 seconds, so control "
                         "waits and then sets the battery to 0 W."
      and len(words(_mq_warn[0])) <= LINE_WORDS)

# UX-11 item 1: a state file saved before UX-11 holds the long reasons, and
# they load unchanged. The ERROR line at start shows the code, and the saved
# text goes on the DEBUG line beside it. Old and new files look the same,
# because the saved reason's format did not change and it opens with its code.
_OLD = {
    "sign": "SIGN_DISAGREEMENT: commanded -1000 W against a meter reading "
            "+1500 W for 180 s continuously (limits: both over 300 W, held "
            "over 180 s)",
    "delivery": "UNIT_NOT_DELIVERING: inverter-a is standing at +2000 W and "
                "delivering +0 W - 2000 W of its share (100 % of it) is not "
                "arriving - while inverter-b is not in that state. Nothing "
                "this add-on reads explains it - held for 121 s continuously "
                "(limit: over 120 s)",
    "circ": "CIRCULATING_POWER: 5800 W measured going round between the "
            "inverters for 61 s, peaking at 6000 W, over the limits of 250 W "
            "and 60 s. The meter cannot see it, because it nets to zero there",
}


class _OldStore(_S):
    data = {"unit/inverter-a/sign_tripped": True,
            "unit/inverter-a/sign_reason": _OLD["sign"],
            "unit/inverter-a/delivery_tripped": True,
            "unit/inverter-a/delivery_reason": _OLD["delivery"],
            "circ_tripped": True, "circ_reason": _OLD["circ"],
            "site_sign_tripped": True, "site_sign_reason": ""}


_restart = []
_rs = _captured("goodwe", lambda: _restart.append(Controller(
    {"setpoint_entity": "number.sp", "meter_source": "off", "max_w": 2000},
    _H(), _OldStore(), _P(), stop=asyncio.Event())))
_rs_err = [r.getMessage() for r in _rs if r.levelno == logging.ERROR]
_rs_dbg = [r.getMessage() for r in _rs if r.levelno == logging.DEBUG]
for _m in _rs_err:
    print(f"       {_m!r} ({len(words(_m))} words)")
check("an old state file's long reasons load unchanged, so the stops stay on",
      _restart and _restart[0].inverters[0].sign_wd.tripped
      and _restart[0].inverters[0].sign_wd.reason == _OLD["sign"]
      and _restart[0].inverters[0].delivery_wd.reason == _OLD["delivery"]
      and _restart[0].circ_wd.reason == _OLD["circ"])
check("...each ERROR line at start is one plain sentence with the code, at "
      f"most {LINE_WORDS} words, and a stop saved without a reason says so",
      len(_rs_err) == 4
      and all(len(words(m)) <= LINE_WORDS and "(limits" not in m
              for m in _rs_err)
      and any(m.endswith(": SIGN_DISAGREEMENT.") for m in _rs_err)
      and any(m.endswith(": UNIT_NOT_DELIVERING.") for m in _rs_err)
      and any(m.endswith(": CIRCULATING_POWER.") for m in _rs_err)
      and any(m.endswith(f": {messages.NO_REASON_SAVED}.") for m in _rs_err))
check("...and the saved text, figures and all, is on a DEBUG line beside it",
      all(any(t in m for m in _rs_dbg) for t in _OLD.values()))
# UX-11 review: `log_level` defaults to info and `events` starts empty, so the
# saved figures also go on the action log, where `status` shows them.
_rs_ev = _restart[0].events if _restart else []
check(f"...and in the action log, which an installer reads at the default "
      f"log level ({len(_rs_ev)} events)",
      all(any(t in e for e in _rs_ev) for t in _OLD.values())
      and any(e.endswith(messages.NO_REASON_SAVED) for e in _rs_ev))

# UX-11 item 4: the action log's clamp count and the shadow-off entry.
_main_src = (APP / "main.py").read_text(encoding="utf-8")
check("the action log counts 'engagements' with a real plural and the "
      "shadow-off entry is plain, not 'engagement(s)' or 'SHADOW MODE OFF -'",
      "engagement(s)" not in _main_src and "SHADOW MODE OFF" not in _main_src
      and messages.count(25, "engagement") == "25 engagements"
      and messages.count(1, "engagement") == "1 engagement")

# ============================================================================== #
print("\n8. the Home screen's sentences (UX-13 P1)")
# app/homeview.py builds every sentence the Home screen shows. Each is rendered
# with real values by `text_samples()` and held to STYLE.md: a title at most 5
# words, a plan line at most 2 sentences of at most 12 words each (the design
# note's new row type), any other line 1 sentence of at most 25 words.
from app import homeview                                          # noqa: E402

PLAN_SENTENCE_WORDS, PLAN_SENTENCES = 12, 2


def home_problems(kind, text):
    if kind == "title":
        return title_problems(text)
    out = text_problems(text, keys=False, entities=False)
    if ";" in text:
        out.append("semicolon")
    if kind == "line":
        if sentences(text) > 1:
            out.append(f"over 1 sentence ({sentences(text)})")
        if len(words(text)) > 25:
            out.append(f"over 25 words ({len(words(text))})")
        return out
    if sentences(text) > PLAN_SENTENCES:
        out.append(f"plan line over {PLAN_SENTENCES} sentences")
    long = [x for x in re.split(r"(?<=[.!?])\s+", text)
            if len(words(x)) > PLAN_SENTENCE_WORDS]
    if long:
        out.append(f"plan sentence over {PLAN_SENTENCE_WORDS} words "
                   f"({len(words(long[0]))}): {long[0]!r}")
    return out


for _lang in ("en", "nl"):         # UX-13 P5: Home in English and Dutch
    _home = homeview.text_samples(_lang)
    _home_bad = [(k, t, p) for k, t in _home for p in [home_problems(k, t)]
                 if p]
    for k, t, p in _home_bad:
        print(f"    {_lang} {k}: {t!r}: {p}")
    check(f"every Home sentence in {_lang} obeys its STYLE limit "
          f"({len(_home)} rendered, {len(_home_bad)} over)",
          len(_home) > 80 and not _home_bad)
_shadowed = [t for k, t in homeview.text_samples() if "ould" in t]
check("in shadow mode the Home plan says what the add-on WOULD do: every "
      "shadow variant of a plan text carries 'would'",
      all("ould" in f(True) for f in (
          lambda sh: homeview.run_grid_what(2.5, sh),
          lambda sh: homeview.run_give_what(2.5, sh),
          lambda sh: homeview.run_cycle_what("idle", sh),
          lambda sh: homeview.run_cycle_why(sh),
          lambda sh: homeview.run_peak_what(3500, sh),
          lambda sh: homeview.run_house_what(sh),
          lambda sh: homeview.run_house_why(sh),
          lambda sh: homeview.plan_head("house", sh),
          lambda sh: homeview.plan_head("trade", sh),
          lambda sh: homeview.line_peak_limit(3500, sh)))
      and not any("ould" in f(False) for f in (
          lambda sh: homeview.run_grid_what(2.5, sh),
          lambda sh: homeview.run_house_what(sh),
          lambda sh: homeview.plan_head("house", sh)))
      and len(_shadowed) > 10)
check("mutation: a 13-word plan sentence and a 6-word title are caught",
      home_problems("plan", "One two three four five six seven eight nine "
                            "ten eleven twelve thirteen.")
      and home_problems("title", "One two three four five six"))

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