Claude review
569f16aa91
- status_payload() also returns the action log as `events`. The page does not render it. Some entries (a dropped inverter's reason, the peak guard's projection, unplaced watts) are written nowhere else, and DOCS still points installers at the action log. - test_peakguard pins the engaged row's ordered pair (the battery's share, then the limit). - test_messages expands setting keys named by constants imported from another app module, so the shown-text 30-word scan counts them. - p1.py: 'A meter message could not be read'. DOCS glossary: control board in plain words, with the hardware in an installer note. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
804 lines
36 KiB
Python
804 lines
36 KiB
Python
"""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 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) --------------------------------------
|
|
BANNED = re.compile(
|
|
r"\b(?:(?:SAFETY|TARIFF|TEL|CFG|FCST|OPT|UX|HW|FW|WITNESS|PKG|ARCH)-\d+|"
|
|
r"integrator|anti-?windup|windup|clamp(?:ed|s)?|i_clamped|saturat\w*|"
|
|
r"BUG SIGNAL|latch(?:ed)?|arbiter|allocator|registers?|0x[0-9A-Fa-f]+|"
|
|
r"layer [123]|heartbeat|watchdog|telegrams?|ACKs?|CRC|OTA|Modbus|planner|"
|
|
r"capaciteitstarief|journalctl|ssh|grep|"
|
|
r"\d+\.\d+\.\d+\.\d+|[\w-]+\.(?:py|md|yaml|json|db))\b|§\s?\d", re.I)
|
|
ENTITY = re.compile(r"\b(?:sensor|number|binary_sensor|input_\w+|switch|select)"
|
|
r"\.[a-z0-9_*]+")
|
|
SNAKE = re.compile(r"\b[a-z0-9]+(?:_[a-z0-9]+)+\b")
|
|
SHOUT = re.compile(r"\b[A-Z]{2,}[A-Z0-9]*(?:[\s-]+[A-Z]{2,}[A-Z0-9]*)+\b")
|
|
PLACEHOLDER = re.compile(r"\{[^{}]*\}|%[-+ #0-9.]*[sdrfgx%]")
|
|
|
|
|
|
def words(text):
|
|
return [w for w in re.findall(r"\S+", PLACEHOLDER.sub("X", text))
|
|
if re.search(r"[A-Za-z0-9]", w)]
|
|
|
|
|
|
def sentences(text):
|
|
"""Sentence count: a stop followed by a space and a capital, or the end."""
|
|
t = PLACEHOLDER.sub("X", text.strip())
|
|
t = re.sub(r"\b(e\.g|i\.e|etc)\.", "eg", t)
|
|
return len([s for s in re.split(r"[.!?](?:\s+|$)", t) if s.strip()])
|
|
|
|
|
|
def title_problems(title):
|
|
out = []
|
|
if len(words(title)) > 5:
|
|
out.append(f"title over 5 words ({len(words(title))})")
|
|
if title.lower().startswith("goodwe"):
|
|
out.append("title carries a GoodWe prefix")
|
|
if title.rstrip().endswith("."):
|
|
out.append("title ends in a full stop")
|
|
return out + text_problems(title, keys=False, entities=False)
|
|
|
|
|
|
def text_problems(text, *, keys, entities):
|
|
t = PLACEHOLDER.sub("X", text)
|
|
out = []
|
|
hit = sorted({m.group(0) for m in BANNED.finditer(t)})
|
|
if hit:
|
|
out.append("banned: " + ", ".join(hit[:4]))
|
|
if not entities and ENTITY.search(t):
|
|
out.append("entity id: " + ENTITY.search(t).group(0))
|
|
if not keys and SNAKE.search(ENTITY.sub("", t)):
|
|
out.append("option key: " + SNAKE.search(ENTITY.sub("", t)).group(0))
|
|
if SHOUT.search(t):
|
|
out.append("capitals: " + SHOUT.search(t).group(0))
|
|
return out
|
|
|
|
|
|
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),
|
|
}
|
|
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 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"), "")
|
|
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")
|
|
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")
|
|
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()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED")
|
|
sys.exit(1)
|
|
print("all passed")
|