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)

# --- 5) comandi che la libreria accetta solo con i loro argomenti ---------- #
# Qui sta la ragione del blocco: il boost senza durata solleva TypeError e il
# bridge risponde "richiesta non valida" all'app.
from pyairahome.commands import CommandBase   # noqa: E402

boost = b.build_command("hot_water_boost_on")
check("boost: mezz'ora di default (richiesta dell'utente)",
      boost._message.hot_water_boost_duration.seconds == 1800,
      f"{boost._message.hot_water_boost_duration.seconds}s")
check("boost: durata scelta in minuti (90 -> 5400s)",
      b.build_command("hot_water_boost_on", 90)
      ._message.hot_water_boost_duration.seconds == 5400)
rifiuta("boost: durata fuori intervallo", "hot_water_boost_on", 5)
rifiuta("boost: durata non numerica", "hot_water_boost_on", "presto")

acs = b.build_command("hot_water_heating_off")
check("ACS spento: un'ora di default", acs._message.time.seconds == 3600,
      f"{acs._message.time.seconds}s")

forzato = b.build_command("force_heating_on")
check("forzato: un'ora di default", forzato._message.duration.seconds == 3600,
      f"{forzato._message.duration.seconds}s")

assenza = b.build_command("away_mode_on")
m = assenza._message
check("assenza: da adesso a +7 giorni",
      abs((m.end_time.seconds - m.current_time.seconds) - 7 * 86400) <= 1,
      f"{m.end_time.seconds - m.current_time.seconds}s")
check("assenza: fine dopo l'inizio", m.end_time.seconds > m.current_time.seconds)
check("assenza: casa tenuta a 15 °C", abs(m.target_room_temperature - 15.0) < 1e-6,
      f"{m.target_room_temperature}")
rifiuta("assenza: zero giorni", "away_mode_on", 0)

# Nessun comando della whitelist deve restare senza un argomento obbligatorio:
# e' il difetto che ha fermato il boost, e la whitelist e' esattamente la lista
# che l'app puo' chiamare, quindi va provata tutta.
campione = {"set_hot_water_target": 55, "set_heating_setpoint": 20,
            "set_cooling_setpoint": 20, "hot_water_boost_on": 60,
            "hot_water_heating_off": 60, "away_mode_on": 7,
            "force_heating_on": 60}
for cid in b.COMMANDS:
    try:
        cmd = b.build_command(cid, campione.get(cid), 1, ctx)
        # La busta e' quello che la libreria manda davvero: se il comando non
        # avesse un nome di campo valido, si romperebbe qui. (I comandi senza
        # parametri, come ping, hanno il messaggio interno vuoto: controllare
        # quello sarebbe sbagliato.)
        busta = CommandBase.build_command(cmd)
        byte = len(busta.SerializeToString())
        check(f"whitelist: {cid} si costruisce e si incarta",
              bool(cmd.get_field()) and byte > 0,
              f"campo {cmd.get_field()}, {byte} byte")
    except Exception as exc:
        check(f"whitelist: {cid} si costruisce e si incarta", False,
              f"{type(exc).__name__}: {exc}")

# --- 6) assenza con date, e conto alla rovescia del boost ----------------- #
from datetime import datetime, timedelta, timezone   # noqa: E402
from zoneinfo import ZoneInfo                        # noqa: E402

ROMA = ZoneInfo("Europe/Rome")


def a_roma(t):
    return t.ToDatetime(tzinfo=timezone.utc).astimezone(ROMA).strftime("%Y-%m-%d %H:%M")


date_di = b.build_command("away_mode_on", None, 1, ctx,
                          {"start": "2026-11-01", "end": "2026-11-08"})
m3 = date_di._message
check("assenza con date: inizio = 1 novembre a mezzanotte",
      a_roma(m3.current_time) == "2026-11-01 00:00", a_roma(m3.current_time))
check("assenza con date: fine = mezzanotte dell'8 (da li' riparte)",
      a_roma(m3.end_time) == "2026-11-08 00:00", a_roma(m3.end_time))
check("assenza con date: casa a 15 °C", abs(m3.target_room_temperature - 15.0) < 1e-6)
rifiuta("assenza: fine prima dell'inizio", "away_mode_on", None, 1, ctx,
        {"start": "2026-11-08", "end": "2026-11-01"})
rifiuta("assenza: data illeggibile", "away_mode_on", None, 1, ctx,
        {"start": "domani", "end": "2026-11-08"})
rifiuta("assenza: oltre 180 giorni", "away_mode_on", None, 1, ctx,
        {"start": "2026-11-01", "end": "2028-11-08"})

fra = (datetime.now(timezone.utc) + timedelta(minutes=20)).isoformat().replace("+00:00", "Z")
check("boost in corso: restano ~1200 secondi",
      abs(b.boost_restanti({"hot_water_boost": {"end_time": fra}}) - 1200) <= 3)
check("boost finito: zero secondi",
      b.boost_restanti({"hot_water_boost": {"end_time": "2026-01-01T00:00:00Z"}}) == 0)
check("boost mai avviato: zero secondi", b.boost_restanti({}) == 0)

# L'impianto puo' mandare la fine del boost SENZA fuso (ora locale). E' la forma
# che ha fatto rispondere 502 a /summary: la sottrazione con "adesso" (che il
# fuso ce l'ha) solleva TypeError, e l'app resta senza dati ("Nessun dato").
# Il conto alla rovescia deve reggere entrambe le forme.
_fine_naive = (datetime.now(b.PLANT_TZ) + timedelta(seconds=1200)).strftime("%Y-%m-%dT%H:%M:%S")
check("boost con fine senza fuso (ora dell'impianto)",
      abs(b.boost_restanti({"hot_water_boost": {"end_time": _fine_naive}}) - 1200) <= 3)
check("riassunto con fine senza fuso (la forma che dava 502)",
      (b.summarize({"hot_water_boost": {"end_time": _fine_naive}}) or {}).get("boost_active") is True)
check("boost con fine fuori orario (fuso diverso)",
      b.boost_restanti({"hot_water_boost": {"end_time": "2020-01-01T00:00:00"}}) == 0)
riassunto = b.summarize({"hot_water_boost": {"end_time": fra}})
check("riassunto: dice che il boost e' in corso", riassunto.get("boost_active") is True)
check("riassunto: porta i secondi del boost",
      abs(riassunto.get("boost_remaining_seconds", 0) - 1200) <= 3,
      str(riassunto.get("boost_remaining_seconds")))
check("riassunto senza boost: spento e zero",
      b.summarize({}).get("boost_active") is False
      and b.summarize({}).get("boost_remaining_seconds") == 0)

# --- 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)