goodwe-addon/goodwe_controller/test_validate.py

"""Runnable check for the startup gate.  `python3 test_validate.py`

Two halves, and the second is the one that matters. The first asserts that the
three refusable configurations are refused. The second asserts that everything
on validate.py's "deliberately not refused" list still STARTS - because the way
a validator dies is not by missing a fault, it is by refusing something harmless
until somebody deletes the call.
"""

import asyncio
import sys

from app.validate import (
    ACTUATOR_WRITE_KEYS, FIRMWARE_INGEST_VETO_S, ConfigRefused, refuse,
    validate_startup_config,
)

fails = []


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


def base(**over):
    """A configuration that must start. Everything else is this, spoilt."""
    o = {"setpoint_entity": "number.goodwe_setpoint", "max_w": 2000,
         "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
         "meter_source": "off"}
    o.update(over)
    return o


def why(problems, needle):
    return any(needle in p for p in problems)


# --------------------------------------------------------------------------- #
print("the baseline must start, or nothing below means anything")
check("a good config with a 5000 W device is accepted",
      validate_startup_config(base(), -5000, 5000) == [])
check("...and with no device reachable yet, too",
      validate_startup_config(base(), None, None) == [])

# --------------------------------------------------------------------------- #
print("1. max_w against the device's own rating")
# ⚠️ THE ORIGINAL BUG. main.py used to log 'the device wins' and run anyway,
# with write_setpoint() clamping, so the configured ceiling was never the real
# one and the only evidence had scrolled off the log.
p = validate_startup_config(base(max_w=8000), -5000, 5000)
check("max_w above the device maximum is refused", len(p) == 1)
check("...and the message names the configured value", why(p, "8000"))
check("...and the device's value", why(p, "5000"))
check("...and says which one wins", why(p, "THE DEVICE WINS"))
check("max_w exactly at the device maximum is fine",
      validate_startup_config(base(max_w=5000), -5000, 5000) == [])
check("max_w under it is fine",
      validate_startup_config(base(max_w=2000), -5000, 5000) == [])
# The charge side is a separate rail and an asymmetric device is legal.
p = validate_startup_config(base(max_w=4000), -2500, 5000)
check("a max_w the device cannot take on the CHARGE side is refused too",
      len(p) == 1 and why(p, "charge side"))
check("...naming the device's floor", why(p, "-2500"))
# ⚠️ Not guessed at: the entity may simply not have booted yet. run_control()
# re-runs this the moment the real rating arrives.
check("an unreachable device skips the range check rather than assuming",
      validate_startup_config(base(max_w=8000), None, None) == [])
check("a non-numeric max_w is refused",
      why(validate_startup_config(base(max_w="lots"), -5000, 5000), "not a number"))

# --------------------------------------------------------------------------- #
print("2. an actuator to write to")
# A controller with no actuator: write_setpoint() returns on the blank entity,
# so no rating is read, no heartbeat is sent, layer 1 pins the inverter at 0 W,
# and the add-on looks perfectly healthy the whole time.
for blank in ("", "   ", None):
    p = validate_startup_config(base(setpoint_entity=blank), None, None)
    check(f"setpoint_entity {blank!r} is refused", len(p) == 1)
check("...and the message says why it is dangerous rather than untidy",
      why(validate_startup_config(base(setpoint_entity=""), None, None),
          "heartbeat"))
# specs §1.5: two writers on one actuator is not expressible in this schema.
check("there is exactly one way to reach the actuator",
      ACTUATOR_WRITE_KEYS == ("setpoint_entity",))

# --------------------------------------------------------------------------- #
print("3. a meter that can never read")
# ⚠️ These predicates belong to p1.validate_meter_options(), which build_source()
# also uses. Asserted here to prove the gate actually consults it - one entry
# point, not two.
check("meter_source off asks nothing of the P1 keys",
      validate_startup_config(base(meter_source="off"), -5000, 5000) == [])
check("homewizard_local with no p1_host is refused",
      why(validate_startup_config(base(meter_source="homewizard_local"),
                                  -5000, 5000), "p1_host"))
check("...and with a host it is accepted",
      validate_startup_config(base(meter_source="homewizard_local",
                                   p1_host="192.168.2.250"), -5000, 5000) == [])
check("a poll cadence at meter_max_age_s is refused at STARTUP, not just disabled",
      why(validate_startup_config(base(meter_source="homewizard_local",
                                       p1_host="m", meter_poll_s=30,
                                       meter_max_age_s=30), -5000, 5000),
          "meter_max_age_s"))
check("ha_signed with no p1_net_entity is refused",
      why(validate_startup_config(base(meter_source="ha_signed"), -5000, 5000),
          "p1_net_entity"))
check("a phase list disagreeing with meter_phases is refused",
      why(validate_startup_config(base(meter_source="ha_signed",
                                       p1_net_entity="sensor.n", meter_phases=3,
                                       p1_phase_net_entities=["a", "b"]),
                                  -5000, 5000), "phase-count"))
check("mqtt_p1 with no topic is refused",
      why(validate_startup_config(base(meter_source="mqtt_p1"), -5000, 5000),
          "meter_mqtt_topic"))
check("ha_dsmr with only half its entities is refused",
      why(validate_startup_config(base(meter_source="ha_dsmr",
                                       p1_import_entity="sensor.i"), -5000, 5000),
          "p1_export_entity"))
check("a misspelt meter_source is refused rather than silently ignored",
      why(validate_startup_config(base(meter_source="ha_signd"), -5000, 5000),
          "is not one of"))

# --------------------------------------------------------------------------- #
print("4. meter_max_age_s against the firmware's compile-time veto")
# ⚠️ The schema allows int(5,300); the firmware's threshold is baked in at 30 s
# and wins silently. Same shape as the max_w case: a configured bound overridden
# by a lower layer, with nothing saying so.
hw = dict(meter_source="homewizard_local", p1_host="m")
p = validate_startup_config(base(**hw, meter_max_age_s=120), -5000, 5000)
check("a meter_max_age_s above the firmware veto is refused", len(p) == 1)
check("...naming the configured value", why(p, "120"))
check("...naming the firmware's value", why(p, "30"))
check("...and saying which one wins", why(p, "THE FIRMWARE WINS"))
check("the tested pair 30/30 is accepted",
      validate_startup_config(base(**hw, meter_max_age_s=30), -5000, 5000) == [])
check("one second above it is refused",
      validate_startup_config(base(**hw, meter_max_age_s=31), -5000, 5000) != [])
# Below the veto the add-on gives up FIRST, which is fail-safe and visible on
# its own status page. Only above is the silent case.
check("below the veto is fine - the add-on being stricter is not a conflict",
      validate_startup_config(base(**hw, meter_max_age_s=10,
                                   meter_poll_s=4), -5000, 5000) == [])
# ⚠️ Scope: meter_max_age_s is read by nothing but P1Ingest, and the veto keys
# off a sensor only published when ingestion runs. With the meter off it governs
# nothing, so refusing would be pure trivia.
check("with meter_source off the number governs nothing and is not refused",
      validate_startup_config(base(meter_source="off", meter_max_age_s=300),
                              -5000, 5000) == [])
check("the constant is the firmware's, not a rounded guess",
      FIRMWARE_INGEST_VETO_S == 30.0)

# --------------------------------------------------------------------------- #
print("max_w must be a positive magnitude")
# Schema-blocked on a real install, but validate_startup_config() is also
# sim/inject.py's entry point and can be handed any dict at all.
for bad in (0, -1, -2000):
    p = validate_startup_config(base(max_w=bad), -5000, 5000)
    check(f"max_w {bad} is refused", len(p) == 1 and why(p, "must be positive"))
check("...and the message explains it would read as a trip, not a mistake",
      why(validate_startup_config(base(max_w=0), -5000, 5000), "contradiction"))
check("the smallest sane max_w still starts",
      validate_startup_config(base(max_w=1), -5000, 5000) == [])

# --------------------------------------------------------------------------- #
print("what must NOT be refused - a validator that fires on trivia gets deleted")
# DOCS.md: the arbiter's `supervised` limit still binds the wire to +-max_w, so
# the only cost of the extra headroom is unwind latency after saturation.
check("integrator_max_w above max_w warns but starts",
      validate_startup_config(base(max_w=2000, integrator_max_w=9000),
                              -5000, 5000) == [])
check("integrator_max_w 0 (follow max_w) starts",
      validate_startup_config(base(integrator_max_w=0), -5000, 5000) == [])
# Fail-safe AND already a red NOT READY banner naming the entity: a refusal on
# top of that adds nothing an operator can act on.
check("a missing soc_entity is left to the readiness banner",
      validate_startup_config(base(soc_entity=""), -5000, 5000) == [])
check("a missing batt_entity likewise",
      validate_startup_config(base(batt_entity=""), -5000, 5000) == [])
# The transient case. ESPHome is routinely not up when the add-on boots, and
# refusing here would turn a device that arrives late into a restart loop.
check("a setpoint entity that is configured but not yet found starts",
      validate_startup_config(base(), None, None) == [])
check("meter_poll_s with thin but non-zero headroom starts",
      validate_startup_config(base(meter_source="homewizard_local", p1_host="m",
                                   meter_poll_s=20, meter_max_age_s=30),
                              -5000, 5000) == [])

# --------------------------------------------------------------------------- #
print("the refusal must be legible as a refusal, not as a crash")
p = validate_startup_config(base(max_w=8000, setpoint_entity=""), -5000, 5000)
check("several problems are all reported, not just the first", len(p) >= 1)
refuse(p)   # prints the banner; the eye is the assertion for its wording
err = ConfigRefused(p)
check("ConfigRefused carries every problem", err.problems == p)
check("...and str() names one of them", str(err) == p[0])

# --------------------------------------------------------------------------- #
print("the LATE refusal, when the device only shows up after the gate has run")
# ⚠️ This path had no test at all and no live demonstration - the review's
# catch. ESPHome is routinely not up when the add-on boots, so hass.limits()
# returns (None, None), the range checks are skipped, and run_control() is the
# first place the device's real rating is ever known. Everything below is about
# what happens when it turns out to conflict.
from app.main import Controller            # noqa: E402


class _LateHass:
    """Reports the range only run_control() gets to see."""

    def __init__(self, lo, hi):
        self.lo, self.hi = lo, hi
        self.writes = []

    async def limits(self, entity):
        return (self.lo, self.hi, 1.0)

    async def number(self, *a, **k):
        return 0.0

    async def set_number(self, entity, value):
        self.writes.append(value)
        return True


class _Store2:
    data = {}

    def set(self, *a):
        pass


class _Pub2:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


def late(opts_over, lo, hi):
    """Drive run_control() against a device that only reports (lo, hi) now.

    ⚠️ Returns whether run_control() EXITED ON ITS OWN. That is the assertion
    that matters: on a refusal it must return without entering the while loop,
    and on a clean config it must not return at all. So the clean case is stopped
    from outside after it has had time to write, rather than waited on - a
    wait_for() here would just time out, which is the correct behaviour dressed
    up as a failure.
    """
    async def go():
        stop = asyncio.Event()
        hass = _LateHass(lo, hi)
        # A device narrower than the shipped 5000 W schema ceiling is exactly
        # the case that cannot be reached through the Supervisor on the live
        # install, so it is reached here instead.
        c = Controller(base(**opts_over), hass, _Store2(), _Pub2(), stop=stop)
        c.auto = True
        task = asyncio.create_task(c.run_control())
        for _ in range(30):                    # up to ~3 s
            if task.done():
                break
            await asyncio.sleep(0.1)
        exited = task.done()
        if not exited:
            c.stopping = True
        await asyncio.wait_for(task, timeout=5)
        return c, stop, hass, exited
    return asyncio.run(go())

c, stop, hass, exited = late({"max_w": 5000}, -3600.0, 3600.0)
check("a device narrower than max_w, discovered late, is refused",
      c.refused is not None and len(c.refused) == 1)
check("...naming both values", why(c.refused, "5000") and why(c.refused, "3600"))
check("...signalling amain to wind down", stop.is_set())
check("...and returning instead of looping forever", exited)
# The whole point: run_control() must not fall through into the while loop. A
# single write here would mean it commanded the inverter under a config it had
# just declared unusable.
check("...having never entered the control loop, so it wrote nothing",
      hass.writes == [])
check("...and it is recorded in the event log an operator reads",
      any("REFUSED" in e for e in c.events))

# The same path when there is no conflict: it must NOT refuse, and it must
# actually regulate - asserted by a write reaching the device, not by absence.
c2, stop2, hass2, exited2 = late({"max_w": 2000}, -5000.0, 5000.0)
check("a late device that agrees does not trip the re-check", c2.refused is None)
check("...and the control loop runs, writing to the device", hass2.writes != [])
check("...and does not signal a shutdown", not stop2.is_set() and not exited2)

# ⚠️ THE FOOTGUN THE REVIEW FOUND. `stop` used to default to None, so this
# construction was legal and the refusal path then returned into a live process
# with no control loop, no exit and no further output.
try:
    Controller(base(), _LateHass(-5000.0, 5000.0), _Store2(), _Pub2())
    check("a Controller with no stop Event is rejected at construction", False)
except TypeError:
    check("a Controller with no stop Event is rejected at construction", True)

# --------------------------------------------------------------------------- #
print()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")