solar-of-things-ha/tests/test_energy_flow.py

"""Tests for the energy-flow fallback mapping.

Covers the fallback added for inverter / WiFi-dongle firmware families that
never populate the historical time-series endpoint, leaving every realtime
entity "unknown" while the portal shows live data (issue #7).

The field names and the sample values in ``ISSUE_7_NIGHT_PAYLOAD`` are taken
verbatim from that report, which was captured at night — hence the zeros.
The daylight payloads (``ISSUE_7_AC_PV_CHARGING`` / ``ISSUE_7_NOAC_PV_DISCHARGING``)
are trimmed from hidemichixt-creator's four-state capture set posted
2026-09-15, which settled the unit and sign questions the night capture
could not.
"""
from __future__ import annotations

import pytest

from custom_components.solar_of_things.api import (
    SolarOfThingsAPI,
    TokenExpiredError,
    has_realtime_values,
    map_energy_flow_fields,
)
from custom_components.solar_of_things.const import ENERGY_FLOW_UNVERIFIED

# Exactly the fields reported in issue #7 (UWB1 inverter, night-time reading).
ISSUE_7_NIGHT_PAYLOAD = {
    "pv1Power": 0,
    "pv2Power": 0,
    "generationPower": 0,
    "batteryPower": 9,
    "positiveTerminalBatteryCurrent": 0.4,
    "negativeTerminalBatteryCurrent": 0,
    "positiveTerminalBatteryVoltage": 26.6,
    "bmsBatteryVoltage": 26.6,
    "batteryPercentage": 100,
    "bmsSOC": 100,
    "load_power": 0,
    "aPhaseOutputVoltage": 222.8,
    "aPhaseOutputFrequency": 50,
}

# Trimmed from UPS_AC_PV_BatteryCharging.txt (issue #7, 2026-09-15): AC
# connected, PV producing, battery charging.
ISSUE_7_AC_PV_CHARGING = {
    "pv1Power": 234,
    "pv2Power": 0,
    "load_power": 0.614,  # kW, per the payload's own "unit" field
    "aPhaseMainsPower": -1100,  # W, per the payload's own "unit" field
    "bPhaseMainsPower": 0,
    "cPhaseMainsPower": 0,
    "positiveTerminalBatteryCurrent": -27,
    "negativeTerminalBatteryCurrent": 0,
    "batteryPower": -720,
    "bmsSOC": 54,
}

# Trimmed from UPS_NoAC_PV_BatteryDischarging.txt (issue #7, 2026-09-15): AC
# disconnected, PV producing, battery discharging.
ISSUE_7_NOAC_PV_DISCHARGING = {
    "pv1Power": 262,
    "pv2Power": 0,
    "load_power": 0.635,
    "aPhaseMainsPower": 0,
    "bPhaseMainsPower": 0,
    "cPhaseMainsPower": 0,
    "positiveTerminalBatteryCurrent": 17.6,
    "negativeTerminalBatteryCurrent": 0,
    "batteryPower": 462,
    "bmsSOC": 54,
}


# ─── Pure mapping ──────────────────────────────────────────────────────────────

def test_issue_7_payload_maps_to_canonical_keys() -> None:
    """The reported payload must populate the previously-unknown sensors."""
    mapped = map_energy_flow_fields(ISSUE_7_NIGHT_PAYLOAD)

    # Units confirmed by a non-zero observed value in the report — publishable.
    assert mapped["pvInputPower"] == 0.0
    assert mapped["batteryVoltage"] == 26.6
    assert mapped["batterySOC"] == 100.0
    assert mapped["batteryPower"] == 9.0
    # Confirmed by the 2026-09-15 daylight captures (see below) — 0 at night
    # is a real reading now, not an unconfirmed guess.
    assert mapped["acOutputActivePower"] == 0.0
    assert mapped["batteryChargingCurrent"] == 0.0
    assert mapped["batteryDischargeCurrent"] == 0.4  # idle trickle, per the report

    # loadPower (camelCase) has no confirmed source — must stay ABSENT rather
    # than publish a possibly-wrong value. See ENERGY_FLOW_UNVERIFIED in const.py.
    assert "loadPower" not in mapped


def test_ac_output_power_mapped_from_load_power_kw() -> None:
    """load_power is confirmed kW by the payload's own "unit" tag → x1000."""
    mapped = map_energy_flow_fields(ISSUE_7_AC_PV_CHARGING)
    assert mapped["acOutputActivePower"] == 614.0


def test_grid_import_and_feed_in_split_from_signed_mains_power() -> None:
    """aPhaseMainsPower is confirmed W; negative means importing from mains.

    Sign confirmed by conservation of power: load (614 W) + battery charging
    (720 W) - PV (234 W) = 1100 W, matching |aPhaseMainsPower| exactly.
    """
    mapped = map_energy_flow_fields(ISSUE_7_AC_PV_CHARGING)
    assert mapped["gridPower"] == 1100.0
    assert mapped["feedInPower"] == 0.0  # not exporting while importing

    # No AC connection at all -> the field reads 0, which is a real
    # "no grid" measurement, not a missing one.
    mapped_no_ac = map_energy_flow_fields(ISSUE_7_NOAC_PV_DISCHARGING)
    assert mapped_no_ac["gridPower"] == 0.0
    assert mapped_no_ac["feedInPower"] == 0.0


def test_battery_charging_and_discharge_current_split_by_sign() -> None:
    """positiveTerminalBatteryCurrent flips sign with charge direction.

    Confirmed across all four states captured for issue #7: negative while
    charging, positive while discharging.
    """
    charging = map_energy_flow_fields(ISSUE_7_AC_PV_CHARGING)
    assert charging["batteryChargingCurrent"] == 27.0
    assert charging["batteryDischargeCurrent"] == 0.0

    discharging = map_energy_flow_fields(ISSUE_7_NOAC_PV_DISCHARGING)
    assert discharging["batteryChargingCurrent"] == 0.0
    assert discharging["batteryDischargeCurrent"] == 17.6


def test_unverified_fields_are_never_mapped() -> None:
    """No field in ENERGY_FLOW_UNVERIFIED may produce a sensor value.

    A 1000x-wrong power value feeds the HA Energy dashboard and long-term
    statistics, which a later fix cannot un-poison; an inverted charge/discharge
    current is actively misleading. Both are worse than an unknown sensor.

    This is the guard that stops a well-meaning change from re-enabling one of
    these before a confirming capture exists.
    """
    for field in ENERGY_FLOW_UNVERIFIED:
        assert map_energy_flow_fields({field: 500}) == {}, field

    # And as a group, mirroring the shape of a real payload.
    assert map_energy_flow_fields({f: 500 for f in ENERGY_FLOW_UNVERIFIED}) == {}


def test_multi_string_pv_is_summed_at_face_value() -> None:
    """Per-string PV inputs are reported in W and summed without scaling."""
    mapped = map_energy_flow_fields(
        {"pv1Power": 1200, "pv2Power": 800, "load_power": 1.5}
    )
    assert mapped["pvInputPower"] == 2000.0
    # load_power would need an unconfirmed kW→W conversion, so it stays out.
    assert "loadPower" not in mapped


def test_generation_power_is_not_used_as_a_pv_fallback() -> None:
    """generationPower is an aggregate in kW and its x1000 is unconfirmed.

    Excluding it means a device reporting ONLY generationPower gets an unknown
    PV sensor rather than one that may be 1000x wrong.
    """
    assert map_energy_flow_fields({"generationPower": 2.4}) == {}
    # A per-string reading is still mapped when one is present.
    assert (
        map_energy_flow_fields({"pv1Power": 500, "generationPower": 2.4})[
            "pvInputPower"
        ]
        == 500.0
    )


@pytest.mark.parametrize(
    ("raw", "expected"),
    [
        (77, 77.0),
        ("77", 77.0),
        ({"value": 63}, 63.0),
        ([10, 20, 44], 44.0),  # latest-wins, as the time-series endpoint returns
    ],
)
def test_accepts_the_value_shapes_this_portal_uses(raw, expected) -> None:
    assert map_energy_flow_fields({"bmsSOC": raw})["batterySOC"] == expected


@pytest.mark.parametrize("raw", [None, True, False, "n/a", "", {}, []])
def test_rejects_unusable_values_rather_than_publishing_garbage(raw) -> None:
    assert "batterySOC" not in map_energy_flow_fields({"bmsSOC": raw})


@pytest.mark.parametrize("payload", [None, {}, [1, 2], "text", 0])
def test_hostile_input_returns_empty_mapping(payload) -> None:
    assert map_energy_flow_fields(payload) == {}


def test_unknown_firmware_fields_are_ignored() -> None:
    assert map_energy_flow_fields({"someBrandNewKey": 5}) == {}


# ─── Realtime probe ────────────────────────────────────────────────────────────

def test_probe_treats_zero_as_a_real_reading() -> None:
    """A device reporting 0 W at night is working — must not trigger fallback."""
    assert has_realtime_values({"pvInputPower": 0}) is True


@pytest.mark.parametrize(
    "values", [{}, {"pvInputPower": None}, {"loadPower": 5}, None, "text"]
)
def test_probe_false_when_no_realtime_key_has_a_value(values) -> None:
    assert has_realtime_values(values) is False


# ─── fetch_latest_data integration ─────────────────────────────────────────────

@pytest.fixture
def api_factory():
    """Build an API instance with stubbed transport, tracking endpoint calls."""

    def _factory(history_fields, flow_fields=None, flow_error=None):
        api = SolarOfThingsAPI(iot_token="test-token")
        api._ensure_token_valid = lambda: None  # no network
        calls = {"time_series": 0, "energy_flow": 0}

        def fake_post(path, payload, **kwargs):
            calls["time_series"] += 1
            return {"code": 0, "data": {"payload": {"fields": history_fields}}}

        def fake_get(path, params, **kwargs):
            calls["energy_flow"] += 1
            if flow_error is not None:
                raise flow_error
            return {
                "code": 0,
                "data": {"deviceAttributeState": {"fields": flow_fields or {}}},
            }

        api._post = fake_post
        api._get = fake_get
        return api, calls

    return _factory


def test_working_device_is_unaffected(api_factory) -> None:
    """Regression guard: values and call pattern must match pre-fallback code."""
    history = {
        "pvInputPower": [100],
        "acOutputActivePower": [1.2],
        "batteryVoltage": [52.0],
        "batteryDischargeCurrent": [2.0],
        "batteryChargingCurrent": [0.0],
        "feedInPower": [0],
        "batterySOC": [90],
    }
    api, calls = api_factory(history, flow_fields={"bmsSOC": 1})
    result = api.fetch_latest_data("device-1")

    assert result["acOutputActivePower"] == 1200.0     # kW → W
    assert result["batteryPower"] == 104.0             # (2.0 - 0.0) * 52.0
    assert result["gridPower"] == 1204.0               # max(0, 1200-100+104+0)
    assert result["loadPower"] == 1200.0
    # The fallback must cost a working install nothing.
    assert calls["energy_flow"] == 0


def test_hpvinv02_alternate_key_names_are_mapped_to_canonical(api_factory) -> None:
    """Siseli HPVINV02 ("Inverter Top One" gather protocol) reports PV/AC/SOC
    under pvPower / outputActivePower / batteryCapacity instead of the
    documented pvInputPower / acOutputActivePower / batterySOC — these stayed
    "unknown" forever until the alias mapping was added.
    """
    history = {
        "pvPower": [1.5],
        "outputActivePower": [1.1],
        "batteryCapacity": [77],
        "batteryVoltage": [52.0],
        "batteryDischargeCurrent": [2.0],
        "batteryChargingCurrent": [0.0],
        "feedInPower": [0],
    }
    api, calls = api_factory(history, flow_fields={"bmsSOC": 1})
    result = api.fetch_latest_data("device-hpvinv02")

    assert result["pvInputPower"] == 1500.0        # kW → W, via the alias
    assert result["acOutputActivePower"] == 1100.0  # kW → W, via the alias
    assert result["batterySOC"] == 77
    # The alternate keys must not leak into the published values.
    assert "pvPower" not in result
    assert "outputActivePower" not in result
    assert "batteryCapacity" not in result
    # Derived values still come from the shared _apply_derived_values path.
    assert result["batteryPower"] == 104.0          # (2.0 - 0.0) * 52.0
    assert result["loadPower"] == 1100.0
    assert calls["energy_flow"] == 0


def test_documented_keys_win_when_a_device_reports_both(api_factory) -> None:
    """A device that already uses the documented key names must be unaffected
    by the alias — the canonical value always wins, the alias is discarded,
    and (critically) pvInputPower is NOT kW->W converted on this path: it's
    already W for every device this integration supported before HPVINV02,
    and converting it unconditionally regressed test_working_device_is_unaffected.
    """
    history = {
        "pvInputPower": [2.0],
        "pvPower": [999],
        "acOutputActivePower": [1.0],
        "batterySOC": [55],
        "batteryCapacity": [10],
    }
    api, _ = api_factory(history, flow_fields={})
    result = api.fetch_latest_data("device-both-keys")

    assert result["pvInputPower"] == 2.0
    assert result["batterySOC"] == 55


def test_fallback_populates_sensors_when_time_series_is_empty(api_factory) -> None:
    api, calls = api_factory({}, flow_fields=ISSUE_7_NIGHT_PAYLOAD)
    result = api.fetch_latest_data("device-2")

    assert calls["energy_flow"] == 1
    assert result["batteryVoltage"] == 26.6
    assert result["batterySOC"] == 100.0
    assert result["batteryPower"] == 9.0


def test_measured_battery_power_is_not_overwritten_by_the_estimate(api_factory) -> None:
    """The flow endpoint reports batteryPower directly; keep it.

    The terminal currents are no longer mapped from flow data (direction
    unconfirmed), so no voltage x current estimate can be derived from this
    payload at all — but the no-clobber precedence must still hold for devices
    that get their currents from the time-series path.
    """
    api, _ = api_factory({}, flow_fields=ISSUE_7_NIGHT_PAYLOAD)
    assert api.fetch_latest_data("device-3")["batteryPower"] == 9.0


def test_time_series_values_win_over_the_fallback(api_factory) -> None:
    api, calls = api_factory({"batterySOC": [42]}, flow_fields={"bmsSOC": 99})
    result = api.fetch_latest_data("device-4")

    assert result["batterySOC"] == 42
    assert calls["energy_flow"] == 0


def test_flow_endpoint_failure_degrades_quietly(api_factory) -> None:
    """An unsupported endpoint must not fail the whole coordinator update."""
    api, _ = api_factory({}, flow_error=RuntimeError("404 not supported"))
    result = api.fetch_latest_data("device-5")

    assert isinstance(result, dict)
    assert result["loadPower"] == 0.0


def test_token_expiry_propagates_for_reauth(api_factory) -> None:
    """TokenExpiredError must reach the coordinator so re-auth can start."""
    api, _ = api_factory({}, flow_error=TokenExpiredError("expired"))
    with pytest.raises(TokenExpiredError):
        api.fetch_latest_data("device-6")


def test_unknown_firmware_logs_field_names_for_reporting(api_factory, caplog) -> None:
    api, _ = api_factory({}, flow_fields={"someBrandNewKey": 5, "another": 7})
    api.fetch_latest_data("device-7")

    assert "another" in caplog.text and "someBrandNewKey" in caplog.text