Files
goodwe-addon/goodwe_controller/test_messages.py
T
adminandClaude Opus 5.5 d017fc67bf
checks / unit (pull_request) Successful in 8m47s
infra-pin / fresh (pull_request) Failing after 1s
UX-08 phase 2b: short, plain status rows and page banners
Every Commissioning row and every page banner is now worded in
app/messages.py (row_*, banner_*): one sentence of at most 15 words, a red
row up to three sentences so a latch still says how it clears, banners at
most 12 words. No row names a Home Assistant entity: sensors by their
setting, solar sources by their place in the list. Reason codes and limits
stay on the log line and event list; rows keep the numbers a reader acts on.

UX-07 item 1: ClampWatchdog counts over-rating and integrator-bound
engagements separately, and after an episode each count shows under its own
alarm's title instead of both under "output clamp".

No row colour, visibility or banner order changed. Rows carry an "id" so the
tests pin meaning rather than wording. The row and banner kinds leave the
allowlist (200 -> 168, cap equal). DOCS.md gains "Status page"; choices are in
DECISIONS/UX-08-short-notifications.md.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LyR88rXETvwhTgx6yHE39p
2026-09-27 02:12:10 +02:00

474 lines
20 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))
os.environ["HOSTNAME"] = "abcd1234-goodwe-controller"
check("inside the add-on the Details line links to its documentation page",
"(/hassio/addon/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",
"](" not in SAMPLES[0].data()["message"]
and "under Notifications" in SAMPLES[0].data()["message"])
del os.environ["HOSTNAME"]
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"
def render(node):
"""A string expression as text, placeholders as {expr}; None otherwise."""
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return node.value
if isinstance(node, ast.JoinedStr):
return "".join(v.value if isinstance(v, ast.Constant)
else "{" + ast.unparse(v.value) + "}" for v in node.values)
if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
a, b = render(node.left), render(node.right)
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), render(node.orelse)
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):
"""Top-level string expressions inside `node` (not their parts)."""
t = render(node)
if t is not None:
yield node, t
return
for child in ast.iter_child_nodes(node):
yield from strings_in(child)
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"))
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):
if len(words(t)) > 1:
out.append(("refusal", f"{f.name}:{n.lineno}", t))
tr = TL.parse_translation_file(HERE / "translations" / "en.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"en.yaml:{path}{key}",
v["name"].split("·")[-1].strip()))
if isinstance(v.get("description"), str):
out.append(("option-desc", f"en.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))})")
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 = 168
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]
check(f"every other red row makes a 'Not ready' banner of at most 12 words "
f"(too long: {_long})", not _long)
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)
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("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")