From 526d5e4f6efb3b4df956703741bee072dc0e8a1f Mon Sep 17 00:00:00 2001 From: SchiZzA Date: Sun, 21 Jun 2026 13:23:34 +0200 Subject: [PATCH] test: reach 100% coverage of custom_components/sws12500 Add focused test modules and extend existing ones to cover every line of the integration (1588 stmts, 0 missing; 295 tests): - test_diagnostics.py: redaction + health_data fallback. - test_binary_battery.py: binary_sensor platform setup / add_new + BatteryBinarySensor. - test_staleness_legacy.py: warmup/threshold stale detection + legacy orphan issue. - test_ecowitt_bridge.py: EcowittBridge parsing/discovery + EcoWittNativeSensor. - test_health.py: HealthCoordinator (online/offline refresh, summary, ingress, forwarding, HTTP endpoint) + HealthDiagnosticSensor. - test_received_ecowitt.py: received_ecowitt_data paths + received_data health branches. - test_windy_more.py: duplicate (409) / rate-limit (429) / 3-strike disable. - test_routes_more.py: RouteInfo.__str__, ingress observer, canonical resolve. - test_utils_conv.py: to_int/to_float edge cases. - config_flow: wslink_port_setup step + initial ecowitt flow. - init: _check_stale time-interval callback. - lifecycle: register_path idempotency (existing-dispatcher branch). Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_binary_battery.py | 217 +++++++++ tests/test_config_flow.py | 57 +++ tests/test_diagnostics.py | 155 +++++++ tests/test_ecowitt_bridge.py | 354 +++++++++++++++ tests/test_health.py | 661 ++++++++++++++++++++++++++++ tests/test_init.py | 38 +- tests/test_integration_lifecycle.py | 24 + tests/test_received_ecowitt.py | 578 ++++++++++++++++++++++++ tests/test_routes_more.py | 82 ++++ tests/test_staleness_legacy.py | 143 ++++++ tests/test_utils_conv.py | 37 ++ tests/test_windy_more.py | 120 +++++ 12 files changed, 2465 insertions(+), 1 deletion(-) create mode 100644 tests/test_binary_battery.py create mode 100644 tests/test_diagnostics.py create mode 100644 tests/test_ecowitt_bridge.py create mode 100644 tests/test_health.py create mode 100644 tests/test_received_ecowitt.py create mode 100644 tests/test_routes_more.py create mode 100644 tests/test_staleness_legacy.py create mode 100644 tests/test_utils_conv.py create mode 100644 tests/test_windy_more.py diff --git a/tests/test_binary_battery.py b/tests/test_binary_battery.py new file mode 100644 index 0000000..17a157d --- /dev/null +++ b/tests/test_binary_battery.py @@ -0,0 +1,217 @@ +"""Tests for the binary sensor platform and battery binary sensor entity. + +Covers: +- `binary_sensor.async_setup_entry` (with and without battery keys in SENSORS_TO_LOAD) +- `binary_sensor.add_new_binary_sensors` (no-op / dedupe / unknown / new) +- `battery_sensors.BatteryBinarySensor` (`is_on` value mapping + `device_info`) +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest + +from custom_components.sws12500.battery_sensors import BatteryBinarySensor +from custom_components.sws12500.battery_sensors_def import BATTERY_BINARY_SENSORS +from custom_components.sws12500.binary_sensor import add_new_binary_sensors, async_setup_entry +from custom_components.sws12500.const import CH2_BATTERY, DOMAIN, INDOOR_BATTERY, OUTSIDE_BATTERY, SENSORS_TO_LOAD +from custom_components.sws12500.data import SWSRuntimeData + + +class _CoordinatorStub: + """Minimal coordinator stub: CoordinatorEntity only stores it, and `is_on` reads `.data`.""" + + def __init__(self, data: dict[str, Any] | None = None) -> None: + self.data: dict[str, Any] = data if data is not None else {} + + +def _make_entry( + *, options: dict[str, Any] | None = None, coordinator: _CoordinatorStub | None = None +) -> tuple[Any, _CoordinatorStub, SWSRuntimeData]: + """Build a config-entry stub carrying typed runtime_data, like the integration does.""" + coordinator = coordinator or _CoordinatorStub() + runtime = SWSRuntimeData( + coordinator=coordinator, # type: ignore[arg-type] + health_coordinator=object(), # type: ignore[arg-type] + last_options={}, + ) + entry = SimpleNamespace( + entry_id="test_entry_id", + options=options if options is not None else {}, + runtime_data=runtime, + ) + return entry, coordinator, runtime + + +@pytest.fixture +def hass(): + # binary_sensor platform setup deletes hass; add_new_binary_sensors also deletes it. + return object() + + +def _capture_add_entities(): + captured: list[Any] = [] + + def _add_entities(entities: list[Any]) -> None: + captured.extend(entities) + + return captured, _add_entities + + +def _desc_for(key: str): + return next(d for d in BATTERY_BINARY_SENSORS if d.key == key) + + +# --- async_setup_entry ----------------------------------------------------- + + +@pytest.mark.asyncio +async def test_setup_creates_entities_for_battery_keys(hass): + entry, coordinator, runtime = _make_entry( + options={SENSORS_TO_LOAD: [OUTSIDE_BATTERY, INDOOR_BATTERY]} + ) + captured, add_entities = _capture_add_entities() + + await async_setup_entry(hass, entry, add_entities) + + # Callback + description map persisted for dynamic entity creation. + assert runtime.add_binary_entities is add_entities + assert set(runtime.binary_descriptions.keys()) == {d.key for d in BATTERY_BINARY_SENSORS} + + # Entities created for the requested battery keys only. + assert all(isinstance(e, BatteryBinarySensor) for e in captured) + created_keys = {e.entity_description.key for e in captured} + assert created_keys == {OUTSIDE_BATTERY, INDOOR_BATTERY} + + # added_binary_keys tracks them. + assert runtime.added_binary_keys == {OUTSIDE_BATTERY, INDOOR_BATTERY} + + # Entities are bound to our coordinator stub. + assert all(e.coordinator is coordinator for e in captured) + + +@pytest.mark.asyncio +async def test_setup_no_battery_keys_adds_nothing(hass): + entry, _coordinator, runtime = _make_entry(options={SENSORS_TO_LOAD: ["outside_temp"]}) + captured, add_entities = _capture_add_entities() + + await async_setup_entry(hass, entry, add_entities) + + # Callback/descriptions still stored, but no entities created. + assert runtime.add_binary_entities is add_entities + assert runtime.binary_descriptions # populated + assert captured == [] + assert runtime.added_binary_keys == set() + + +@pytest.mark.asyncio +async def test_setup_no_sensors_to_load_option_adds_nothing(hass): + entry, _coordinator, runtime = _make_entry() # no options at all + captured, add_entities = _capture_add_entities() + + await async_setup_entry(hass, entry, add_entities) + + assert captured == [] + assert runtime.added_binary_keys == set() + + +# --- add_new_binary_sensors ------------------------------------------------ + + +def test_add_new_is_noop_when_callback_missing(hass): + entry, _coordinator, runtime = _make_entry() + # Platform not set up yet -> no stored callback. + assert runtime.add_binary_entities is None + + # Must not raise and must not populate anything. + add_new_binary_sensors(hass, entry, [OUTSIDE_BATTERY]) + assert runtime.added_binary_keys == set() + + +def test_add_new_ignores_already_added_keys(hass): + entry, _coordinator, runtime = _make_entry() + captured, add_entities = _capture_add_entities() + runtime.add_binary_entities = add_entities + runtime.binary_descriptions = {d.key: d for d in BATTERY_BINARY_SENSORS} + runtime.added_binary_keys = {OUTSIDE_BATTERY} + + add_new_binary_sensors(hass, entry, [OUTSIDE_BATTERY]) + + # Already added -> no new entities, callback not invoked. + assert captured == [] + assert runtime.added_binary_keys == {OUTSIDE_BATTERY} + + +def test_add_new_ignores_unknown_keys(hass): + entry, _coordinator, runtime = _make_entry() + captured, add_entities = _capture_add_entities() + runtime.add_binary_entities = add_entities + runtime.binary_descriptions = {d.key: d for d in BATTERY_BINARY_SENSORS} + + add_new_binary_sensors(hass, entry, ["totally_unknown_key"]) + + assert captured == [] + assert runtime.added_binary_keys == set() + + +def test_add_new_adds_new_known_keys(hass): + entry, coordinator, runtime = _make_entry() + captured, add_entities = _capture_add_entities() + runtime.add_binary_entities = add_entities + runtime.binary_descriptions = {d.key: d for d in BATTERY_BINARY_SENSORS} + + # Mix of new known, unknown, and a key we'll mark already-added. + runtime.added_binary_keys = {INDOOR_BATTERY} + add_new_binary_sensors( + hass, entry, [OUTSIDE_BATTERY, CH2_BATTERY, INDOOR_BATTERY, "unknown"] + ) + + created_keys = {e.entity_description.key for e in captured} + assert created_keys == {OUTSIDE_BATTERY, CH2_BATTERY} + assert all(isinstance(e, BatteryBinarySensor) for e in captured) + assert all(e.coordinator is coordinator for e in captured) + assert runtime.added_binary_keys == {INDOOR_BATTERY, OUTSIDE_BATTERY, CH2_BATTERY} + + +# --- BatteryBinarySensor --------------------------------------------------- + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("0", True), # low battery -> on + (0, True), + ("1", False), # battery OK -> off + (1, False), + (None, None), # missing + ("", None), # empty string + ("x", None), # non-int + ([], None), # non-int / TypeError path + ], +) +def test_is_on_value_mapping(raw, expected): + coordinator = _CoordinatorStub({OUTSIDE_BATTERY: raw}) + sensor = BatteryBinarySensor(coordinator, _desc_for(OUTSIDE_BATTERY)) + + assert sensor.is_on is expected + assert sensor.unique_id == f"{OUTSIDE_BATTERY}_binary" + + +def test_is_on_key_absent_returns_none(): + coordinator = _CoordinatorStub({}) # key not present at all + sensor = BatteryBinarySensor(coordinator, _desc_for(OUTSIDE_BATTERY)) + + assert sensor.is_on is None + + +def test_device_info(): + coordinator = _CoordinatorStub() + sensor = BatteryBinarySensor(coordinator, _desc_for(OUTSIDE_BATTERY)) + + info = sensor.device_info + assert info["name"] == "Weather Station SWS 12500" + assert info["manufacturer"] == "Schizza" + assert info["model"] == "Weather Station SWS 12500" + assert info["identifiers"] == {(DOMAIN,)} diff --git a/tests/test_config_flow.py b/tests/test_config_flow.py index 46eb0d4..0fc71e1 100644 --- a/tests/test_config_flow.py +++ b/tests/test_config_flow.py @@ -25,6 +25,7 @@ from custom_components.sws12500.const import ( WINDY_STATION_ID, WINDY_STATION_PW, WSLINK, + WSLINK_ADDON_PORT, ) from homeassistant import config_entries @@ -410,3 +411,59 @@ async def test_options_flow_ecowitt_uses_get_url_placeholders_and_webhook_defaul ) assert done["type"] == "create_entry" assert done["data"][ECOWITT_ENABLED] is True + + +@pytest.mark.asyncio +async def test_options_flow_wslink_port_setup(hass, enable_custom_integrations) -> None: + """The WSLink add-on port step shows a form and stores the port.""" + # A falsy stored port exercises the 443 default fallback. + entry = MockConfigEntry(domain=DOMAIN, data={}, options={WSLINK_ADDON_PORT: 0}) + entry.add_to_hass(hass) + + init = await hass.config_entries.options.async_init(entry.entry_id) + assert init["type"] == "menu" + + form = await hass.config_entries.options.async_configure( + init["flow_id"], user_input={"next_step_id": "wslink_port_setup"} + ) + assert form["type"] == "form" + assert form["step_id"] == "wslink_port_setup" + + done = await hass.config_entries.options.async_configure( + init["flow_id"], user_input={WSLINK_ADDON_PORT: 8443} + ) + assert done["type"] == "create_entry" + assert done["data"][WSLINK_ADDON_PORT] == 8443 + + +@pytest.mark.asyncio +async def test_config_flow_ecowitt_initial_setup(hass, enable_custom_integrations) -> None: + """Initial config flow: user menu -> ecowitt step creates an Ecowitt-only entry.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": config_entries.SOURCE_USER} + ) + assert result["type"] == "menu" + + with patch( + "custom_components.sws12500.config_flow.get_url", + return_value="http://example.local:8123", + ): + form = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={"next_step_id": "ecowitt"} + ) + assert form["type"] == "form" + assert form["step_id"] == "ecowitt" + placeholders = form.get("description_placeholders") or {} + assert placeholders["url"] == "example.local" + assert placeholders["webhook_id"] + + done = await hass.config_entries.flow.async_configure( + form["flow_id"], + user_input={ + ECOWITT_WEBHOOK_ID: placeholders["webhook_id"], + ECOWITT_ENABLED: True, + }, + ) + assert done["type"] == "create_entry" + assert done["data"][ECOWITT_ENABLED] is True + assert done["data"][LEGACY_ENABLED] is False diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py new file mode 100644 index 0000000..4a4d704 --- /dev/null +++ b/tests/test_diagnostics.py @@ -0,0 +1,155 @@ +"""Tests for the SWS12500 config entry diagnostics. + +These cover both branches of `async_get_config_entry_diagnostics`: +- `runtime_data.health_data` is populated and returned directly. +- `runtime_data.health_data` is None, so it falls back to the live + `health_coordinator.data` snapshot. + +In both cases secret keys listed in `TO_REDACT` must be replaced by the +Home Assistant `async_redact_data` sentinel, while non-secret values pass +through untouched. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +from pytest_homeassistant_custom_component.common import MockConfigEntry + +from custom_components.sws12500.const import ( + API_ID, + API_KEY, + DOMAIN, + POCASI_CZ_API_ID, + POCASI_CZ_API_KEY, + WINDY_STATION_ID, + WINDY_STATION_PW, +) +from custom_components.sws12500.data import SWSRuntimeData +from custom_components.sws12500.diagnostics import async_get_config_entry_diagnostics + +REDACTED = "**REDACTED**" + + +def _make_runtime(*, health_data, health_coordinator) -> SWSRuntimeData: + """Build a runtime data container with lightweight stub coordinators. + + Diagnostics only ever reads `health_data` and `health_coordinator.data`, + so the real coordinators can be replaced with plain stubs. + """ + return SWSRuntimeData( + coordinator=object(), # type: ignore[arg-type] + health_coordinator=health_coordinator, # type: ignore[arg-type] + last_options={}, + health_data=health_data, + ) + + +async def test_diagnostics_uses_persisted_health_data(hass) -> None: + """When `health_data` is present it is returned and secrets are redacted.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={ + API_ID: "secret-api-id", + API_KEY: "secret-api-key", + "name": "Station", + }, + options={ + WINDY_STATION_ID: "secret-windy-id", + WINDY_STATION_PW: "secret-windy-pw", + "interval": 60, + }, + ) + entry.add_to_hass(hass) + + health_data = { + "ID": "secret-station-id", + "PASSWORD": "secret-station-pw", + POCASI_CZ_API_ID: "secret-pocasi-id", + POCASI_CZ_API_KEY: "secret-pocasi-key", + "wsid": "secret-wsid", + "wspw": "secret-wspw", + "status": "ok", + } + # A separate coordinator snapshot that must NOT be used in this branch. + health_coordinator = SimpleNamespace(data={"status": "stale-should-not-appear"}) + + entry.runtime_data = _make_runtime( + health_data=health_data, + health_coordinator=health_coordinator, + ) + + result = await async_get_config_entry_diagnostics(hass, entry) + + # entry_data: secrets redacted, plain value preserved. + assert result["entry_data"][API_ID] == REDACTED + assert result["entry_data"][API_KEY] == REDACTED + assert result["entry_data"]["name"] == "Station" + + # entry_options: secrets redacted, plain value preserved. + assert result["entry_options"][WINDY_STATION_ID] == REDACTED + assert result["entry_options"][WINDY_STATION_PW] == REDACTED + assert result["entry_options"]["interval"] == 60 + + # health_data: the persisted payload is used (not the coordinator snapshot). + assert result["health_data"]["status"] == "ok" + assert result["health_data"]["ID"] == REDACTED + assert result["health_data"]["PASSWORD"] == REDACTED + assert result["health_data"][POCASI_CZ_API_ID] == REDACTED + assert result["health_data"][POCASI_CZ_API_KEY] == REDACTED + assert result["health_data"]["wsid"] == REDACTED + assert result["health_data"]["wspw"] == REDACTED + + # The original payload must be untouched (deepcopy is used internally). + assert health_data["ID"] == "secret-station-id" + + +async def test_diagnostics_falls_back_to_coordinator_data(hass) -> None: + """When `health_data` is None it falls back to the coordinator snapshot.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={API_ID: "secret-api-id", "name": "Station"}, + options={}, + ) + entry.add_to_hass(hass) + + coordinator_data = { + "ID": "secret-station-id", + "PASSWORD": "secret-station-pw", + "status": "live", + } + health_coordinator = SimpleNamespace(data=coordinator_data) + + entry.runtime_data = _make_runtime( + health_data=None, + health_coordinator=health_coordinator, + ) + + result = await async_get_config_entry_diagnostics(hass, entry) + + assert result["entry_data"][API_ID] == REDACTED + assert result["entry_data"]["name"] == "Station" + assert result["entry_options"] == {} + + # Fallback snapshot is used and redacted. + assert result["health_data"]["status"] == "live" + assert result["health_data"]["ID"] == REDACTED + assert result["health_data"]["PASSWORD"] == REDACTED + + +async def test_diagnostics_empty_health_data(hass) -> None: + """A falsy fallback snapshot yields an empty health_data dict.""" + entry = MockConfigEntry(domain=DOMAIN, data={}, options={}) + entry.add_to_hass(hass) + + health_coordinator = SimpleNamespace(data=None) + entry.runtime_data = _make_runtime( + health_data=None, + health_coordinator=health_coordinator, + ) + + result = await async_get_config_entry_diagnostics(hass, entry) + + assert result["health_data"] == {} + assert result["entry_data"] == {} + assert result["entry_options"] == {} diff --git a/tests/test_ecowitt_bridge.py b/tests/test_ecowitt_bridge.py new file mode 100644 index 0000000..255f41e --- /dev/null +++ b/tests/test_ecowitt_bridge.py @@ -0,0 +1,354 @@ +"""Tests for the Ecowitt bridge and native passthrough sensor. + +Covers `custom_components.sws12500.ecowitt`: +- `EcowittBridge`: set_add_entities, process_payload, _on_new_sensor branches, + and the `unmapped_sensor` / `all_sensors` properties. +- `EcoWittNativeSensor`: __init__ (mapped/unmapped stype, station / no station), + native_value, async_added_to_hass / async_will_remove_from_hass and _handle_update. + +The tests drive real `aioecowitt` parsing where practical and construct +`EcoWittSensor` objects directly to exercise deterministic branches. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock + +from aioecowitt import EcoWittSensor, EcoWittSensorTypes +from aioecowitt.station import EcoWittStation +import pytest + +from custom_components.sws12500.const import DOMAIN, REMAP_ECOWITT_COMPAT +from custom_components.sws12500.ecowitt import STYPE_TO_HA, EcowittBridge, EcoWittNativeSensor + +# A realistic Ecowitt POST payload. `model` is required by aioecowitt's +# station extraction. Contains both internally mapped fields (tempf, humidity, +# windspeedmph, baromrelin, dewpointf) and unmapped fields (pm25_ch1, co2). +_PAYLOAD: dict[str, Any] = { + "PASSKEY": "ABC123", + "stationtype": "GW1000", + "model": "GW1000", + "dateutc": "2024-01-01 00:00:00", + "freq": "868M", + "tempf": "68.0", + "humidity": "50", + "windspeedmph": "1.0", + "baromrelin": "29.9", + "pm25_ch1": "12.0", + "co2": "400", +} + + +def _make_bridge() -> EcowittBridge: + """Build a bridge with lightweight hass / config stubs. + + The bridge only stores hass/config; parsing is delegated to aioecowitt. + """ + hass = SimpleNamespace() + config = SimpleNamespace(options={}) + return EcowittBridge(hass, config) + + +def _make_sensor( + *, + key: str = "pm25_ch1", + name: str = "PM2.5 CH1", + stype: EcoWittSensorTypes = EcoWittSensorTypes.PM25, + station: EcoWittStation | None = None, + value: Any = 12.0, +) -> EcoWittSensor: + """Construct a real EcoWittSensor for entity-level tests.""" + if station is None: + station = EcoWittStation( + station="GW1000", + model="GW1000", + frequence="868M", + key="ABC123", + ) + sensor = EcoWittSensor(name, key, stype, station) + sensor.value = value + return sensor + + +# --------------------------------------------------------------------------- # +# EcowittBridge +# --------------------------------------------------------------------------- # + + +def test_bridge_init_registers_new_sensor_cb() -> None: + """The bridge wires its own _on_new_sensor into the listener.""" + bridge = _make_bridge() + assert bridge._on_new_sensor in bridge._listener.new_sensor_cb + assert bridge._add_entities_cb is None + + +def test_set_add_entities_stores_callback() -> None: + """set_add_entities stores the platform callback.""" + bridge = _make_bridge() + cb = MagicMock() + bridge.set_add_entities(cb) + assert bridge._add_entities_cb is cb + + +@pytest.mark.asyncio +async def test_process_payload_returns_mapped_result() -> None: + """process_payload parses payload and returns only internally mapped keys.""" + bridge = _make_bridge() + + result = await bridge.process_payload(dict(_PAYLOAD)) + + # Mapped fields present in the payload are remapped to internal keys. + assert result[REMAP_ECOWITT_COMPAT["tempf"]] == "68.0" + assert result[REMAP_ECOWITT_COMPAT["humidity"]] == "50" + assert result[REMAP_ECOWITT_COMPAT["windspeedmph"]] == "1.0" + assert result[REMAP_ECOWITT_COMPAT["baromrelin"]] == "29.9" + + # Unmapped fields never appear in mapped_result. + assert all(k in REMAP_ECOWITT_COMPAT.values() for k in result) + + # aioecowitt populated the listener with sensors. + assert bridge.all_sensors + assert "ABC123.tempf" in bridge.all_sensors + + +@pytest.mark.asyncio +async def test_process_payload_no_mapped_fields() -> None: + """A payload without mapped fields yields an empty mapped_result.""" + bridge = _make_bridge() + data = { + "PASSKEY": "ABC123", + "stationtype": "GW1000", + "model": "GW1000", + "dateutc": "2024-01-01 00:00:00", + "co2": "400", + } + result = await bridge.process_payload(data) + assert result == {} + + +@pytest.mark.asyncio +async def test_process_payload_creates_native_entities_for_unmapped() -> None: + """With a callback set, unmapped sensors become native entities.""" + bridge = _make_bridge() + created: list[EcoWittNativeSensor] = [] + bridge.set_add_entities(lambda entities: created.extend(entities)) + + await bridge.process_payload(dict(_PAYLOAD)) + + created_keys = {e._ecowitt_sensor.key for e in created} + # Unmapped sensors got native entities ... + assert "pm25_ch1" in created_keys + assert "co2" in created_keys + # ... but mapped sensors did NOT. + assert "tempf" not in created_keys + assert "humidity" not in created_keys + assert all(isinstance(e, EcoWittNativeSensor) for e in created) + + +def test_on_new_sensor_skips_mapped_key() -> None: + """A sensor whose key has an internal mapping creates no entity.""" + bridge = _make_bridge() + bridge.set_add_entities(MagicMock()) + sensor = _make_sensor(key="tempf", stype=EcoWittSensorTypes.TEMPERATURE_F) + + bridge._on_new_sensor(sensor) + + bridge._add_entities_cb.assert_not_called() + assert "tempf" not in bridge._know_native_keys + + +def test_on_new_sensor_skips_already_known() -> None: + """A sensor whose key is already tracked creates no new entity.""" + bridge = _make_bridge() + cb = MagicMock() + bridge.set_add_entities(cb) + bridge._know_native_keys.add("pm25_ch1") + sensor = _make_sensor(key="pm25_ch1") + + bridge._on_new_sensor(sensor) + + cb.assert_not_called() + + +def test_on_new_sensor_no_callback_is_noop() -> None: + """Without a platform callback the discovery is a no-op (not tracked).""" + bridge = _make_bridge() + sensor = _make_sensor(key="pm25_ch1") + + bridge._on_new_sensor(sensor) # _add_entities_cb is None + + assert "pm25_ch1" not in bridge._know_native_keys + + +def test_on_new_sensor_creates_entity() -> None: + """An unmapped, unknown sensor creates and registers a native entity.""" + bridge = _make_bridge() + cb = MagicMock() + bridge.set_add_entities(cb) + sensor = _make_sensor(key="pm25_ch1") + + bridge._on_new_sensor(sensor) + + assert "pm25_ch1" in bridge._know_native_keys + cb.assert_called_once() + (entities,) = cb.call_args.args + assert len(entities) == 1 + assert isinstance(entities[0], EcoWittNativeSensor) + assert entities[0]._ecowitt_sensor is sensor + + +@pytest.mark.asyncio +async def test_unmapped_sensor_property() -> None: + """unmapped_sensor returns only sensors without an internal mapping.""" + bridge = _make_bridge() + await bridge.process_payload(dict(_PAYLOAD)) + + unmapped = bridge.unmapped_sensor + keys = {s.key for s in unmapped.values()} + + assert "pm25_ch1" in keys + assert "co2" in keys + # Mapped sensors are excluded. + assert "tempf" not in keys + assert "humidity" not in keys + + +@pytest.mark.asyncio +async def test_all_sensors_property() -> None: + """all_sensors returns the listener's full sensor dict.""" + bridge = _make_bridge() + await bridge.process_payload(dict(_PAYLOAD)) + + assert bridge.all_sensors is bridge._listener.sensors + assert "ABC123.tempf" in bridge.all_sensors + assert "ABC123.pm25_ch1" in bridge.all_sensors + + +# --------------------------------------------------------------------------- # +# EcoWittNativeSensor +# --------------------------------------------------------------------------- # + + +def test_native_sensor_init_with_mapped_stype() -> None: + """A known stype sets device class / unit / state class and device info.""" + station = EcoWittStation( + station="GW1000", model="GW1000", frequence="868M", key="ABC123" + ) + sensor = _make_sensor(key="pm25_ch1", stype=EcoWittSensorTypes.PM25, station=station) + + entity = EcoWittNativeSensor(sensor) + + assert entity._attr_unique_id == "ecowitt_pm25_ch1" + assert entity._attr_name == "PM2.5 CH1" + assert entity._attr_translation_key is None + + device_class, unit, state_class = STYPE_TO_HA[EcoWittSensorTypes.PM25] + assert entity._attr_device_class == device_class + assert entity._attr_native_unit_of_measurement == unit + assert entity._attr_state_class == state_class + + # Device info groups the entity under the station device. + info = entity._attr_device_info + assert info["name"] == "Ecowitt GW1000" + assert info["model"] == "GW1000" + assert (DOMAIN, "ecowitt_ABC123") in info["identifiers"] + assert info["manufacturer"] == "Ecowitt impl. from Schizza for SWS12500" + + +def test_native_sensor_init_with_unmapped_stype() -> None: + """An unknown stype leaves device class / unit / state class unset. + + Device info is still attached (the recent change). + """ + assert EcoWittSensorTypes.INTERNAL not in STYPE_TO_HA + station = EcoWittStation( + station="GW1000", model="GW1000", frequence="868M", key="ABC123" + ) + sensor = _make_sensor( + key="runtime", + name="Runtime", + stype=EcoWittSensorTypes.INTERNAL, + station=station, + value="1000", + ) + + entity = EcoWittNativeSensor(sensor) + + # No HA metadata set for unknown types. + assert getattr(entity, "_attr_device_class", None) is None + assert getattr(entity, "_attr_native_unit_of_measurement", None) is None + assert getattr(entity, "_attr_state_class", None) is None + + # Device info is still present. + info = entity._attr_device_info + assert info["name"] == "Ecowitt GW1000" + assert (DOMAIN, "ecowitt_ABC123") in info["identifiers"] + + +def test_native_sensor_init_without_station() -> None: + """A sensor with no station falls back to generic device info.""" + sensor = _make_sensor(key="pm25_ch1", stype=EcoWittSensorTypes.PM25) + # Force the no-station branch. + sensor.station = None # type: ignore[assignment] + + entity = EcoWittNativeSensor(sensor) + + info = entity._attr_device_info + assert info["name"] == "Ecowitt station" + assert info["model"] is None + assert (DOMAIN, "ecowitt") in info["identifiers"] + + +def test_native_value_returns_value() -> None: + """native_value returns the underlying sensor value.""" + sensor = _make_sensor(value=42.0) + entity = EcoWittNativeSensor(sensor) + assert entity.native_value == 42.0 + + +@pytest.mark.parametrize("empty", [None, ""]) +def test_native_value_none_or_empty(empty: Any) -> None: + """native_value maps None and "" to None.""" + sensor = _make_sensor(value=empty) + entity = EcoWittNativeSensor(sensor) + assert entity.native_value is None + + +@pytest.mark.asyncio +async def test_added_and_removed_callback_lifecycle() -> None: + """async_added/async_will_remove register and unregister the update cb.""" + sensor = _make_sensor() + entity = EcoWittNativeSensor(sensor) + + assert entity._handle_update not in sensor.update_cb + + await entity.async_added_to_hass() + assert entity._handle_update in sensor.update_cb + + await entity.async_will_remove_from_hass() + assert entity._handle_update not in sensor.update_cb + + +@pytest.mark.asyncio +async def test_will_remove_when_callback_absent() -> None: + """async_will_remove is safe when the callback was never registered.""" + sensor = _make_sensor() + entity = EcoWittNativeSensor(sensor) + + # Not added; removal must not raise and must not touch the list. + assert entity._handle_update not in sensor.update_cb + await entity.async_will_remove_from_hass() + assert entity._handle_update not in sensor.update_cb + + +def test_handle_update_writes_ha_state() -> None: + """_handle_update forwards to async_write_ha_state.""" + sensor = _make_sensor() + entity = EcoWittNativeSensor(sensor) + entity.async_write_ha_state = MagicMock() # type: ignore[method-assign] + + entity._handle_update() + + entity.async_write_ha_state.assert_called_once_with() diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..7840cfe --- /dev/null +++ b/tests/test_health.py @@ -0,0 +1,661 @@ +"""Coverage tests for the health coordinator and health diagnostic sensors. + +These tests exercise the runtime health model end to end without requiring real +network access. The add-on reachability check in `_async_update_data` is the only +network-bound path; it is covered by monkeypatching the aiohttp session factory +and the network helpers (`async_get_source_ip`, `get_url`). + +Architecture notes (see CRITICAL ARCHITECTURE NOTES in the task brief): +- `HealthCoordinator(hass, config)` forwards `config_entry=config` to + `DataUpdateCoordinator.__init__`, which calls `config.async_on_unload(...)`. + A `MockConfigEntry` supports that, so we always pass one. +- Health persistence writes `config.runtime_data.health_data`; we set + `entry.runtime_data` to a real `SWSRuntimeData` to cover the success path and + also test the `AttributeError` branch when it is missing. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from datetime import datetime +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import aiohttp +from aiohttp import ClientConnectionError +import pytest +from pytest_homeassistant_custom_component.common import MockConfigEntry + +from custom_components.sws12500 import health_coordinator as hc, health_sensor as hs +from custom_components.sws12500.const import ( + DEFAULT_URL, + DOMAIN, + ECOWITT_URL_PREFIX, + HEALTH_URL, + POCASI_CZ_ENABLED, + WINDY_ENABLED, + WSLINK, + WSLINK_ADDON_PORT, + WSLINK_URL, +) +from custom_components.sws12500.data import SWSRuntimeData +from custom_components.sws12500.health_coordinator import HealthCoordinator +from custom_components.sws12500.routes import Routes + +# --------------------------------------------------------------------------- +# Helpers / fixtures +# --------------------------------------------------------------------------- + + +def _make_entry(options: dict[str, Any] | None = None) -> MockConfigEntry: + """Create a config entry usable by the coordinator constructor.""" + return MockConfigEntry(domain=DOMAIN, data={}, options=options or {}) + + +@pytest.fixture +def entry() -> MockConfigEntry: + """A config entry with empty options.""" + return _make_entry() + + +def _attach_runtime_data(entry: MockConfigEntry, coordinator: HealthCoordinator) -> None: + """Attach a real SWSRuntimeData so the persistence success path is exercised.""" + entry.runtime_data = SWSRuntimeData( + coordinator=MagicMock(), + health_coordinator=coordinator, + last_options={}, + ) + + +class _FakeResponse: + """Minimal aiohttp-like response.""" + + def __init__(self, status: int, json_data: Any = None, json_exc: Exception | None = None) -> None: + self.status = status + self._json_data = json_data + self._json_exc = json_exc + + async def json(self, content_type: Any = None) -> Any: # noqa: ARG002 - signature match + if self._json_exc is not None: + raise self._json_exc + return self._json_data + + +class _FakeSession: + """Fake aiohttp session whose `.get()` returns an async context manager. + + `responses` maps a URL substring to either a `_FakeResponse` or an Exception + that should be raised when entering the context manager. + """ + + def __init__(self, responses: dict[str, Any]) -> None: + self._responses = responses + + def get(self, url: str): # noqa: D401 - mimic aiohttp + outcome: Any = None + for needle, value in self._responses.items(): + if needle in url: + outcome = value + break + + @asynccontextmanager + async def _ctx(): + if isinstance(outcome, Exception): + raise outcome + yield outcome + + return _ctx() + + +def _patch_network(monkeypatch, session: _FakeSession, ip: str = "1.2.3.4") -> None: + """Patch the coordinator network helpers to deterministic values.""" + monkeypatch.setattr(hc, "async_get_clientsession", lambda _hass, _verify=False: session) + monkeypatch.setattr(hc, "async_get_source_ip", AsyncMock(return_value=ip)) + monkeypatch.setattr(hc, "get_url", lambda _hass: "http://ha:8123") + + +# --------------------------------------------------------------------------- +# Module-level helpers in health_coordinator.py +# --------------------------------------------------------------------------- + + +def test_protocol_name() -> None: + assert hc._protocol_name(True) == "wslink" + assert hc._protocol_name(False) == "wu" + + +def test_protocol_from_path_all_branches() -> None: + assert hc._protocol_from_path(WSLINK_URL) == "wslink" + assert hc._protocol_from_path(DEFAULT_URL) == "wu" + assert hc._protocol_from_path(HEALTH_URL) == "health" + assert hc._protocol_from_path(ECOWITT_URL_PREFIX + "/abc") == "ecowitt" + assert hc._protocol_from_path("/something/else") == "unknown" + + +def test_empty_forwarding_state() -> None: + enabled = hc._empty_forwarding_state(True) + assert enabled == { + "enabled": True, + "last_status": "idle", + "last_error": None, + "last_attempt_at": None, + } + disabled = hc._empty_forwarding_state(False) + assert disabled["enabled"] is False + assert disabled["last_status"] == "disabled" + + +def test_default_health_data() -> None: + entry = _make_entry({WSLINK: True, WINDY_ENABLED: True, POCASI_CZ_ENABLED: False}) + data = hc._default_health_data(entry) + + assert data["configured_protocol"] == "wslink" + assert data["active_protocol"] == "wslink" + assert data["integration_status"] == "online_wslink" + assert data["addon"]["online"] is False + assert data["addon"]["paths"] == {"wslink": WSLINK_URL, "wu": DEFAULT_URL} + assert data["forwarding"]["windy"]["enabled"] is True + assert data["forwarding"]["pocasi"]["enabled"] is False + assert data["last_ingress"]["reason"] == "no_data" + + +def test_default_health_data_wu_default() -> None: + data = hc._default_health_data(_make_entry()) + assert data["configured_protocol"] == "wu" + assert data["integration_status"] == "online_wu" + + +# --------------------------------------------------------------------------- +# HealthCoordinator construction & persistence +# --------------------------------------------------------------------------- + + +def test_store_runtime_health_success(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + payload = {"hello": "world"} + coordinator._store_runtime_health(payload) + + assert entry.runtime_data.health_data == payload + # deepcopy: stored value is independent of the source dict + assert entry.runtime_data.health_data is not payload + + +def test_store_runtime_health_missing_runtime_data(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + # entry.runtime_data is unset -> AttributeError branch must be swallowed. + assert getattr(entry, "runtime_data", None) is None + coordinator._store_runtime_health({"a": 1}) # must not raise + + +def test_commit_publishes_and_persists(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + new_data = hc._default_health_data(entry) + new_data["integration_status"] = "custom" + + result = coordinator._commit(new_data) + + assert result is new_data + assert coordinator.data["integration_status"] == "custom" + assert entry.runtime_data.health_data["integration_status"] == "custom" + + +# --------------------------------------------------------------------------- +# _refresh_summary branches +# --------------------------------------------------------------------------- + + +def test_refresh_summary_degraded_by_reason(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + data = hc._default_health_data(entry) + data["configured_protocol"] = "wu" + data["last_ingress"] = {"protocol": "wu", "accepted": False, "reason": "route_disabled"} + + coordinator._refresh_summary(data) + + assert data["integration_status"] == "degraded" + # not accepted -> active falls back to configured protocol + assert data["active_protocol"] == "wu" + + +def test_refresh_summary_degraded_by_protocol_mismatch(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + data = hc._default_health_data(entry) + data["configured_protocol"] = "wu" + # last protocol differs from configured -> degraded even when accepted + data["last_ingress"] = {"protocol": "wslink", "accepted": True, "reason": "accepted"} + + coordinator._refresh_summary(data) + + assert data["integration_status"] == "degraded" + # accepted + recognized protocol -> active protocol tracks the ingress + assert data["active_protocol"] == "wslink" + + +def test_refresh_summary_online_protocol(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + data = hc._default_health_data(entry) + data["configured_protocol"] = "wslink" + data["last_ingress"] = {"protocol": "wslink", "accepted": True, "reason": "accepted"} + + coordinator._refresh_summary(data) + + assert data["integration_status"] == "online_wslink" + assert data["active_protocol"] == "wslink" + + +def test_refresh_summary_online_idle(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + data = hc._default_health_data(entry) + data["configured_protocol"] = "wu" + data["last_ingress"] = {"protocol": "unknown", "accepted": False, "reason": "no_data"} + + coordinator._refresh_summary(data) + + assert data["integration_status"] == "online_idle" + assert data["active_protocol"] == "wu" + + +# --------------------------------------------------------------------------- +# _async_update_data: offline and online paths +# --------------------------------------------------------------------------- + + +async def test_async_update_data_addon_offline(hass, monkeypatch) -> None: + entry = _make_entry({WSLINK_ADDON_PORT: 8443}) + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + session = _FakeSession({"/healthz": ClientConnectionError("boom")}) + _patch_network(monkeypatch, session) + + data = await coordinator._async_update_data() + + addon = data["addon"] + assert addon["online"] is False + assert addon["health_url"] == "https://1.2.3.4:8443/healthz" + assert addon["info_url"] == "https://1.2.3.4:8443/status/internal" + assert addon["home_assistant_url"] == "http://ha:8123" + assert addon["home_assistant_source_ip"] == "1.2.3.4" + assert addon["raw_status"] is None + assert addon["name"] is None + + +async def test_async_update_data_addon_online(hass, monkeypatch) -> None: + entry = _make_entry() # no WSLINK_ADDON_PORT -> default 443 + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + raw_status = { + "addon": "wslink-addon", + "version": "1.2.3", + "listen": {"port": 8443, "tls": True}, + "upstream": {"ha_port": 8123}, + "paths": {"wslink": "/custom/wslink", "wu": "/custom/wu"}, + } + session = _FakeSession( + { + "/healthz": _FakeResponse(200), + "/status/internal": _FakeResponse(200, json_data=raw_status), + } + ) + _patch_network(monkeypatch, session) + + data = await coordinator._async_update_data() + + addon = data["addon"] + assert addon["online"] is True + assert addon["health_url"] == "https://1.2.3.4:443/healthz" + assert addon["name"] == "wslink-addon" + assert addon["version"] == "1.2.3" + assert addon["listen_port"] == 8443 + assert addon["tls"] is True + assert addon["upstream_ha_port"] == 8123 + assert addon["paths"] == {"wslink": "/custom/wslink", "wu": "/custom/wu"} + assert addon["raw_status"] == raw_status + + +async def test_async_update_data_info_endpoint_value_error(hass, monkeypatch) -> None: + """Online add-on but the info endpoint returns invalid JSON.""" + entry = _make_entry() + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + info_resp = _FakeResponse(200, json_exc=ValueError("bad json")) + session = _FakeSession( + { + "/healthz": _FakeResponse(200), + "/status/internal": info_resp, + } + ) + _patch_network(monkeypatch, session) + + data = await coordinator._async_update_data() + + addon = data["addon"] + assert addon["online"] is True + assert addon["raw_status"] is None + # raw_status falsy -> add-on metadata stays at defaults + assert addon["name"] is None + assert addon["version"] is None + + +async def test_async_update_data_info_non_200(hass, monkeypatch) -> None: + """Online add-on but info endpoint replies non-200 -> raw_status stays None.""" + entry = _make_entry() + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + session = _FakeSession( + { + "/healthz": _FakeResponse(200), + "/status/internal": _FakeResponse(503), + } + ) + _patch_network(monkeypatch, session) + + data = await coordinator._async_update_data() + assert data["addon"]["online"] is True + assert data["addon"]["raw_status"] is None + + +# --------------------------------------------------------------------------- +# update_routing +# --------------------------------------------------------------------------- + + +def test_update_routing_none(hass) -> None: + entry = _make_entry({WSLINK: True}) + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + coordinator.update_routing(None) + + assert coordinator.data["configured_protocol"] == "wslink" + # routes block unchanged from default when None passed + assert coordinator.data["routes"]["wu_enabled"] is False + + +def test_update_routing_with_routes(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + routes = Routes() + coordinator.update_routing(routes) + + routes_block = coordinator.data["routes"] + assert routes_block["wu_enabled"] is False + assert routes_block["wslink_enabled"] is False + assert routes_block["health_enabled"] is False + assert routes_block["snapshot"] == {} + + +# --------------------------------------------------------------------------- +# record_dispatch +# --------------------------------------------------------------------------- + + +def test_record_dispatch_skips_health(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + before = coordinator.data["last_ingress"].copy() + + request = SimpleNamespace(path=HEALTH_URL, method="GET") + coordinator.record_dispatch(request, route_enabled=True, reason=None) + + # health path is ignored -> last_ingress untouched + assert coordinator.data["last_ingress"] == before + + +def test_record_dispatch_records(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + request = SimpleNamespace(path=WSLINK_URL, method="POST") + coordinator.record_dispatch(request, route_enabled=True, reason=None) + + ingress = coordinator.data["last_ingress"] + assert ingress["protocol"] == "wslink" + assert ingress["path"] == WSLINK_URL + assert ingress["method"] == "POST" + assert ingress["route_enabled"] is True + assert ingress["accepted"] is False + assert ingress["reason"] == "pending" + assert ingress["time"] is not None + + +def test_record_dispatch_records_with_reason(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + request = SimpleNamespace(path=DEFAULT_URL, method="GET") + coordinator.record_dispatch(request, route_enabled=False, reason="route_disabled") + + ingress = coordinator.data["last_ingress"] + assert ingress["reason"] == "route_disabled" + # route_disabled reason makes the summary degraded + assert coordinator.data["integration_status"] == "degraded" + + +# --------------------------------------------------------------------------- +# update_ingress_result +# --------------------------------------------------------------------------- + + +def test_update_ingress_result_accepted(hass) -> None: + entry = _make_entry({WSLINK: True}) + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + request = SimpleNamespace(path=WSLINK_URL, method="POST") + coordinator.update_ingress_result(request, accepted=True, authorized=True) + + ingress = coordinator.data["last_ingress"] + assert ingress["accepted"] is True + assert ingress["authorized"] is True + assert ingress["reason"] == "accepted" + assert ingress["protocol"] == "wslink" + assert coordinator.data["integration_status"] == "online_wslink" + + +def test_update_ingress_result_rejected_default_reason(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + request = SimpleNamespace(path=DEFAULT_URL, method="GET") + coordinator.update_ingress_result(request, accepted=False, authorized=False) + + ingress = coordinator.data["last_ingress"] + assert ingress["accepted"] is False + assert ingress["reason"] == "rejected" + + +def test_update_ingress_result_explicit_reason(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + request = SimpleNamespace(path=DEFAULT_URL, method="GET") + coordinator.update_ingress_result( + request, accepted=False, authorized=None, reason="unauthorized" + ) + + assert coordinator.data["last_ingress"]["reason"] == "unauthorized" + assert coordinator.data["integration_status"] == "degraded" + + +# --------------------------------------------------------------------------- +# update_forwarding +# --------------------------------------------------------------------------- + + +def test_update_forwarding(hass, entry) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + windy = SimpleNamespace( + enabled=True, last_status="ok", last_error=None, last_attempt_at="2026-06-20T10:00:00" + ) + pocasi = SimpleNamespace( + enabled=False, last_status="disabled", last_error="oops", last_attempt_at=None + ) + + coordinator.update_forwarding(windy, pocasi) + + forwarding = coordinator.data["forwarding"] + assert forwarding["windy"] == { + "enabled": True, + "last_status": "ok", + "last_error": None, + "last_attempt_at": "2026-06-20T10:00:00", + } + assert forwarding["pocasi"]["last_error"] == "oops" + + +# --------------------------------------------------------------------------- +# health_status HTTP endpoint +# --------------------------------------------------------------------------- + + +async def test_health_status_endpoint(hass, entry, monkeypatch) -> None: + coordinator = HealthCoordinator(hass, entry) + _attach_runtime_data(entry, coordinator) + + # Avoid network: stub the refresh that health_status awaits. + monkeypatch.setattr(coordinator, "async_request_refresh", AsyncMock(return_value=None)) + + request = SimpleNamespace(path=HEALTH_URL, method="GET") + response = await coordinator.health_status(request) + + assert isinstance(response, aiohttp.web.Response) + assert response.status == 200 + coordinator.async_request_refresh.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# health_sensor.py module helpers +# --------------------------------------------------------------------------- + + +def test_resolve_path_nested() -> None: + data = {"addon": {"online": True}} + assert hs._resolve_path(data, ("addon", "online")) is True + + +def test_resolve_path_missing_key() -> None: + data = {"addon": {}} + assert hs._resolve_path(data, ("addon", "missing")) is None + + +def test_resolve_path_intermediate_not_dict() -> None: + # Intermediate value is not a dict -> _resolve_path returns None. + data = {"addon": "not-a-dict"} + assert hs._resolve_path(data, ("addon", "online")) is None + + +def test_on_off() -> None: + assert hs._on_off(True) == "on" + assert hs._on_off(False) == "off" + assert hs._on_off(None) == "off" + + +def test_accepted_state() -> None: + assert hs._accepted_state(True) == "accepted" + assert hs._accepted_state(False) == "rejected" + + +def test_authorized_state() -> None: + assert hs._authorized_state(None) == "unknown" + assert hs._authorized_state(True) == "authorized" + assert hs._authorized_state(False) == "unauthorized" + + +def test_timestamp_or_none() -> None: + assert hs._timestamp_or_none(123) is None + assert hs._timestamp_or_none(None) is None + parsed = hs._timestamp_or_none("2026-06-20T10:00:00+00:00") + assert isinstance(parsed, datetime) + + +# --------------------------------------------------------------------------- +# health_sensor.py async_setup_entry +# --------------------------------------------------------------------------- + + +async def test_async_setup_entry_creates_sensors(hass, entry) -> None: + coordinator = MagicMock() + coordinator.data = {} + entry.runtime_data = SWSRuntimeData( + coordinator=MagicMock(), + health_coordinator=coordinator, + last_options={}, + ) + + added: list[Any] = [] + + def _add_entities(entities: Any) -> None: + added.extend(entities) + + await hs.async_setup_entry(hass, entry, _add_entities) + + assert len(added) == len(hs.HEALTH_SENSOR_DESCRIPTIONS) + assert all(isinstance(sensor, hs.HealthDiagnosticSensor) for sensor in added) + + +# --------------------------------------------------------------------------- +# HealthDiagnosticSensor +# --------------------------------------------------------------------------- + + +def _description(key: str) -> hs.HealthSensorEntityDescription: + """Return the bundled description for `key`.""" + return next(d for d in hs.HEALTH_SENSOR_DESCRIPTIONS if d.key == key) + + +def _stub_coordinator(data: dict[str, Any]) -> Any: + """CoordinatorEntity.__init__ only stores the coordinator; a stub suffices.""" + return SimpleNamespace(data=data) + + +def test_sensor_native_value_without_value_fn() -> None: + coordinator = _stub_coordinator({"integration_status": "online_wu"}) + sensor = hs.HealthDiagnosticSensor(coordinator, _description("integration_health")) + assert sensor.native_value == "online_wu" + + +def test_sensor_native_value_with_value_fn() -> None: + coordinator = _stub_coordinator({"addon": {"online": True}}) + sensor = hs.HealthDiagnosticSensor(coordinator, _description("wslink_addon_status")) + assert sensor.native_value == "online" + + +def test_sensor_extra_state_attributes_for_integration_health() -> None: + data = {"integration_status": "online_wu", "addon": {"online": False}} + coordinator = _stub_coordinator(data) + sensor = hs.HealthDiagnosticSensor(coordinator, _description("integration_health")) + assert sensor.extra_state_attributes == data + + +def test_sensor_extra_state_attributes_for_other_keys() -> None: + coordinator = _stub_coordinator({"active_protocol": "wu"}) + sensor = hs.HealthDiagnosticSensor(coordinator, _description("active_protocol")) + assert sensor.extra_state_attributes is None + + +def test_sensor_device_info() -> None: + coordinator = _stub_coordinator({}) + sensor = hs.HealthDiagnosticSensor(coordinator, _description("active_protocol")) + info = sensor.device_info + assert info["name"] == "Weather Station SWS 12500" + assert info["manufacturer"] == "Schizza" + assert info["model"] == "Weather Station SWS 12500" + + +def test_sensor_unique_id_and_category() -> None: + coordinator = _stub_coordinator({}) + sensor = hs.HealthDiagnosticSensor(coordinator, _description("active_protocol")) + assert sensor.unique_id == "active_protocol_health" + assert sensor.entity_category == hs.EntityCategory.DIAGNOSTIC diff --git a/tests/test_init.py b/tests/test_init.py index fc6cbf2..5e8fc01 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -15,7 +15,7 @@ to keep these tests focused on setup logic. from __future__ import annotations -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock import pytest from pytest_homeassistant_custom_component.common import MockConfigEntry @@ -23,6 +23,7 @@ from pytest_homeassistant_custom_component.common import MockConfigEntry from custom_components.sws12500 import WeatherDataUpdateCoordinator, async_setup_entry from custom_components.sws12500.const import DOMAIN from custom_components.sws12500.data import SWSRuntimeData +from homeassistant.util import dt as dt_util @pytest.fixture @@ -113,3 +114,38 @@ async def test_weather_data_update_coordinator_can_be_constructed( coordinator = WeatherDataUpdateCoordinator(hass, config_entry) assert coordinator.hass is hass assert coordinator.config is config_entry + + +async def test_check_stale_callback_runs_update( + hass, config_entry: MockConfigEntry, monkeypatch +): + """The hourly _check_stale callback registered during setup runs the stale check.""" + config_entry.add_to_hass(hass) + + monkeypatch.setattr( + "custom_components.sws12500.register_path", + lambda _hass, _coordinator, _coordinator_h, _entry: True, + ) + monkeypatch.setattr( + "custom_components.sws12500.HealthCoordinator.async_config_entry_first_refresh", + AsyncMock(return_value=None), + ) + hass.config_entries.async_forward_entry_setups = AsyncMock(return_value=True) + + # Capture the time-interval callback async_setup_entry registers. + captured: dict = {} + + def _capture(_hass, action, _interval): + captured["cb"] = action + return lambda: None + + monkeypatch.setattr("custom_components.sws12500.async_track_time_interval", _capture) + + stale = MagicMock() + monkeypatch.setattr("custom_components.sws12500.update_stale_sensors_issue", stale) + + assert await async_setup_entry(hass, config_entry) is True + assert "cb" in captured + + captured["cb"](dt_util.utcnow()) + stale.assert_called_once_with(hass, config_entry) diff --git a/tests/test_integration_lifecycle.py b/tests/test_integration_lifecycle.py index a23044d..3872f1d 100644 --- a/tests/test_integration_lifecycle.py +++ b/tests/test_integration_lifecycle.py @@ -413,3 +413,27 @@ async def test_received_data_auth_unauthorized_and_incorrect_data_paths(hass): await coordinator2.received_data( _RequestStub(query={"ID": "id", "PASSWORD": "key"}) ) # type: ignore[arg-type] + + +@pytest.mark.asyncio +async def test_register_path_idempotent_when_routes_exist(hass_with_http): + """A second register_path call reuses the existing dispatcher (no new aiohttp routes).""" + entry = MockConfigEntry( + domain=DOMAIN, + data={}, + options={API_ID: "id", API_KEY: "key", WSLINK: False}, + ) + entry.add_to_hass(hass_with_http) + + coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry) + coordinator_health = HealthCoordinator(hass_with_http, entry) + + assert register_path(hass_with_http, coordinator, coordinator_health, entry) is True + router: _RouterStub = hass_with_http.http.app.router + get_calls_after_first = list(router.add_get_calls) + post_calls_after_first = list(router.add_post_calls) + + # Routes already a Routes instance -> else branch; nothing re-registered on aiohttp. + assert register_path(hass_with_http, coordinator, coordinator_health, entry) is True + assert router.add_get_calls == get_calls_after_first + assert router.add_post_calls == post_calls_after_first diff --git a/tests/test_received_ecowitt.py b/tests/test_received_ecowitt.py new file mode 100644 index 0000000..f4d4e90 --- /dev/null +++ b/tests/test_received_ecowitt.py @@ -0,0 +1,578 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +from aiohttp.web_exceptions import HTTPUnauthorized +import pytest + +from custom_components.sws12500.const import ( + API_ID, + API_KEY, + DEV_DBG, + ECOWITT_ENABLED, + ECOWITT_WEBHOOK_ID, + POCASI_CZ_ENABLED, + SENSORS_TO_LOAD, + WINDY_ENABLED, + WSLINK, +) +from custom_components.sws12500.coordinator import WeatherDataUpdateCoordinator +from homeassistant.util import dt as dt_util + + +@dataclass(slots=True) +class _EcowittRequestStub: + """Minimal aiohttp Request stub for the Ecowitt endpoint. + + The coordinator uses `webdata.match_info.get("webhook_id", "")` and + `await webdata.post()`. + """ + + match_info: dict[str, Any] = field(default_factory=dict) + post_data: dict[str, Any] | None = None + + async def post(self) -> dict[str, Any]: + return self.post_data or {} + + +@dataclass(slots=True) +class _RequestStub: + """Minimal aiohttp Request stub for the legacy endpoint. + + The coordinator uses `webdata.query` and `await webdata.post()`. + """ + + query: dict[str, Any] + post_data: dict[str, Any] | None = None + + async def post(self) -> dict[str, Any]: + return self.post_data or {} + + +def _make_entry( + *, + wslink: bool = False, + api_id: str | None = "id", + api_key: str | None = "key", + windy_enabled: bool = False, + pocasi_enabled: bool = False, + dev_debug: bool = False, + ecowitt_enabled: bool = False, + ecowitt_webhook_id: str | None = None, + health: Any = None, +) -> Any: + """Create a minimal config entry stub with `.options` and `.entry_id`.""" + options: dict[str, Any] = { + WSLINK: wslink, + WINDY_ENABLED: windy_enabled, + POCASI_CZ_ENABLED: pocasi_enabled, + ECOWITT_ENABLED: ecowitt_enabled, + DEV_DBG: dev_debug, + } + if api_id is not None: + options[API_ID] = api_id + if api_key is not None: + options[API_KEY] = api_key + if ecowitt_webhook_id is not None: + options[ECOWITT_WEBHOOK_ID] = ecowitt_webhook_id + + entry = SimpleNamespace() + entry.entry_id = "test_entry_id" + entry.options = options + # DataUpdateCoordinator.__init__ calls config_entry.async_on_unload(...) when a + # config_entry is passed (see WeatherDataUpdateCoordinator.__init__). + entry.async_on_unload = lambda *_args, **_kwargs: None + # Per-entry runtime state lives on entry.runtime_data (SWSRuntimeData) since v2.0. + entry.runtime_data = SimpleNamespace( + health_coordinator=health, + add_sensor_entities=None, + add_binary_entities=None, + last_seen={}, + started_at=dt_util.utcnow(), + ) + return entry + + +def _make_health_stub() -> Any: + """Create a MagicMock-based stub for the HealthCoordinator branches.""" + return SimpleNamespace( + update_ingress_result=MagicMock(), + update_forwarding=MagicMock(), + ) + + +# --------------------------------------------------------------------------- +# received_ecowitt_data +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_received_ecowitt_disabled_returns_403_and_reports_health(hass, monkeypatch): + """Branch 1: Ecowitt disabled -> 403 and health.update_ingress_result(disabled).""" + health = _make_health_stub() + entry = _make_entry(ecowitt_enabled=False, health=health) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + request = _EcowittRequestStub(match_info={"webhook_id": "abc"}) + resp = await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + + assert resp.status == 403 + assert resp.text == "Ecowitt disabled" + health.update_ingress_result.assert_called_once() + _args, kwargs = health.update_ingress_result.call_args + assert kwargs["accepted"] is False + assert kwargs["authorized"] is None + assert kwargs["reason"] == "ecowitt_disabled" + + +@pytest.mark.asyncio +async def test_received_ecowitt_disabled_no_health(hass, monkeypatch): + """Branch 1 without health: covers the `if health` False edge for disabled.""" + entry = _make_entry(ecowitt_enabled=False, health=None) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + request = _EcowittRequestStub(match_info={"webhook_id": "abc"}) + resp = await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + + assert resp.status == 403 + + +@pytest.mark.asyncio +async def test_received_ecowitt_missing_webhook_id_raises_unauthorized(hass, monkeypatch): + """Branch 2: enabled but expected webhook id missing -> HTTPUnauthorized.""" + health = _make_health_stub() + entry = _make_entry( + ecowitt_enabled=True, ecowitt_webhook_id=None, health=health + ) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + request = _EcowittRequestStub(match_info={"webhook_id": "abc"}) + with pytest.raises(HTTPUnauthorized): + await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + + health.update_ingress_result.assert_called_once() + _args, kwargs = health.update_ingress_result.call_args + assert kwargs["accepted"] is False + assert kwargs["authorized"] is False + assert kwargs["reason"] == "ecowitt_invalid_webhook_id" + + +@pytest.mark.asyncio +async def test_received_ecowitt_mismatched_webhook_id_raises_unauthorized(hass, monkeypatch): + """Branch 2: enabled but webhook id mismatch -> HTTPUnauthorized (no health).""" + entry = _make_entry( + ecowitt_enabled=True, ecowitt_webhook_id="expected", health=None + ) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + request = _EcowittRequestStub(match_info={"webhook_id": "wrong"}) + with pytest.raises(HTTPUnauthorized): + await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + + +@pytest.mark.asyncio +async def test_received_ecowitt_success_full_pipeline_with_health_autodiscovery_forwarding_devlog( + hass, monkeypatch +): + """Branch 3 success: covers lines 132-172. + + - process_payload returns a mapped dict + - check_disabled returns new keys -> autodiscovery (update_options, + add_new_binary_sensors, add_new_sensors) + - async_set_updated_data + last_seen + update_stale_sensors_issue + - health.update_ingress_result(accepted) + windy + pocasi forwarding + - health.update_forwarding + - dev log via anonymize + """ + health = _make_health_stub() + entry = _make_entry( + ecowitt_enabled=True, + ecowitt_webhook_id="hook", + windy_enabled=True, + pocasi_enabled=True, + dev_debug=True, + health=health, + ) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + mapped = {"outside_temp": "20"} + coordinator.ecowitt_bridge.process_payload = AsyncMock(return_value=mapped) + + # Autodiscovery: one new sensor key. + monkeypatch.setattr( + "custom_components.sws12500.coordinator.check_disabled", + lambda _mapped, _config: ["outside_temp"], + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.loaded_sensors", lambda _c: [] + ) + + update_options = AsyncMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.update_options", update_options + ) + add_new_sensors = MagicMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.add_new_sensors", add_new_sensors + ) + add_new_binary_sensors = MagicMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.add_new_binary_sensors", + add_new_binary_sensors, + ) + update_stale = MagicMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.update_stale_sensors_issue", + update_stale, + ) + + coordinator.windy.push_data_to_windy = AsyncMock() + coordinator.pocasi.push_data_to_server = AsyncMock() + + anonymize = MagicMock(return_value={"safe": True}) + monkeypatch.setattr("custom_components.sws12500.coordinator.anonymize", anonymize) + log_info = MagicMock() + monkeypatch.setattr("custom_components.sws12500.coordinator._LOGGER.info", log_info) + + coordinator.async_set_updated_data = MagicMock() + + request = _EcowittRequestStub( + match_info={"webhook_id": "hook"}, post_data={"tempf": "68"} + ) + resp = await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + + assert resp.status == 200 + + coordinator.ecowitt_bridge.process_payload.assert_awaited_once() + + # Autodiscovery side-effects. + update_options.assert_awaited_once() + args, _kwargs = update_options.await_args + assert args[2] == SENSORS_TO_LOAD + assert "outside_temp" in args[3] + add_new_sensors.assert_called_once_with(hass, entry, ["outside_temp"]) + add_new_binary_sensors.assert_called_once_with(hass, entry, ["outside_temp"]) + + # Coordinator data + staleness + last_seen. + coordinator.async_set_updated_data.assert_called_once_with(mapped) + update_stale.assert_called_once() + assert "outside_temp" in entry.runtime_data.last_seen + + # Forwarding: windy receives the raw data dict + False, pocasi receives "WU". + coordinator.windy.push_data_to_windy.assert_awaited_once() + w_args, _ = coordinator.windy.push_data_to_windy.await_args + assert isinstance(w_args[0], dict) + assert w_args[1] is False + coordinator.pocasi.push_data_to_server.assert_awaited_once() + p_args, _ = coordinator.pocasi.push_data_to_server.await_args + assert p_args[1] == "WU" + + # Health branches. + health.update_ingress_result.assert_called_once() + _ia, ikw = health.update_ingress_result.call_args + assert ikw["accepted"] is True + assert ikw["authorized"] is True + assert ikw["reason"] == "accepted" + health.update_forwarding.assert_called_once_with(coordinator.windy, coordinator.pocasi) + + # Dev log. + anonymize.assert_called_once() + log_info.assert_called_once() + + +@pytest.mark.asyncio +async def test_received_ecowitt_success_no_health_no_autodiscovery_no_forwarding( + hass, monkeypatch +): + """Branch 3 success with all optional branches False. + + - health is None (skips both `if health` blocks) + - check_disabled returns [] (skips autodiscovery body) + - windy/pocasi disabled (skips forwarding) + - dev debug off (skips dev log) + """ + entry = _make_entry( + ecowitt_enabled=True, + ecowitt_webhook_id="hook", + windy_enabled=False, + pocasi_enabled=False, + dev_debug=False, + health=None, + ) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + mapped = {"outside_temp": "20"} + coordinator.ecowitt_bridge.process_payload = AsyncMock(return_value=mapped) + + monkeypatch.setattr( + "custom_components.sws12500.coordinator.check_disabled", + lambda _mapped, _config: [], + ) + update_stale = MagicMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.update_stale_sensors_issue", + update_stale, + ) + + coordinator.windy.push_data_to_windy = AsyncMock() + coordinator.pocasi.push_data_to_server = AsyncMock() + coordinator.async_set_updated_data = MagicMock() + + request = _EcowittRequestStub(match_info={"webhook_id": "hook"}) + resp = await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + + assert resp.status == 200 + coordinator.async_set_updated_data.assert_called_once_with(mapped) + update_stale.assert_called_once() + coordinator.windy.push_data_to_windy.assert_not_awaited() + coordinator.pocasi.push_data_to_server.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_received_ecowitt_autodiscovery_extends_with_loaded_sensors(hass, monkeypatch): + """Line 138: cover the `_loaded_sensors := loaded_sensors(...)` extend branch.""" + entry = _make_entry(ecowitt_enabled=True, ecowitt_webhook_id="hook", health=None) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + mapped = {"new": "1"} + coordinator.ecowitt_bridge.process_payload = AsyncMock(return_value=mapped) + + monkeypatch.setattr( + "custom_components.sws12500.coordinator.check_disabled", + lambda _mapped, _config: ["new"], + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.loaded_sensors", lambda _c: ["existing"] + ) + + update_options = AsyncMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.update_options", update_options + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.add_new_sensors", MagicMock() + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.add_new_binary_sensors", MagicMock() + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.update_stale_sensors_issue", MagicMock() + ) + coordinator.async_set_updated_data = MagicMock() + + request = _EcowittRequestStub(match_info={"webhook_id": "hook"}) + resp = await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + + assert resp.status == 200 + update_options.assert_awaited_once() + args, _kwargs = update_options.await_args + assert args[2] == SENSORS_TO_LOAD + assert set(args[3]) >= {"new", "existing"} + + +@pytest.mark.asyncio +async def test_health_coordinator_attribute_error_returns_none(hass, monkeypatch): + """Lines 92-93: runtime_data without health_coordinator -> AttributeError -> None.""" + entry = _make_entry(ecowitt_enabled=True, ecowitt_webhook_id="hook") + # Replace runtime_data with one that lacks `health_coordinator`. + entry.runtime_data = SimpleNamespace(last_seen={}) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + assert coordinator._health_coordinator() is None + + mapped = {"outside_temp": "20"} + coordinator.ecowitt_bridge.process_payload = AsyncMock(return_value=mapped) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.check_disabled", + lambda _mapped, _config: [], + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.update_stale_sensors_issue", MagicMock() + ) + coordinator.async_set_updated_data = MagicMock() + + request = _EcowittRequestStub(match_info={"webhook_id": "hook"}) + resp = await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + assert resp.status == 200 + + +@pytest.mark.asyncio +async def test_received_ecowitt_empty_mapped_skips_update_block(hass, monkeypatch): + """Branch 3 success but process_payload returns empty -> skips the `if mapped_data` block.""" + entry = _make_entry(ecowitt_enabled=True, ecowitt_webhook_id="hook", health=None) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + coordinator.ecowitt_bridge.process_payload = AsyncMock(return_value={}) + + update_stale = MagicMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.update_stale_sensors_issue", + update_stale, + ) + coordinator.async_set_updated_data = MagicMock() + + request = _EcowittRequestStub(match_info={"webhook_id": "hook"}) + resp = await coordinator.received_ecowitt_data(request) # type: ignore[arg-type] + + assert resp.status == 200 + coordinator.async_set_updated_data.assert_not_called() + update_stale.assert_not_called() + + +# --------------------------------------------------------------------------- +# received_data health branches (lines 196, 207, 225, 236, 252, 304, 321) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_received_data_wu_missing_credentials_reports_health(hass, monkeypatch): + """Line 196: WU missing credentials -> health.update_ingress_result.""" + health = _make_health_stub() + entry = _make_entry(wslink=False, health=health) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + with pytest.raises(HTTPUnauthorized): + await coordinator.received_data(_RequestStub(query={"foo": "bar"})) # type: ignore[arg-type] + + health.update_ingress_result.assert_called_once() + _a, kw = health.update_ingress_result.call_args + assert kw["reason"] == "missing_credentials" + assert kw["accepted"] is False + assert kw["authorized"] is False + + +@pytest.mark.asyncio +async def test_received_data_wslink_missing_credentials_reports_health(hass, monkeypatch): + """Line 207: WSLink missing credentials -> health.update_ingress_result.""" + health = _make_health_stub() + entry = _make_entry(wslink=True, health=health) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + with pytest.raises(HTTPUnauthorized): + await coordinator.received_data(_RequestStub(query={"foo": "bar"})) # type: ignore[arg-type] + + health.update_ingress_result.assert_called_once() + _a, kw = health.update_ingress_result.call_args + assert kw["reason"] == "missing_credentials" + + +@pytest.mark.asyncio +async def test_received_data_missing_api_id_reports_health(hass, monkeypatch): + """Line 225: missing API ID -> health.update_ingress_result(config_missing_api_id).""" + from custom_components.sws12500.coordinator import IncorrectDataError + + health = _make_health_stub() + entry = _make_entry(wslink=False, api_id=None, api_key="key", health=health) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + with pytest.raises(IncorrectDataError): + await coordinator.received_data( + _RequestStub(query={"ID": "id", "PASSWORD": "key"}) + ) # type: ignore[arg-type] + + health.update_ingress_result.assert_called_once() + _a, kw = health.update_ingress_result.call_args + assert kw["reason"] == "config_missing_api_id" + assert kw["authorized"] is None + + +@pytest.mark.asyncio +async def test_received_data_missing_api_key_reports_health(hass, monkeypatch): + """Line 236: missing API KEY -> health.update_ingress_result(config_missing_api_key).""" + from custom_components.sws12500.coordinator import IncorrectDataError + + health = _make_health_stub() + entry = _make_entry(wslink=False, api_id="id", api_key=None, health=health) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + with pytest.raises(IncorrectDataError): + await coordinator.received_data( + _RequestStub(query={"ID": "id", "PASSWORD": "key"}) + ) # type: ignore[arg-type] + + health.update_ingress_result.assert_called_once() + _a, kw = health.update_ingress_result.call_args + assert kw["reason"] == "config_missing_api_key" + + +@pytest.mark.asyncio +async def test_received_data_wrong_credentials_reports_health(hass, monkeypatch): + """Line 252: wrong credentials -> health.update_ingress_result(unauthorized).""" + health = _make_health_stub() + entry = _make_entry(wslink=False, api_id="id", api_key="key", health=health) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + with pytest.raises(HTTPUnauthorized): + await coordinator.received_data( + _RequestStub(query={"ID": "id", "PASSWORD": "wrong"}) + ) # type: ignore[arg-type] + + health.update_ingress_result.assert_called_once() + _a, kw = health.update_ingress_result.call_args + assert kw["reason"] == "unauthorized" + assert kw["authorized"] is False + + +@pytest.mark.asyncio +async def test_received_data_success_with_health_autodiscovery_and_binary(hass, monkeypatch): + """Lines 304 + 321: success path with health stub and autodiscovery. + + - check_disabled returns a key -> add_new_binary_sensors (line 294/304-region) + - health accepted branch (line 304) + health.update_forwarding (line 321) + """ + health = _make_health_stub() + entry = _make_entry(wslink=False, api_id="id", api_key="key", health=health) + coordinator = WeatherDataUpdateCoordinator(hass, entry) + + remapped = {"x": "1"} + monkeypatch.setattr( + "custom_components.sws12500.coordinator.remap_items", lambda _d: remapped + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.check_disabled", + lambda _r, _c: ["x"], + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.loaded_sensors", lambda _c: [] + ) + + async def _translations(_hass, _domain, _key, **_kwargs): + return "Name" + + monkeypatch.setattr( + "custom_components.sws12500.coordinator.translations", _translations + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.translated_notification", AsyncMock() + ) + monkeypatch.setattr( + "custom_components.sws12500.coordinator.update_options", AsyncMock() + ) + add_new_sensors = MagicMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.add_new_sensors", add_new_sensors + ) + add_new_binary_sensors = MagicMock() + monkeypatch.setattr( + "custom_components.sws12500.coordinator.add_new_binary_sensors", + add_new_binary_sensors, + ) + + coordinator.async_set_updated_data = MagicMock() + + resp = await coordinator.received_data( + _RequestStub(query={"ID": "id", "PASSWORD": "key"}) + ) # type: ignore[arg-type] + + assert resp.status == 200 + add_new_binary_sensors.assert_called_once() + + # Health accepted (line 304) and forwarding (line 321). + assert health.update_ingress_result.call_count == 1 + _a, kw = health.update_ingress_result.call_args + assert kw["accepted"] is True + assert kw["reason"] == "accepted" + health.update_forwarding.assert_called_once_with(coordinator.windy, coordinator.pocasi) diff --git a/tests/test_routes_more.py b/tests/test_routes_more.py new file mode 100644 index 0000000..3b43db8 --- /dev/null +++ b/tests/test_routes_more.py @@ -0,0 +1,82 @@ +"""Additional Routes coverage: __str__, ingress observer calls, canonical fallback.""" + +from __future__ import annotations + +from dataclasses import dataclass +from types import SimpleNamespace +from unittest.mock import MagicMock + +from aiohttp.web import Response +import pytest + +from custom_components.sws12500.routes import RouteInfo, Routes + + +@dataclass(slots=True) +class _RouteStub: + method: str + + +@dataclass(slots=True) +class _RequestStub: + method: str + path: str + + +@pytest.fixture +def routes() -> Routes: + return Routes() + + +async def _handler(_request) -> Response: + return Response(text="OK", status=200) + + +def test_routeinfo_str_contains_fields() -> None: + info = RouteInfo("/a", route=_RouteStub(method="GET"), handler=_handler, enabled=True) + text = str(info) + assert "RouteInfo(" in text + assert "url_path=/a" in text + assert "enabled=True" in text + + +async def test_dispatch_unknown_path_notifies_observer(routes: Routes) -> None: + observer = MagicMock() + routes.set_ingress_observer(observer) + + await routes.dispatch(_RequestStub(method="GET", path="/nope")) # type: ignore[arg-type] + + observer.assert_called_once() + args = observer.call_args.args + assert args[1] is False + assert args[2] == "route_not_registered" + + +async def test_dispatch_known_path_notifies_observer(routes: Routes) -> None: + routes.add_route("/a", _RouteStub(method="GET"), _handler, enabled=True) + observer = MagicMock() + routes.set_ingress_observer(observer) + + resp = await routes.dispatch(_RequestStub(method="GET", path="/a")) # type: ignore[arg-type] + assert resp.status == 200 + + observer.assert_called_once() + args = observer.call_args.args + assert args[1] is True + assert args[2] is None + + +async def test_dispatch_resolves_via_canonical_resource(routes: Routes) -> None: + """A path with a parameter resolves through the aiohttp resource canonical URL.""" + canonical = "/weatherhub/{webhook_id}" + routes.add_route(canonical, _RouteStub(method="POST"), _handler, enabled=True) + + # Direct key "POST:/weatherhub/abc" is absent; fall back to resource.canonical. + request = SimpleNamespace( + method="POST", + path="/weatherhub/abc", + match_info=SimpleNamespace(route=SimpleNamespace(resource=SimpleNamespace(canonical=canonical))), + ) + + resp = await routes.dispatch(request) # type: ignore[arg-type] + assert resp.status == 200 diff --git a/tests/test_staleness_legacy.py b/tests/test_staleness_legacy.py new file mode 100644 index 0000000..6b25c89 --- /dev/null +++ b/tests/test_staleness_legacy.py @@ -0,0 +1,143 @@ +"""Tests for stale-sensor and legacy-battery Repairs issues. + +Covers: +- staleness.py: warmup short-circuit, stale detection (never seen / seen long ago), + fresh sensors clearing the issue, and the create/delete branches of + `update_stale_sensors_issue`. +- legacy.py: orphan legacy battery sensor detection and the create/delete branches + of `update_legacy_battery_issue`. + +Uses the real `hass` fixture so the issue/entity registries behave like production. +""" + +from __future__ import annotations + +from datetime import timedelta + +from pytest_homeassistant_custom_component.common import MockConfigEntry + +from custom_components.sws12500.const import DOMAIN, SENSORS_TO_LOAD +from custom_components.sws12500.data import SWSRuntimeData +from custom_components.sws12500.legacy import _legacy_battery_issue_id, update_legacy_battery_issue +from custom_components.sws12500.staleness import _find_stale_keys, _stale_sensor_issue_id, update_stale_sensors_issue +from homeassistant.helpers import entity_registry as er, issue_registry as ir +from homeassistant.util import dt as dt_util + + +def _make_entry(hass, options: dict) -> MockConfigEntry: + """Create a config entry with typed runtime data attached and added to hass.""" + entry = MockConfigEntry(domain=DOMAIN, data={}, options=options) + entry.add_to_hass(hass) + entry.runtime_data = SWSRuntimeData( + coordinator=object(), # type: ignore[arg-type] + health_coordinator=object(), # type: ignore[arg-type] + last_options=dict(options), + ) + return entry + + +# --------------------------------------------------------------------------- # +# staleness.py +# --------------------------------------------------------------------------- # + + +async def test_warmup_returns_no_stale_keys_and_no_issue(hass): + """During the warmup period no key is considered stale and no issue is raised.""" + entry = _make_entry(hass, {SENSORS_TO_LOAD: ["outside_temp"]}) + # started_at = now -> within WARMUP_PERIOD even with a never-seen key. + entry.runtime_data.started_at = dt_util.utcnow() + + assert _find_stale_keys(entry) == [] + + update_stale_sensors_issue(hass, entry) + + issue_id = _stale_sensor_issue_id(entry) + assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is None + + +async def test_never_seen_key_is_stale_and_creates_issue(hass): + """Past warmup, a loaded key that was never seen is stale -> issue created.""" + entry = _make_entry(hass, {SENSORS_TO_LOAD: ["outside_temp"]}) + entry.runtime_data.started_at = dt_util.utcnow() - timedelta(hours=2) + # last_seen left empty -> outside_temp never reported. + + assert _find_stale_keys(entry) == ["outside_temp"] + + update_stale_sensors_issue(hass, entry) + + issue_id = _stale_sensor_issue_id(entry) + issue = ir.async_get(hass).async_get_issue(DOMAIN, issue_id) + assert issue is not None + assert issue.translation_key == "stale_sensors_detected" + assert issue.translation_placeholders == {"sensors": "outside_temp"} + + +async def test_recently_seen_key_is_not_stale_and_clears_issue(hass): + """Past warmup, a key seen just now is fresh -> issue absent/cleared.""" + entry = _make_entry(hass, {SENSORS_TO_LOAD: ["outside_temp"]}) + entry.runtime_data.started_at = dt_util.utcnow() - timedelta(hours=2) + entry.runtime_data.last_seen = {"outside_temp": dt_util.utcnow()} + + assert _find_stale_keys(entry) == [] + + update_stale_sensors_issue(hass, entry) + + issue_id = _stale_sensor_issue_id(entry) + assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is None + + +async def test_key_seen_long_ago_is_stale(hass): + """A key last seen beyond STALE_THRESHOLD (24h) is stale.""" + entry = _make_entry(hass, {SENSORS_TO_LOAD: ["outside_temp"]}) + entry.runtime_data.started_at = dt_util.utcnow() - timedelta(hours=2) + entry.runtime_data.last_seen = {"outside_temp": dt_util.utcnow() - timedelta(hours=48)} + + assert _find_stale_keys(entry) == ["outside_temp"] + + +async def test_existing_stale_issue_is_deleted_when_keys_become_fresh(hass): + """A previously-created stale issue is cleared once the sensor reports again.""" + entry = _make_entry(hass, {SENSORS_TO_LOAD: ["outside_temp"]}) + entry.runtime_data.started_at = dt_util.utcnow() - timedelta(hours=2) + + # First pass: stale -> issue created. + update_stale_sensors_issue(hass, entry) + issue_id = _stale_sensor_issue_id(entry) + assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is not None + + # Sensor reports -> second pass clears the issue. + entry.runtime_data.last_seen = {"outside_temp": dt_util.utcnow()} + update_stale_sensors_issue(hass, entry) + assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is None + + +# --------------------------------------------------------------------------- # +# legacy.py +# --------------------------------------------------------------------------- # + + +async def test_legacy_battery_sensor_creates_issue(hass): + """A legacy (non-binary) battery sensor in the entity registry raises an issue.""" + entry = _make_entry(hass, {}) + + ent_reg = er.async_get(hass) + ent_reg.async_get_or_create("sensor", DOMAIN, "outside_battery", config_entry=entry) + + update_legacy_battery_issue(hass, entry) + + issue_id = _legacy_battery_issue_id(entry) + issue = ir.async_get(hass).async_get_issue(DOMAIN, issue_id) + assert issue is not None + assert issue.translation_key == "legacy_battery_sensor_deprecated" + assert issue.translation_placeholders is not None + assert issue.translation_placeholders["remove_version"] == "2.1.0" + + +async def test_no_legacy_battery_sensor_clears_issue(hass): + """With no legacy battery sensor present, the issue is deleted/absent.""" + entry = _make_entry(hass, {}) + + update_legacy_battery_issue(hass, entry) + + issue_id = _legacy_battery_issue_id(entry) + assert ir.async_get(hass).async_get_issue(DOMAIN, issue_id) is None diff --git a/tests/test_utils_conv.py b/tests/test_utils_conv.py new file mode 100644 index 0000000..c252388 --- /dev/null +++ b/tests/test_utils_conv.py @@ -0,0 +1,37 @@ +"""Coverage for to_int / to_float edge cases.""" + +from __future__ import annotations + +import pytest + +from custom_components.sws12500.utils import to_float, to_int + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (None, None), + ("", None), + (" ", None), + ("x", None), + ("5", 5), + (7, 7), + ], +) +def test_to_int_edge_cases(value, expected) -> None: + assert to_int(value) == expected + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (None, None), + ("", None), + (" ", None), + ("x", None), + ("5.5", 5.5), + (7, 7.0), + ], +) +def test_to_float_edge_cases(value, expected) -> None: + assert to_float(value) == expected diff --git a/tests/test_windy_more.py b/tests/test_windy_more.py new file mode 100644 index 0000000..49a6f24 --- /dev/null +++ b/tests/test_windy_more.py @@ -0,0 +1,120 @@ +"""Additional Windy branch coverage: duplicate (409), rate limit (429), 3-strike disable.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from custom_components.sws12500.const import WINDY_ENABLED, WINDY_LOGGER_ENABLED, WINDY_STATION_ID, WINDY_STATION_PW +from custom_components.sws12500.windy_func import WindyDuplicatePayloadDetected, WindyPush, WindyRateLimitExceeded + + +@dataclass(slots=True) +class _FakeResponse: + status: int + + async def __aenter__(self) -> "_FakeResponse": + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + return None + + +class _FakeSession: + def __init__(self, response: _FakeResponse) -> None: + self._response = response + + def get(self, url: str, *, params=None, headers=None): + return self._response + + +@pytest.fixture +def hass(): + return SimpleNamespace() + + +def _make_entry(**options: Any): + defaults = { + WINDY_LOGGER_ENABLED: False, + WINDY_ENABLED: True, + WINDY_STATION_ID: "station", + WINDY_STATION_PW: "token", + } + defaults.update(options) + return SimpleNamespace(options=defaults) + + +def test_verify_response_duplicate_raises(hass): + wp = WindyPush(hass, _make_entry()) + with pytest.raises(WindyDuplicatePayloadDetected): + wp.verify_windy_response(_FakeResponse(status=409)) + + +def test_verify_response_rate_limit_raises(hass): + wp = WindyPush(hass, _make_entry()) + with pytest.raises(WindyRateLimitExceeded): + wp.verify_windy_response(_FakeResponse(status=429)) + + +@pytest.mark.asyncio +async def test_push_duplicate_payload_sets_status_and_counts(monkeypatch, hass): + wp = WindyPush(hass, _make_entry()) + wp.next_update = datetime.now() - timedelta(seconds=1) + + monkeypatch.setattr( + "custom_components.sws12500.windy_func.async_get_clientsession", + lambda _h: _FakeSession(_FakeResponse(status=409)), + ) + + ok = await wp.push_data_to_windy({"a": "b"}) + assert ok is True + assert wp.last_status == "duplicate" + assert wp.invalid_response_count == 1 + + +@pytest.mark.asyncio +async def test_push_rate_limited_pauses_five_minutes(monkeypatch, hass): + wp = WindyPush(hass, _make_entry()) + wp.next_update = datetime.now() - timedelta(seconds=1) + + monkeypatch.setattr( + "custom_components.sws12500.windy_func.async_get_clientsession", + lambda _h: _FakeSession(_FakeResponse(status=429)), + ) + + before = datetime.now() + ok = await wp.push_data_to_windy({"a": "b"}) + assert ok is True + assert wp.last_status == "rate_limited_remote" + # next_update pushed ~5 minutes out by the rate-limit handler. + assert wp.next_update > before + timedelta(minutes=4) + + +@pytest.mark.asyncio +async def test_push_duplicate_third_strike_disables(monkeypatch, hass): + wp = WindyPush(hass, _make_entry()) + wp.invalid_response_count = 2 # next duplicate makes it 3 -> finally disables + wp.next_update = datetime.now() - timedelta(seconds=1) + + update_options = AsyncMock(return_value=True) + monkeypatch.setattr( + "custom_components.sws12500.windy_func.update_options", update_options + ) + monkeypatch.setattr( + "custom_components.sws12500.windy_func.persistent_notification.create", + MagicMock(), + ) + monkeypatch.setattr( + "custom_components.sws12500.windy_func.async_get_clientsession", + lambda _h: _FakeSession(_FakeResponse(status=409)), + ) + + ok = await wp.push_data_to_windy({"a": "b"}) + assert ok is True + assert wp.invalid_response_count == 3 + update_options.assert_awaited_once_with(hass, wp.config, WINDY_ENABLED, False)