adminandClaude Opus 5.5
63dee6db1e
The Supervisor questions for the Details link are asked together with a 3 s budget instead of one after the other at 10 s each, so a hung Supervisor cannot hold control start for 20 s. In shadow mode the circulating and site-direction latches are pinned to keep their own red row that says how they clear. Plainer inverter and battery wording. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
604 lines
27 KiB
Python
604 lines
27 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"
|
|
|
|
|
|
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 = 167
|
|
|
|
|
|
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)
|
|
|
|
# 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)
|
|
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")
|