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 sys

from app.validate import (
    ACTUATOR_WRITE_KEYS, 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("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()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")