"""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 RATCHET for the kinds not rewritten yet. Every status-row and banner literal, 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 = 200 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) allow = read_allowlist(ALLOW_FILE) by_kind = {} for kind, *_ in found: 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) # ============================================================================== # 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 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")