harbour-aira/bridge/test_bridge_commands.py

#!/usr/bin/env python3
"""Verifica offline dei comandi del bridge. NON invia nulla all'impianto:
costruisce i messaggi protobuf e ne controlla il contenuto.

    .venv/bin/python test_bridge_commands.py
"""
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))
import aira_bridge as b   # noqa: E402

esiti = []


def check(nome, cond, extra=""):
    esiti.append((nome, bool(cond), extra))
    print(f"{'OK  ' if cond else 'FALLITO'} {nome}{(' -> ' + extra) if extra else ''}")


# --- contesto: come lo stato reale dell'impianto --------------------------- #
ctx = {"zone_setpoints_heating": {"zone_1": 19.0, "zone_2": 0.0},
       "zone_setpoints_cooling": {"zone_1": 22.0, "zone_2": 0.0}}

# --- 1) riscaldamento, zona 1 --------------------------------------------- #
cmd = b.build_command("set_heating_setpoint", 21.5, 1, ctx)
msg = cmd._message
check("zona 1: obiettivo applicato", abs(msg.zone_setpoints.zone_1 - 21.5) < 1e-6,
      f"zone_1={msg.zone_setpoints.zone_1}")
check("zona 1: zona 2 preservata", msg.zone_setpoints.zone_2 == 0.0,
      f"zone_2={msg.zone_setpoints.zone_2}")
check("zona 1: kind = riscaldamento (1)", msg.kind == 1, f"kind={msg.kind}")
check("messaggio serializzabile", len(msg.SerializeToString()) > 0,
      f"{len(msg.SerializeToString())} byte")

# --- 2) raffrescamento, zona 2 (con zona 1 da preservare) ------------------ #
ctx2 = {"zone_setpoints_cooling": {"zone_1": 22.0, "zone_2": 24.0},
        "zone_setpoints_heating": {"zone_1": 19.0, "zone_2": 0.0}}
cmd = b.build_command("set_cooling_setpoint", 23.0, 2, ctx2)
msg = cmd._message
check("zona 2: obiettivo applicato", abs(msg.zone_setpoints.zone_2 - 23.0) < 1e-6,
      f"zone_2={msg.zone_setpoints.zone_2}")
check("zona 2: zona 1 preservata (22.0)", msg.zone_setpoints.zone_1 == 22.0,
      f"zone_1={msg.zone_setpoints.zone_1}")
check("raffrescamento: kind = 2", msg.kind == 2, f"kind={msg.kind}")

# --- 3) senza contesto: non deve rompersi -------------------------------- #
cmd = b.build_command("set_heating_setpoint", 20.0)
check("senza contesto: nessuna eccezione", cmd._message.zone_setpoints.zone_1 == 20.0)

# --- 4) valori rifiutati -------------------------------------------------- #
def rifiuta(nome, *args, **kw):
    try:
        b.build_command(*args, **kw)
    except ValueError as exc:
        check(nome, True, str(exc))
    except Exception as exc:                       # inatteso
        check(nome, False, f"eccezione sbagliata: {type(exc).__name__}: {exc}")
    else:
        check(nome, False, "accettato, doveva essere rifiutato")


rifiuta("rifiuta 40 °C (troppo alta)", "set_heating_setpoint", 40.0, 1, ctx)
rifiuta("rifiuta 3 °C (troppo bassa)", "set_heating_setpoint", 3.0, 1, ctx)
rifiuta("rifiuta zona 3", "set_heating_setpoint", 20.0, 3, ctx)
rifiuta("rifiuta valore non numerico", "set_heating_setpoint", "caldo", 1, ctx)
rifiuta("rifiuta comando sconosciuto", "factory_reset", None, 1, ctx)

# --- 5) il riassunto espone i nuovi campi --------------------------------- #
stato_file = Path("/tmp/state.json")
if stato_file.exists():
    st = json.loads(stato_file.read_text())["state"]
    s = b.summarize(st)
    for k in ("zone1_setpoint", "zone1_temperature", "zone1_mode",
              "zone2_setpoint", "zones"):
        check(f"riassunto contiene {k}", k in s, repr(s.get(k)))
    check("zona 1: obiettivo presente", s.get("zone1_setpoint") is not None,
          f"{s.get('zone1_setpoint')} °C")

    # force_heating e' un oggetto {enabled, remaining_time}: leggerlo con bool()
    # darebbe sempre vero (era il bug). Qui si verifica la lettura del campo.
    spento = b.summarize(dict(st, force_heating={"enabled": False, "remaining_time": "0s"}))
    acceso = b.summarize(dict(st, force_heating={"enabled": True, "remaining_time": "1h"}))
    check("riassunto: forzato spento -> falso", spento.get("force_heating") is False,
          repr(spento.get("force_heating")))
    check("riassunto: forzato acceso -> vero", acceso.get("force_heating") is True,
          repr(acceso.get("force_heating")))
    check("riassunto: durata forzato riportata", acceso.get("force_heating_remaining") == "1h",
          repr(acceso.get("force_heating_remaining")))
    for k in ("hot_water_heating", "configured_modes", "pump_active_state",
              "outdoor_defrost", "signature_lights"):
        check(f"riassunto contiene {k}", k in s, repr(s.get(k)))
else:
    print("(salto il controllo del riassunto: /tmp/state.json assente)")

# --- 6) statistiche: conversioni pure ------------------------------------- #
check("money_value: 2710000000 / 10^9 = 2.71", b.money_value(2710000000, 9) == 2.71,
      repr(b.money_value(2710000000, 9)))
check("money_value: valori assenti -> 0.0", b.money_value(None, None) == 0.0)
check("wh_to_kwh: 29300 Wh = 29.3 kWh", b.wh_to_kwh(29300) == 29.3, repr(b.wh_to_kwh(29300)))
check("wh_to_kwh: 70844.5 Wh = 70.8 kWh", b.wh_to_kwh(70844.53878839477) == 70.8,
      repr(b.wh_to_kwh(70844.53878839477)))
check("wh_to_kwh: zero -> None (nessun dato)", b.wh_to_kwh(0) is None)
check("wh_to_kwh: valore assente -> None", b.wh_to_kwh(None) is None)

class _YM:            # finto YearMonth / LocalDateTime
    def __init__(self, y, m, d=0):
        self.year, self.month, self.day = y, m, d

check("_ym: 2026-09", b._ym(_YM(2026, 9)) == "2026-09", b._ym(_YM(2026, 9)))
check("_data: 2026-10-01", b._data(_YM(2026, 10, 1)) == "2026-10-01", b._data(_YM(2026, 10, 1)))

# --- 7) riassunto: blocco dispositivo (stato sintetico, nessun file) ------- #
# I valori del termostato arrivano con i decimali come interi (228 -> 22.8).
finto = {
    "heat_pump_id": {"value": "1f4b26c4-1800-4b21-8ea2-236925a92aeb"},
    "versions": {"linux_build_id": "6.10.0", "connectivity_manager": "3.30.1",
                 "climate_control_villa": "3.9.12", "outdoor_unit_application": "1.30.0",
                 "outdoor_unit_eeprom": "0.114.0"},
    "thermostats": [{"article_number": "201884", "zone": "ZONE_1", "rssi": -79,
                     "last_update": {"actual_temperature": 228, "humidity": 734,
                                     "warning_low_battery_level": False}}],
    "time_zone": {"time_zone_identifier": "Europe/Rome"},
    "zone_setpoints_heating": {"zone_1": 19.0, "zone_2": 0.0},
    "zone_temperatures": {"zone_1": 22.8, "zone_2": 0.0},
}
d = b.summarize(finto)
check("dispositivo: id breve di 6 cifre", d.get("device_id") == "1f4b26", repr(d.get("device_id")))
check("dispositivo: uuid completo", d.get("device_uuid") == "1f4b26c4-1800-4b21-8ea2-236925a92aeb")
check("dispositivo: versione piattaforma", d.get("version_platform") == "6.10.0")
check("dispositivo: versione connettività", d.get("version_connectivity") == "3.30.1")
check("dispositivo: versione unità esterna", d.get("version_outdoor_unit") == "1.30.0")
check("dispositivo: connessione", d.get("connected_via") == "Cloud")
check("dispositivo: fuso orario", d.get("timezone") == "Europe/Rome")
check("termostato: zona 1 da 'ZONE_1'", d.get("thermostat_zone") == 1, repr(d.get("thermostat_zone")))
check("termostato: 228 -> 22.8 °C", d.get("thermostat_temperature") == 22.8,
      repr(d.get("thermostat_temperature")))
check("termostato: 734 -> 73.4 %", d.get("thermostat_humidity") == 73.4,
      repr(d.get("thermostat_humidity")))
check("termostato: segnale riportato", d.get("thermostat_rssi") == -79)
check("termostato: batteria non scarica", d.get("thermostat_battery_low") is False)
check("senza termostato: nessun errore",
      b.summarize({k: v for k, v in finto.items() if k != "thermostats"}).get("thermostat_zone") is None)

# --- esito ---------------------------------------------------------------- #
falliti = [n for n, ok, _ in esiti if not ok]
print(f"\n{len(esiti) - len(falliti)}/{len(esiti)} controlli superati")
sys.exit(1 if falliti else 0)