goodwe-addon/goodwe_controller/test_docs.py

"""Runnable check that the operator docs say what the code does.  `python3 test_docs.py`

DOC-1. Two things the docs got wrong and an installer would have copied:

1. The `inverters:` example in DOCS.md must be ACCEPTED by the shipped
   validator (`app/validate.py`) and carry every key config.yaml's schema
   requires per entry. It omitted `bms_soc_floor` and would not start.
2. The P1 age entity the ESP32 watchdog subscribes to must be documented under
   the id Home Assistant composes from what `app/mqtt.py` announces, and the
   old wrong id must appear in no shipped doc.

Each part also runs its own mutation (habit 1 of MUTATION.md): the spoilt input
must turn it red, or the green above it means nothing.
"""

import os
import re
import sys

from app import mqtt
from app.validate import validate_startup_config

HERE = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.dirname(HERE)

fails = []


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


def read(*parts):
    with open(os.path.join(*parts), encoding="utf-8") as fh:
        return fh.read()


DOCS = read(HERE, "DOCS.md")
CFG = read(HERE, "config.yaml")


# --------------------------------------------------------------------------- #
# A reader for the two YAML shapes used here. No PyYAML in the checks image
# (test_defaults_pin.py says why), so this is by hand, and part 1a checks it
# actually found something.
def scalar(raw):
    raw = raw.split("#", 1)[0].strip()
    if raw[:1] in "\"'" and raw[-1:] == raw[:1] and len(raw) >= 2:
        return raw[1:-1]
    if raw in ("true", "false"):
        return raw == "true"
    if raw == "[]":
        return []
    for cast in (int, float):
        try:
            return cast(raw)
        except ValueError:
            pass
    return raw


def parse_example(text):
    """Top-level `key: value` lines plus one list of flat mappings."""
    out, cur_list, cur = {}, None, None
    for line in text.splitlines():
        if not line.split("#", 1)[0].strip():
            continue
        m = re.match(r"^([a-z0-9_]+):\s*(.*)$", line)
        if m:
            if m.group(2).split("#", 1)[0].strip():
                out[m.group(1)] = scalar(m.group(2))
                cur_list = None
            else:
                cur_list = out[m.group(1)] = []
            continue
        m = re.match(r"^\s+-\s+([a-z0-9_]+):\s*(.*)$", line)
        if m and cur_list is not None:
            cur = {m.group(1): scalar(m.group(2))}
            cur_list.append(cur)
            continue
        m = re.match(r"^\s+([a-z0-9_]+):\s*(.*)$", line)
        if m and cur is not None:
            cur[m.group(1)] = scalar(m.group(2))
            continue
        raise ValueError(f"unparsed line in the DOCS.md example: {line!r}")
    return out


# --------------------------------------------------------------------------- #
print("DOC-1 - the docs say what the code does\n")
print("1. the `inverters:` example in DOCS.md starts")

section = DOCS[DOCS.index("### More than one inverter (`inverters`)"):]
section = section[:section.index("\n### ", 1)]
block = re.search(r"```yaml\n(.*?)```", section, re.S).group(1)
example = parse_example(block)
entries = example.get("inverters") or []

# The options a fresh install is given, from config.yaml itself: the example is
# pasted on top of these, exactly as an installer does.
opts_block = CFG[CFG.index("\noptions:"):CFG.index("\nschema:")]
shipped = {m.group(1): scalar(m.group(2))
           for m in re.finditer(r"^  ([a-z0-9_]+): *(.*)$", opts_block, re.M)}

# config.yaml's per-entry schema: required keys are the ones without `?`.
schema_block = CFG[CFG.index("\nschema:"):]
inv_schema = re.search(r"^  inverters:\n((?:    .*\n)+)", schema_block, re.M).group(1)
schema_keys = dict(re.findall(r"^    -? ?\s*([a-z0-9_]+): *(\S+)", inv_schema, re.M))
required = sorted(k for k, t in schema_keys.items() if not t.endswith("?"))

check(f"1a. the reader found the example: {len(entries)} entries, "
      f"{len(shipped)} shipped options, schema keys {sorted(schema_keys)}",
      len(entries) == 2 and all(len(e) >= 4 for e in entries)
      and len(shipped) > 50 and "bms_soc_floor" in required)

missing = [(e.get("name"), k) for e in entries for k in required if k not in e]
check(f"1b. every entry carries every key the schema requires {required} "
      f"(missing: {missing})", not missing)

unknown = [(e.get("name"), k) for e in entries for k in e if k not in schema_keys]
check(f"1c. no entry uses a key the schema does not declare (found: {unknown})",
      not unknown)


def with_example(ex):
    opts = dict(shipped)
    opts.update(ex)
    return opts


problems = validate_startup_config(with_example(example))
for p in problems:
    print(f"      refused: {p[:160]}")
check("1d. the shipped validator accepts the example as documented", problems == [])

# Precondition: the example clears the flat setpoint_entity, so without its
# roster there is no actuator. If this passed, 1d graded the flat keys, not the
# roster.
flat_only = {k: v for k, v in example.items() if k != "inverters"}
check("1e. ...and it is the ROSTER that 1d graded (the example without it is "
      "refused)", validate_startup_config(with_example(flat_only)) != [])

# The mutation this card is about: the key the docs used to leave out.
spoilt = dict(example, inverters=[
    {k: v for k, v in e.items() if k != "bms_soc_floor"} for e in entries])
sp = validate_startup_config(with_example(spoilt))
check("1f. MUTATION: the example with `bms_soc_floor` removed is refused, "
      "and the refusal names it",
      sp != [] and any("bms_soc_floor" in p for p in sp))

# --------------------------------------------------------------------------- #
print("\n2. the P1 age entity is documented under its real id")


def compose(device_name, entity_name):
    """HA's entity_id from device name + entity name (mqtt.py's note on
    DEVICE). Written here, not borrowed from mqtt._slug, so a change to that
    helper cannot move both sides of the comparison."""
    s = re.sub(r"[^a-z0-9]+", "_", f"{device_name} {entity_name}".lower())
    return "sensor." + s.strip("_")


p1_rows = [r for r in mqtt.SENSORS if r[0] == "p1_age"]
check("2a. mqtt.py announces exactly one `p1_age` sensor", len(p1_rows) == 1)
real_id = compose(mqtt.DEVICE["name"], p1_rows[0][2]) if p1_rows else ""
print(f"      composed from mqtt.py: {real_id}")

# Anchor outside mqtt.py: the id recorded from the live install.
recorded = read(HERE, "tools", "rehearsal", "announced-reference.txt")
check("2b. ...and that composes to the id recorded from the live install "
      "(tools/rehearsal/announced-reference.txt)",
      re.search(rf"^{re.escape(real_id)}\s+goodwe_ctl_p1_age\b", recorded, re.M)
      is not None)

m = re.search(r"^#### `(sensor\.[a-z0-9_]+)`\n\n(?:.*\n){0,6}?.*stale-input watchdog",
              DOCS, re.M)
documented = m.group(1) if m else None
check(f"2c. DOCS.md's watchdog-entity section is headed with that id "
      f"(found {documented!r})", documented == real_id)

# The wrong id, anywhere an installer reads. CHANGELOG.md and DECISIONS/ are
# history and keep what was said at the time.
SHIPPED_DOCS = [
    os.path.join(HERE, "DOCS.md"), os.path.join(HERE, "config.yaml"),
    os.path.join(REPO, "README.md"), os.path.join(REPO, "FIELD-GUIDE.md"),
    os.path.join(REPO, "docs", "build_guide.py"),
    os.path.join(REPO, "docs", "field-guide.html"),
]
WRONG = "p1_sample_age_s"
hits = []
for path in SHIPPED_DOCS:
    for n, line in enumerate(read(path).splitlines(), 1):
        if WRONG in line:
            hits.append(f"{os.path.relpath(path, REPO)}:{n}")
check(f"2d. `{WRONG}` appears in no shipped doc (hits: {hits})", not hits)

# MUTATION: the old heading must fail 2c's comparison.
bad = DOCS.replace(f"#### `{real_id}`", f"#### `sensor.{WRONG}`")
mb = re.search(r"^#### `(sensor\.[a-z0-9_]+)`\n\n(?:.*\n){0,6}?.*stale-input watchdog",
               bad, re.M)
check("2e. MUTATION: the old heading is found and does not match",
      mb is not None and mb.group(1) != real_id)

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