test: align suite with v2.0 runtime_data architecture
The suite predated the coordinator extraction and the typed runtime_data migration: 5 modules failed to import and 18 tests failed. - Import IncorrectDataError/WeatherDataUpdateCoordinator from .coordinator. - Point received_data monkeypatch targets at custom_components.sws12500.coordinator.* - Stub config entries with async_on_unload + runtime_data (SWSRuntimeData). - Rewrite test_data, test_sensor_platform and test_integration_lifecycle off the removed ENTRY_* hass.data keys onto runtime_data (route dispatcher reuse, ecowitt POST route, no hass.data pop on unload, health first_refresh mocked). - Update config_flow tests for the user menu (pws step) and the options menu gaining wslink_port_setup. - Fix VOCLevel.EXCELENT typo and stale t9 expectations (tuple battery list, HCHO int coercion, connection-gated subset). Result: 168 passed, 0 collection errors. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
SchiZzA
parent
25341f79b3
commit
50d0e9fae7
|
|
@ -13,6 +13,7 @@ from custom_components.sws12500.const import (
|
|||
ECOWITT_ENABLED,
|
||||
ECOWITT_WEBHOOK_ID,
|
||||
INVALID_CREDENTIALS,
|
||||
LEGACY_ENABLED,
|
||||
POCASI_CZ_API_ID,
|
||||
POCASI_CZ_API_KEY,
|
||||
POCASI_CZ_ENABLED,
|
||||
|
|
@ -36,9 +37,15 @@ async def test_config_flow_user_form_then_create_entry(
|
|||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["type"] == "menu"
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
form = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={"next_step_id": "pws"}
|
||||
)
|
||||
assert form["type"] == "form"
|
||||
assert form["step_id"] == "pws"
|
||||
|
||||
user_input = {
|
||||
API_ID: "my_id",
|
||||
API_KEY: "my_key",
|
||||
|
|
@ -46,12 +53,16 @@ async def test_config_flow_user_form_then_create_entry(
|
|||
DEV_DBG: False,
|
||||
}
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=user_input
|
||||
form["flow_id"], user_input=user_input
|
||||
)
|
||||
assert result2["type"] == "create_entry"
|
||||
assert result2["title"] == DOMAIN
|
||||
assert result2["data"] == user_input
|
||||
assert result2["options"] == user_input
|
||||
# The PWS step augments user input with legacy/ecowitt flags.
|
||||
assert result2["data"][API_ID] == "my_id"
|
||||
assert result2["data"][API_KEY] == "my_key"
|
||||
assert result2["data"][LEGACY_ENABLED] is True
|
||||
assert result2["data"][ECOWITT_ENABLED] is False
|
||||
assert result2["options"] == result2["data"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -62,7 +73,13 @@ async def test_config_flow_user_invalid_credentials_api_id(
|
|||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["type"] == "menu"
|
||||
|
||||
form = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={"next_step_id": "pws"}
|
||||
)
|
||||
assert form["type"] == "form"
|
||||
assert form["step_id"] == "pws"
|
||||
|
||||
user_input = {
|
||||
API_ID: INVALID_CREDENTIALS[0],
|
||||
|
|
@ -71,10 +88,10 @@ async def test_config_flow_user_invalid_credentials_api_id(
|
|||
DEV_DBG: False,
|
||||
}
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=user_input
|
||||
form["flow_id"], user_input=user_input
|
||||
)
|
||||
assert result2["type"] == "form"
|
||||
assert result2["step_id"] == "user"
|
||||
assert result2["step_id"] == "pws"
|
||||
assert result2["errors"][API_ID] == "valid_credentials_api"
|
||||
|
||||
|
||||
|
|
@ -86,7 +103,13 @@ async def test_config_flow_user_invalid_credentials_api_key(
|
|||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["type"] == "menu"
|
||||
|
||||
form = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={"next_step_id": "pws"}
|
||||
)
|
||||
assert form["type"] == "form"
|
||||
assert form["step_id"] == "pws"
|
||||
|
||||
user_input = {
|
||||
API_ID: "ok_id",
|
||||
|
|
@ -95,10 +118,10 @@ async def test_config_flow_user_invalid_credentials_api_key(
|
|||
DEV_DBG: False,
|
||||
}
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=user_input
|
||||
form["flow_id"], user_input=user_input
|
||||
)
|
||||
assert result2["type"] == "form"
|
||||
assert result2["step_id"] == "user"
|
||||
assert result2["step_id"] == "pws"
|
||||
assert result2["errors"][API_KEY] == "valid_credentials_key"
|
||||
|
||||
|
||||
|
|
@ -110,7 +133,13 @@ async def test_config_flow_user_invalid_credentials_match(
|
|||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": config_entries.SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == "form"
|
||||
assert result["type"] == "menu"
|
||||
|
||||
form = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input={"next_step_id": "pws"}
|
||||
)
|
||||
assert form["type"] == "form"
|
||||
assert form["step_id"] == "pws"
|
||||
|
||||
user_input = {
|
||||
API_ID: "same",
|
||||
|
|
@ -119,10 +148,10 @@ async def test_config_flow_user_invalid_credentials_match(
|
|||
DEV_DBG: False,
|
||||
}
|
||||
result2 = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], user_input=user_input
|
||||
form["flow_id"], user_input=user_input
|
||||
)
|
||||
assert result2["type"] == "form"
|
||||
assert result2["step_id"] == "user"
|
||||
assert result2["step_id"] == "pws"
|
||||
assert result2["errors"]["base"] == "valid_credentials_match"
|
||||
|
||||
|
||||
|
|
@ -135,7 +164,7 @@ async def test_options_flow_init_menu(hass, enable_custom_integrations) -> None:
|
|||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
assert result["type"] == "menu"
|
||||
assert result["step_id"] == "init"
|
||||
assert set(result["menu_options"]) == {"basic", "ecowitt", "windy", "pocasi"}
|
||||
assert set(result["menu_options"]) == {"basic", "wslink_port_setup", "ecowitt", "windy", "pocasi"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
|
|
@ -1,13 +1,46 @@
|
|||
from custom_components.sws12500.data import (
|
||||
ENTRY_ADD_ENTITIES,
|
||||
ENTRY_COORDINATOR,
|
||||
ENTRY_DESCRIPTIONS,
|
||||
ENTRY_LAST_OPTIONS,
|
||||
)
|
||||
"""Tests for the typed per-entry runtime data container."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from custom_components.sws12500.data import SWSRuntimeData
|
||||
|
||||
|
||||
def test_data_constants():
|
||||
assert ENTRY_COORDINATOR == "coordinator"
|
||||
assert ENTRY_ADD_ENTITIES == "async_add_entities"
|
||||
assert ENTRY_DESCRIPTIONS == "sensor_descriptions"
|
||||
assert ENTRY_LAST_OPTIONS == "last_options"
|
||||
def test_runtime_data_defaults():
|
||||
"""SWSRuntimeData exposes the expected fields with safe defaults.
|
||||
|
||||
The ad-hoc hass.data[DOMAIN][entry_id] string keys (ENTRY_*) were replaced by
|
||||
this typed dataclass stored on entry.runtime_data in v2.0.
|
||||
"""
|
||||
runtime = SWSRuntimeData(
|
||||
coordinator=object(), # type: ignore[arg-type]
|
||||
health_coordinator=object(), # type: ignore[arg-type]
|
||||
last_options={"legacy_enabled": True},
|
||||
)
|
||||
|
||||
assert runtime.last_options == {"legacy_enabled": True}
|
||||
|
||||
# Optional platform wiring defaults.
|
||||
assert runtime.add_sensor_entities is None
|
||||
assert runtime.sensor_descriptions == {}
|
||||
assert runtime.add_binary_entities is None
|
||||
assert runtime.binary_descriptions == {}
|
||||
assert runtime.added_binary_keys == set()
|
||||
|
||||
# Diagnostics / staleness defaults.
|
||||
assert runtime.health_data is None
|
||||
assert runtime.last_seen == {}
|
||||
assert runtime.started_at is not None
|
||||
|
||||
|
||||
def test_runtime_data_collections_are_independent_per_instance():
|
||||
"""Mutable defaults must not be shared between instances."""
|
||||
a = SWSRuntimeData(coordinator=object(), health_coordinator=object(), last_options={}) # type: ignore[arg-type]
|
||||
b = SWSRuntimeData(coordinator=object(), health_coordinator=object(), last_options={}) # type: ignore[arg-type]
|
||||
|
||||
a.sensor_descriptions["x"] = object() # type: ignore[assignment]
|
||||
a.added_binary_keys.add("y")
|
||||
a.last_seen["z"] = object() # type: ignore[assignment]
|
||||
|
||||
assert b.sensor_descriptions == {}
|
||||
assert b.added_binary_keys == set()
|
||||
assert b.last_seen == {}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,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
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -43,6 +44,14 @@ async def test_async_setup_entry_creates_runtime_state(
|
|||
lambda _hass, _coordinator, _coordinator_h, _entry: True,
|
||||
)
|
||||
|
||||
# Calling async_setup_entry directly leaves the entry in NOT_LOADED state, so the
|
||||
# health coordinator's first refresh (which requires SETUP_IN_PROGRESS and does
|
||||
# network I/O) is mocked out to keep this test focused on setup wiring.
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.HealthCoordinator.async_config_entry_first_refresh",
|
||||
AsyncMock(return_value=None),
|
||||
)
|
||||
|
||||
# Avoid depending on Home Assistant integration loader in this test.
|
||||
# This keeps the test focused on our integration's setup behavior.
|
||||
monkeypatch.setattr(
|
||||
|
|
@ -54,9 +63,12 @@ async def test_async_setup_entry_creates_runtime_state(
|
|||
result = await async_setup_entry(hass, config_entry)
|
||||
assert result is True
|
||||
|
||||
# Per-entry state now lives on entry.runtime_data (SWSRuntimeData), not in
|
||||
# hass.data[DOMAIN][entry_id]. hass.data[DOMAIN] only holds shared route state.
|
||||
assert DOMAIN in hass.data
|
||||
assert config_entry.entry_id in hass.data[DOMAIN]
|
||||
assert isinstance(hass.data[DOMAIN][config_entry.entry_id], dict)
|
||||
assert isinstance(config_entry.runtime_data, SWSRuntimeData)
|
||||
assert config_entry.runtime_data.coordinator is not None
|
||||
assert config_entry.runtime_data.health_coordinator is not None
|
||||
|
||||
|
||||
async def test_async_setup_entry_forwards_sensor_platform(
|
||||
|
|
@ -72,6 +84,14 @@ async def test_async_setup_entry_forwards_sensor_platform(
|
|||
lambda _hass, _coordinator, _coordinator_h, _entry: True,
|
||||
)
|
||||
|
||||
# Calling async_setup_entry directly leaves the entry in NOT_LOADED state, so the
|
||||
# health coordinator's first refresh (which requires SETUP_IN_PROGRESS and does
|
||||
# network I/O) is mocked out to keep this test focused on setup wiring.
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.HealthCoordinator.async_config_entry_first_refresh",
|
||||
AsyncMock(return_value=None),
|
||||
)
|
||||
|
||||
# Patch forwarding so we don't need to load real platforms for this unit/integration test.
|
||||
hass.config_entries.async_forward_entry_setups = AsyncMock(return_value=True)
|
||||
|
||||
|
|
|
|||
|
|
@ -9,26 +9,23 @@ from aiohttp.web_exceptions import HTTPUnauthorized
|
|||
import pytest
|
||||
from pytest_homeassistant_custom_component.common import MockConfigEntry
|
||||
|
||||
from custom_components.sws12500 import (
|
||||
HealthCoordinator,
|
||||
IncorrectDataError,
|
||||
WeatherDataUpdateCoordinator,
|
||||
async_setup_entry,
|
||||
async_unload_entry,
|
||||
register_path,
|
||||
update_listener,
|
||||
)
|
||||
from custom_components.sws12500 import async_setup_entry, async_unload_entry, register_path, update_listener
|
||||
from custom_components.sws12500.const import (
|
||||
API_ID,
|
||||
API_KEY,
|
||||
DEFAULT_URL,
|
||||
DOMAIN,
|
||||
ECOWITT_URL_PREFIX,
|
||||
HEALTH_URL,
|
||||
SENSORS_TO_LOAD,
|
||||
WSLINK,
|
||||
WSLINK_URL,
|
||||
)
|
||||
from custom_components.sws12500.data import ENTRY_COORDINATOR, ENTRY_LAST_OPTIONS
|
||||
from custom_components.sws12500.coordinator import IncorrectDataError, WeatherDataUpdateCoordinator
|
||||
from custom_components.sws12500.data import SWSRuntimeData
|
||||
from custom_components.sws12500.health_coordinator import HealthCoordinator
|
||||
|
||||
ECOWITT_PATH = ECOWITT_URL_PREFIX + "/{webhook_id}"
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
|
|
@ -70,16 +67,27 @@ def hass_with_http(hass):
|
|||
return hass
|
||||
|
||||
|
||||
def _mock_health_first_refresh(monkeypatch) -> None:
|
||||
"""Calling async_setup_entry directly leaves the entry NOT_LOADED.
|
||||
|
||||
The health coordinator's first refresh requires SETUP_IN_PROGRESS and does network
|
||||
I/O, so we mock it out to keep these lifecycle tests focused on wiring.
|
||||
"""
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.HealthCoordinator.async_config_entry_first_refresh",
|
||||
AsyncMock(return_value=None),
|
||||
)
|
||||
|
||||
|
||||
# --- register_path ---------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_path_registers_routes_and_stores_dispatcher(hass_with_http):
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={
|
||||
API_ID: "id",
|
||||
API_KEY: "key",
|
||||
WSLINK: False,
|
||||
},
|
||||
options={API_ID: "id", API_KEY: "key", WSLINK: False},
|
||||
)
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
|
|
@ -89,21 +97,19 @@ async def test_register_path_registers_routes_and_stores_dispatcher(hass_with_ht
|
|||
ok = register_path(hass_with_http, coordinator, coordinator_health, entry)
|
||||
assert ok is True
|
||||
|
||||
# Router registrations
|
||||
# Router registrations: GET for legacy/wslink/health, POST for wslink + ecowitt.
|
||||
router: _RouterStub = hass_with_http.http.app.router
|
||||
assert [p for (p, _h) in router.add_get_calls] == [
|
||||
DEFAULT_URL,
|
||||
WSLINK_URL,
|
||||
HEALTH_URL,
|
||||
]
|
||||
assert [p for (p, _h) in router.add_post_calls] == [WSLINK_URL]
|
||||
assert [p for (p, _h) in router.add_post_calls] == [WSLINK_URL, ECOWITT_PATH]
|
||||
|
||||
# Dispatcher stored
|
||||
# Dispatcher stored under the shared (cross-reload) hass.data[DOMAIN].
|
||||
assert DOMAIN in hass_with_http.data
|
||||
assert "routes" in hass_with_http.data[DOMAIN]
|
||||
routes = hass_with_http.data[DOMAIN]["routes"]
|
||||
routes = hass_with_http.data[DOMAIN].get("routes")
|
||||
assert routes is not None
|
||||
# show_enabled() should return a string
|
||||
assert isinstance(routes.show_enabled(), str)
|
||||
|
||||
|
||||
|
|
@ -116,18 +122,13 @@ async def test_register_path_raises_config_entry_not_ready_on_router_runtime_err
|
|||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={
|
||||
API_ID: "id",
|
||||
API_KEY: "key",
|
||||
WSLINK: False,
|
||||
},
|
||||
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)
|
||||
|
||||
# Make router raise RuntimeError on add
|
||||
router: _RouterStub = hass_with_http.http.app.router
|
||||
router.raise_on_add = RuntimeError("router broken")
|
||||
|
||||
|
|
@ -145,26 +146,24 @@ async def test_register_path_checked_hass_data_wrong_type_raises_config_entry_no
|
|||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={
|
||||
API_ID: "id",
|
||||
API_KEY: "key",
|
||||
WSLINK: False,
|
||||
},
|
||||
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)
|
||||
|
||||
# Force wrong type under DOMAIN so `checked(..., dict)` fails.
|
||||
hass_with_http.data[DOMAIN] = []
|
||||
hass_with_http.data[DOMAIN] = [] # wrong type -> checked(..., dict) fails
|
||||
|
||||
with pytest.raises(ConfigEntryNotReady):
|
||||
register_path(hass_with_http, coordinator, coordinator_health, entry)
|
||||
|
||||
|
||||
# --- async_setup_entry -----------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_setup_entry_creates_entry_dict_and_coordinator_and_forwards_platforms(
|
||||
async def test_async_setup_entry_creates_runtime_data_and_forwards_platforms(
|
||||
hass_with_http,
|
||||
monkeypatch,
|
||||
):
|
||||
|
|
@ -175,7 +174,7 @@ async def test_async_setup_entry_creates_entry_dict_and_coordinator_and_forwards
|
|||
)
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
# Avoid loading actual platforms via HA loader.
|
||||
_mock_health_first_refresh(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
hass_with_http.config_entries,
|
||||
"async_forward_entry_setups",
|
||||
|
|
@ -185,17 +184,14 @@ async def test_async_setup_entry_creates_entry_dict_and_coordinator_and_forwards
|
|||
ok = await async_setup_entry(hass_with_http, entry)
|
||||
assert ok is True
|
||||
|
||||
# Runtime storage exists and is a dict
|
||||
assert DOMAIN in hass_with_http.data
|
||||
assert entry.entry_id in hass_with_http.data[DOMAIN]
|
||||
entry_data = hass_with_http.data[DOMAIN][entry.entry_id]
|
||||
assert isinstance(entry_data, dict)
|
||||
# Per-entry state now lives on entry.runtime_data (SWSRuntimeData).
|
||||
assert isinstance(entry.runtime_data, SWSRuntimeData)
|
||||
assert isinstance(entry.runtime_data.coordinator, WeatherDataUpdateCoordinator)
|
||||
assert isinstance(entry.runtime_data.last_options, dict)
|
||||
|
||||
# Coordinator stored and last options snapshot stored
|
||||
assert isinstance(entry_data.get(ENTRY_COORDINATOR), WeatherDataUpdateCoordinator)
|
||||
assert isinstance(entry_data.get(ENTRY_LAST_OPTIONS), dict)
|
||||
# Shared dispatcher registered under hass.data[DOMAIN].
|
||||
assert "routes" in hass_with_http.data[DOMAIN]
|
||||
|
||||
# Forwarded setups invoked
|
||||
hass_with_http.config_entries.async_forward_entry_setups.assert_awaited()
|
||||
|
||||
|
||||
|
|
@ -203,12 +199,7 @@ async def test_async_setup_entry_creates_entry_dict_and_coordinator_and_forwards
|
|||
async def test_async_setup_entry_fatal_when_register_path_returns_false(
|
||||
hass_with_http, monkeypatch
|
||||
):
|
||||
"""Cover the fatal branch when `register_path` returns False.
|
||||
|
||||
async_setup_entry does:
|
||||
routes_enabled = register_path(...)
|
||||
if not routes_enabled: raise PlatformNotReady
|
||||
"""
|
||||
"""Cover the fatal branch when `register_path` returns False -> PlatformNotReady."""
|
||||
from homeassistant.exceptions import PlatformNotReady
|
||||
|
||||
entry = MockConfigEntry(
|
||||
|
|
@ -218,17 +209,14 @@ async def test_async_setup_entry_fatal_when_register_path_returns_false(
|
|||
)
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
# Ensure there are no pre-registered routes so async_setup_entry calls register_path.
|
||||
# No pre-registered routes -> async_setup_entry calls register_path.
|
||||
hass_with_http.data.setdefault(DOMAIN, {})
|
||||
hass_with_http.data[DOMAIN].pop("routes", None)
|
||||
|
||||
# Force register_path to return False
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.register_path",
|
||||
lambda _hass, _coordinator, _coordinator_h, _entry: False,
|
||||
)
|
||||
|
||||
# Forwarding shouldn't be reached; patch anyway to avoid accidental loader calls.
|
||||
monkeypatch.setattr(
|
||||
hass_with_http.config_entries,
|
||||
"async_forward_entry_setups",
|
||||
|
|
@ -240,10 +228,11 @@ async def test_async_setup_entry_fatal_when_register_path_returns_false(
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_setup_entry_reuses_existing_coordinator_and_switches_routes(
|
||||
async def test_async_setup_entry_reuses_route_dispatcher_and_switches_protocol(
|
||||
hass_with_http,
|
||||
monkeypatch,
|
||||
):
|
||||
"""On reload the shared route dispatcher is reused; the coordinator is recreated."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
|
|
@ -251,29 +240,19 @@ async def test_async_setup_entry_reuses_existing_coordinator_and_switches_routes
|
|||
)
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
# Pretend setup already happened and a coordinator exists
|
||||
hass_with_http.data.setdefault(DOMAIN, {})
|
||||
existing_coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry)
|
||||
hass_with_http.data[DOMAIN][entry.entry_id] = {
|
||||
ENTRY_COORDINATOR: existing_coordinator,
|
||||
ENTRY_LAST_OPTIONS: dict(entry.options),
|
||||
}
|
||||
# Pre-register routes once (legacy/WU active).
|
||||
initial_coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry)
|
||||
initial_health = HealthCoordinator(hass_with_http, entry)
|
||||
register_path(hass_with_http, initial_coordinator, initial_health, entry)
|
||||
routes_before = hass_with_http.data[DOMAIN]["routes"]
|
||||
assert routes_before.path_enabled(DEFAULT_URL) is True
|
||||
|
||||
# Provide pre-registered routes dispatcher
|
||||
routes = hass_with_http.data[DOMAIN].get("routes")
|
||||
if routes is None:
|
||||
# Create a dispatcher via register_path once
|
||||
coordinator_health = HealthCoordinator(hass_with_http, entry)
|
||||
register_path(hass_with_http, existing_coordinator, coordinator_health, entry)
|
||||
routes = hass_with_http.data[DOMAIN]["routes"]
|
||||
|
||||
# Turn on WSLINK to trigger dispatcher switching.
|
||||
# ConfigEntry.options cannot be changed directly; use async_update_entry.
|
||||
# Switch to WSLink and run setup again.
|
||||
hass_with_http.config_entries.async_update_entry(
|
||||
entry, options={**dict(entry.options), WSLINK: True}
|
||||
)
|
||||
|
||||
# Avoid loading actual platforms via HA loader.
|
||||
_mock_health_first_refresh(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
hass_with_http.config_entries,
|
||||
"async_forward_entry_setups",
|
||||
|
|
@ -283,34 +262,42 @@ async def test_async_setup_entry_reuses_existing_coordinator_and_switches_routes
|
|||
ok = await async_setup_entry(hass_with_http, entry)
|
||||
assert ok is True
|
||||
|
||||
# Coordinator reused (same object)
|
||||
entry_data = hass_with_http.data[DOMAIN][entry.entry_id]
|
||||
assert entry_data[ENTRY_COORDINATOR] is existing_coordinator
|
||||
# Same dispatcher object reused (survives across reloads).
|
||||
assert hass_with_http.data[DOMAIN]["routes"] is routes_before
|
||||
# Protocol switched to WSLink.
|
||||
assert routes_before.path_enabled(WSLINK_URL) is True
|
||||
assert routes_before.path_enabled(DEFAULT_URL) is False
|
||||
# A fresh coordinator is wired onto entry.runtime_data.
|
||||
assert isinstance(entry.runtime_data, SWSRuntimeData)
|
||||
assert isinstance(entry.runtime_data.coordinator, WeatherDataUpdateCoordinator)
|
||||
|
||||
|
||||
# --- update_listener -------------------------------------------------------
|
||||
|
||||
|
||||
def _entry_with_runtime(hass, *, options: dict[str, Any]) -> MockConfigEntry:
|
||||
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
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_listener_skips_reload_when_only_sensors_to_load_changes(
|
||||
hass_with_http,
|
||||
):
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
entry = _entry_with_runtime(
|
||||
hass_with_http,
|
||||
options={API_ID: "id", API_KEY: "key", SENSORS_TO_LOAD: ["a"]},
|
||||
)
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
# Seed hass.data snapshot
|
||||
hass_with_http.data.setdefault(DOMAIN, {})
|
||||
hass_with_http.data[DOMAIN][entry.entry_id] = {
|
||||
# Seed the full old options snapshot. If we only store SENSORS_TO_LOAD here,
|
||||
# update_listener will detect differences for other keys (e.g. auth keys) and reload.
|
||||
ENTRY_LAST_OPTIONS: dict(entry.options),
|
||||
}
|
||||
|
||||
hass_with_http.config_entries.async_reload = AsyncMock()
|
||||
|
||||
# Only SENSORS_TO_LOAD changes.
|
||||
# ConfigEntry.options cannot be changed directly; use async_update_entry.
|
||||
hass_with_http.config_entries.async_update_entry(
|
||||
entry, options={**dict(entry.options), SENSORS_TO_LOAD: ["a", "b"]}
|
||||
)
|
||||
|
|
@ -318,9 +305,8 @@ async def test_update_listener_skips_reload_when_only_sensors_to_load_changes(
|
|||
await update_listener(hass_with_http, entry)
|
||||
|
||||
hass_with_http.config_entries.async_reload.assert_not_awaited()
|
||||
# Snapshot should be updated
|
||||
entry_data = hass_with_http.data[DOMAIN][entry.entry_id]
|
||||
assert entry_data[ENTRY_LAST_OPTIONS] == dict(entry.options)
|
||||
# The snapshot on runtime_data is refreshed.
|
||||
assert entry.runtime_data.last_options == dict(entry.options)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -328,22 +314,13 @@ async def test_update_listener_triggers_reload_when_other_option_changes(
|
|||
hass_with_http,
|
||||
monkeypatch,
|
||||
):
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
entry = _entry_with_runtime(
|
||||
hass_with_http,
|
||||
options={API_ID: "id", API_KEY: "key", SENSORS_TO_LOAD: ["a"], WSLINK: False},
|
||||
)
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
hass_with_http.data.setdefault(DOMAIN, {})
|
||||
hass_with_http.data[DOMAIN][entry.entry_id] = {
|
||||
ENTRY_LAST_OPTIONS: dict(entry.options),
|
||||
}
|
||||
|
||||
hass_with_http.config_entries.async_reload = AsyncMock(return_value=True)
|
||||
|
||||
# Change a different option.
|
||||
# ConfigEntry.options cannot be changed directly; use async_update_entry.
|
||||
hass_with_http.config_entries.async_update_entry(
|
||||
entry, options={**dict(entry.options), WSLINK: True}
|
||||
)
|
||||
|
|
@ -358,76 +335,58 @@ async def test_update_listener_triggers_reload_when_other_option_changes(
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_listener_missing_snapshot_stores_current_options_then_reloads(
|
||||
hass_with_http,
|
||||
):
|
||||
"""Cover update_listener branch where the options snapshot is missing/invalid.
|
||||
|
||||
This hits:
|
||||
entry_data[ENTRY_LAST_OPTIONS] = dict(entry.options)
|
||||
and then proceeds to reload.
|
||||
"""
|
||||
async def test_update_listener_without_runtime_snapshot_reloads(hass_with_http):
|
||||
"""When runtime_data is not a valid snapshot, update_listener reloads."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={API_ID: "id", API_KEY: "key", SENSORS_TO_LOAD: ["a"], WSLINK: False},
|
||||
)
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
hass_with_http.data.setdefault(DOMAIN, {})
|
||||
# Store an invalid snapshot type to force the "No/invalid snapshot" branch.
|
||||
hass_with_http.data[DOMAIN][entry.entry_id] = {ENTRY_LAST_OPTIONS: "invalid"}
|
||||
# Not an SWSRuntimeData instance -> the skip-reload fast path is bypassed.
|
||||
entry.runtime_data = "invalid"
|
||||
|
||||
hass_with_http.config_entries.async_reload = AsyncMock(return_value=True)
|
||||
|
||||
await update_listener(hass_with_http, entry)
|
||||
|
||||
entry_data = hass_with_http.data[DOMAIN][entry.entry_id]
|
||||
assert entry_data[ENTRY_LAST_OPTIONS] == dict(entry.options)
|
||||
hass_with_http.config_entries.async_reload.assert_awaited_once_with(entry.entry_id)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_unload_entry_pops_runtime_data_on_success(hass_with_http):
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={API_ID: "id", API_KEY: "key"},
|
||||
)
|
||||
entry.add_to_hass(hass_with_http)
|
||||
# --- async_unload_entry ----------------------------------------------------
|
||||
|
||||
hass_with_http.data.setdefault(DOMAIN, {})
|
||||
hass_with_http.data[DOMAIN][entry.entry_id] = {ENTRY_COORDINATOR: object()}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_unload_entry_returns_true_on_success(hass_with_http):
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={}, options={API_ID: "id", API_KEY: "key"})
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
hass_with_http.config_entries.async_unload_platforms = AsyncMock(return_value=True)
|
||||
|
||||
ok = await async_unload_entry(hass_with_http, entry)
|
||||
|
||||
assert ok is True
|
||||
assert entry.entry_id not in hass_with_http.data[DOMAIN]
|
||||
hass_with_http.config_entries.async_unload_platforms.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_unload_entry_keeps_runtime_data_on_failure(hass_with_http):
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
options={API_ID: "id", API_KEY: "key"},
|
||||
)
|
||||
async def test_async_unload_entry_returns_false_on_failure(hass_with_http):
|
||||
entry = MockConfigEntry(domain=DOMAIN, data={}, options={API_ID: "id", API_KEY: "key"})
|
||||
entry.add_to_hass(hass_with_http)
|
||||
|
||||
hass_with_http.data.setdefault(DOMAIN, {})
|
||||
hass_with_http.data[DOMAIN][entry.entry_id] = {ENTRY_COORDINATOR: object()}
|
||||
|
||||
hass_with_http.config_entries.async_unload_platforms = AsyncMock(return_value=False)
|
||||
|
||||
ok = await async_unload_entry(hass_with_http, entry)
|
||||
|
||||
assert ok is False
|
||||
assert entry.entry_id in hass_with_http.data[DOMAIN]
|
||||
|
||||
|
||||
# --- coordinator auth (lifecycle-adjacent) ---------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_received_data_auth_unauthorized_and_incorrect_data_paths(hass):
|
||||
"""A few lifecycle-adjacent assertions to cover coordinator auth behavior in __init__.py."""
|
||||
"""Cover coordinator auth behavior reachable from the webhook entrypoint."""
|
||||
entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={},
|
||||
|
|
@ -447,9 +406,7 @@ async def test_received_data_auth_unauthorized_and_incorrect_data_paths(hass):
|
|||
) # type: ignore[arg-type]
|
||||
|
||||
# Missing API_ID in options -> IncorrectDataError
|
||||
entry2 = MockConfigEntry(
|
||||
domain=DOMAIN, data={}, options={API_KEY: "key", WSLINK: False}
|
||||
)
|
||||
entry2 = MockConfigEntry(domain=DOMAIN, data={}, options={API_KEY: "key", WSLINK: False})
|
||||
entry2.add_to_hass(hass)
|
||||
coordinator2 = WeatherDataUpdateCoordinator(hass, entry2)
|
||||
with pytest.raises(IncorrectDataError):
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ from unittest.mock import AsyncMock, MagicMock
|
|||
from aiohttp.web_exceptions import HTTPUnauthorized
|
||||
import pytest
|
||||
|
||||
from custom_components.sws12500 import IncorrectDataError, WeatherDataUpdateCoordinator
|
||||
from custom_components.sws12500.const import (
|
||||
API_ID,
|
||||
API_KEY,
|
||||
|
|
@ -20,6 +19,8 @@ from custom_components.sws12500.const import (
|
|||
WSLINK,
|
||||
WSLINK_URL,
|
||||
)
|
||||
from custom_components.sws12500.coordinator import IncorrectDataError, WeatherDataUpdateCoordinator
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
|
|
@ -60,6 +61,18 @@ def _make_entry(
|
|||
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.
|
||||
# received_data writes last_seen and reads health_coordinator / add_*_entities.
|
||||
entry.runtime_data = SimpleNamespace(
|
||||
health_coordinator=None,
|
||||
add_sensor_entities=None,
|
||||
add_binary_entities=None,
|
||||
last_seen={},
|
||||
started_at=dt_util.utcnow(),
|
||||
)
|
||||
return entry
|
||||
|
||||
|
||||
|
|
@ -135,13 +148,13 @@ async def test_received_data_success_remaps_and_updates_coordinator_data(
|
|||
# Patch remapping so this test doesn't depend on mapping tables.
|
||||
remapped = {"outside_temp": "10"}
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.remap_items",
|
||||
"custom_components.sws12500.coordinator.remap_items",
|
||||
lambda _data: remapped,
|
||||
)
|
||||
|
||||
# Ensure no autodiscovery triggers
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: [],
|
||||
)
|
||||
|
||||
|
|
@ -162,12 +175,12 @@ async def test_received_data_success_wslink_uses_wslink_remap(hass, monkeypatch)
|
|||
|
||||
remapped = {"ws_temp": "1"}
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.remap_wslink_items",
|
||||
"custom_components.sws12500.coordinator.remap_wslink_items",
|
||||
lambda _data: remapped,
|
||||
)
|
||||
# If the wrong remapper is used, we'd crash because we won't patch it:
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: [],
|
||||
)
|
||||
|
||||
|
|
@ -188,11 +201,11 @@ async def test_received_data_forwards_to_windy_when_enabled(hass, monkeypatch):
|
|||
coordinator.windy.push_data_to_windy = AsyncMock()
|
||||
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.remap_items",
|
||||
"custom_components.sws12500.coordinator.remap_items",
|
||||
lambda _data: {"k": "v"},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: [],
|
||||
)
|
||||
|
||||
|
|
@ -216,11 +229,11 @@ async def test_received_data_forwards_to_pocasi_when_enabled(hass, monkeypatch):
|
|||
coordinator.pocasi.push_data_to_server = AsyncMock()
|
||||
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.remap_wslink_items",
|
||||
"custom_components.sws12500.coordinator.remap_wslink_items",
|
||||
lambda _data: {"k": "v"},
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: [],
|
||||
)
|
||||
|
||||
|
|
@ -246,35 +259,35 @@ async def test_received_data_autodiscovery_updates_options_notifies_and_adds_sen
|
|||
|
||||
# Arrange: remapped payload contains keys that are disabled.
|
||||
remapped = {"a": "1", "b": "2"}
|
||||
monkeypatch.setattr("custom_components.sws12500.remap_items", lambda _d: remapped)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.remap_items", lambda _d: remapped)
|
||||
|
||||
# Autodiscovery finds two sensors to add
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: ["a", "b"],
|
||||
)
|
||||
|
||||
# No previously loaded sensors
|
||||
monkeypatch.setattr("custom_components.sws12500.loaded_sensors", lambda _c: [])
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.loaded_sensors", lambda _c: [])
|
||||
|
||||
# translations returns a friendly name for each sensor key
|
||||
async def _translations(_hass, _domain, _key, **_kwargs):
|
||||
# return something non-None so it's included in human readable string
|
||||
return "Name"
|
||||
|
||||
monkeypatch.setattr("custom_components.sws12500.translations", _translations)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.translations", _translations)
|
||||
|
||||
translated_notification = AsyncMock()
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.translated_notification", translated_notification
|
||||
"custom_components.sws12500.coordinator.translated_notification", translated_notification
|
||||
)
|
||||
|
||||
update_options = AsyncMock()
|
||||
monkeypatch.setattr("custom_components.sws12500.update_options", update_options)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.update_options", update_options)
|
||||
|
||||
add_new_sensors = MagicMock()
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.sensor.add_new_sensors", add_new_sensors
|
||||
"custom_components.sws12500.coordinator.add_new_sensors", add_new_sensors
|
||||
)
|
||||
|
||||
coordinator.async_set_updated_data = MagicMock()
|
||||
|
|
@ -313,19 +326,19 @@ async def test_received_data_autodiscovery_human_readable_empty_branch_via_check
|
|||
coordinator = WeatherDataUpdateCoordinator(hass, entry)
|
||||
|
||||
remapped = {"a": "1"}
|
||||
monkeypatch.setattr("custom_components.sws12500.remap_items", lambda _d: remapped)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.remap_items", lambda _d: remapped)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: ["a"],
|
||||
)
|
||||
monkeypatch.setattr("custom_components.sws12500.loaded_sensors", lambda _c: [])
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.loaded_sensors", lambda _c: [])
|
||||
|
||||
# Return a translation so the list comprehension would normally include an item.
|
||||
async def _translations(_hass, _domain, _key, **_kwargs):
|
||||
return "Name"
|
||||
|
||||
monkeypatch.setattr("custom_components.sws12500.translations", _translations)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.translations", _translations)
|
||||
|
||||
# Force checked(...) to return None when the code tries to validate translate_sensors as list[str].
|
||||
def _checked_override(value, expected_type):
|
||||
|
|
@ -333,19 +346,19 @@ async def test_received_data_autodiscovery_human_readable_empty_branch_via_check
|
|||
return None
|
||||
return value
|
||||
|
||||
monkeypatch.setattr("custom_components.sws12500.checked", _checked_override)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.checked", _checked_override)
|
||||
|
||||
translated_notification = AsyncMock()
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.translated_notification", translated_notification
|
||||
"custom_components.sws12500.coordinator.translated_notification", translated_notification
|
||||
)
|
||||
|
||||
update_options = AsyncMock()
|
||||
monkeypatch.setattr("custom_components.sws12500.update_options", update_options)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.update_options", update_options)
|
||||
|
||||
add_new_sensors = MagicMock()
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.sensor.add_new_sensors", add_new_sensors
|
||||
"custom_components.sws12500.coordinator.add_new_sensors", add_new_sensors
|
||||
)
|
||||
|
||||
coordinator.async_set_updated_data = MagicMock()
|
||||
|
|
@ -371,33 +384,33 @@ async def test_received_data_autodiscovery_extends_with_loaded_sensors_branch(
|
|||
coordinator = WeatherDataUpdateCoordinator(hass, entry)
|
||||
|
||||
remapped = {"new": "1"}
|
||||
monkeypatch.setattr("custom_components.sws12500.remap_items", lambda _d: remapped)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.remap_items", lambda _d: remapped)
|
||||
|
||||
# Autodiscovery finds one new sensor
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: ["new"],
|
||||
)
|
||||
|
||||
# Pretend there are already loaded sensors in options
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.loaded_sensors", lambda _c: ["existing"]
|
||||
"custom_components.sws12500.coordinator.loaded_sensors", lambda _c: ["existing"]
|
||||
)
|
||||
|
||||
async def _translations(_hass, _domain, _key, **_kwargs):
|
||||
return "Name"
|
||||
|
||||
monkeypatch.setattr("custom_components.sws12500.translations", _translations)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.translations", _translations)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.translated_notification", AsyncMock()
|
||||
"custom_components.sws12500.coordinator.translated_notification", AsyncMock()
|
||||
)
|
||||
|
||||
update_options = AsyncMock()
|
||||
monkeypatch.setattr("custom_components.sws12500.update_options", update_options)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.update_options", update_options)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.sensor.add_new_sensors", MagicMock()
|
||||
"custom_components.sws12500.coordinator.add_new_sensors", MagicMock()
|
||||
)
|
||||
|
||||
coordinator.async_set_updated_data = MagicMock()
|
||||
|
|
@ -424,31 +437,31 @@ async def test_received_data_autodiscovery_translations_all_none_still_notifies_
|
|||
coordinator = WeatherDataUpdateCoordinator(hass, entry)
|
||||
|
||||
remapped = {"a": "1"}
|
||||
monkeypatch.setattr("custom_components.sws12500.remap_items", lambda _d: remapped)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.remap_items", lambda _d: remapped)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: ["a"],
|
||||
)
|
||||
monkeypatch.setattr("custom_components.sws12500.loaded_sensors", lambda _c: [])
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.loaded_sensors", lambda _c: [])
|
||||
|
||||
# Force translations to return None for every lookup -> translate_sensors becomes None and human_readable ""
|
||||
async def _translations(_hass, _domain, _key, **_kwargs):
|
||||
return None
|
||||
|
||||
monkeypatch.setattr("custom_components.sws12500.translations", _translations)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.translations", _translations)
|
||||
|
||||
translated_notification = AsyncMock()
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.translated_notification", translated_notification
|
||||
"custom_components.sws12500.coordinator.translated_notification", translated_notification
|
||||
)
|
||||
|
||||
update_options = AsyncMock()
|
||||
monkeypatch.setattr("custom_components.sws12500.update_options", update_options)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.update_options", update_options)
|
||||
|
||||
add_new_sensors = MagicMock()
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.sensor.add_new_sensors", add_new_sensors
|
||||
"custom_components.sws12500.coordinator.add_new_sensors", add_new_sensors
|
||||
)
|
||||
|
||||
coordinator.async_set_updated_data = MagicMock()
|
||||
|
|
@ -469,17 +482,17 @@ async def test_received_data_dev_logging_calls_anonymize_and_logs(hass, monkeypa
|
|||
entry = _make_entry(wslink=False, api_id="id", api_key="key", dev_debug=True)
|
||||
coordinator = WeatherDataUpdateCoordinator(hass, entry)
|
||||
|
||||
monkeypatch.setattr("custom_components.sws12500.remap_items", lambda _d: {"k": "v"})
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.remap_items", lambda _d: {"k": "v"})
|
||||
monkeypatch.setattr(
|
||||
"custom_components.sws12500.check_disabled",
|
||||
"custom_components.sws12500.coordinator.check_disabled",
|
||||
lambda _remaped_items, _config: [],
|
||||
)
|
||||
|
||||
anonymize = MagicMock(return_value={"safe": True})
|
||||
monkeypatch.setattr("custom_components.sws12500.anonymize", anonymize)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator.anonymize", anonymize)
|
||||
|
||||
log_info = MagicMock()
|
||||
monkeypatch.setattr("custom_components.sws12500._LOGGER.info", log_info)
|
||||
monkeypatch.setattr("custom_components.sws12500.coordinator._LOGGER.info", log_info)
|
||||
|
||||
coordinator.async_set_updated_data = MagicMock()
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
|
|
@ -17,11 +17,7 @@ from custom_components.sws12500.const import (
|
|||
WIND_SPEED,
|
||||
WSLINK,
|
||||
)
|
||||
from custom_components.sws12500.data import (
|
||||
ENTRY_ADD_ENTITIES,
|
||||
ENTRY_COORDINATOR,
|
||||
ENTRY_DESCRIPTIONS,
|
||||
)
|
||||
from custom_components.sws12500.data import SWSRuntimeData
|
||||
from custom_components.sws12500.sensor import (
|
||||
WeatherSensor,
|
||||
_auto_enable_derived_sensors,
|
||||
|
|
@ -32,25 +28,55 @@ from custom_components.sws12500.sensors_weather import SENSOR_TYPES_WEATHER_API
|
|||
from custom_components.sws12500.sensors_wslink import SENSOR_TYPES_WSLINK
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class _ConfigEntryStub:
|
||||
entry_id: str
|
||||
options: dict[str, Any]
|
||||
class _EcowittBridgeStub:
|
||||
"""Records the platform callback the sensor setup wires into the bridge."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.add_entities_cb: Any = None
|
||||
|
||||
def set_add_entities(self, callback: Any) -> None:
|
||||
self.add_entities_cb = callback
|
||||
|
||||
|
||||
class _CoordinatorStub:
|
||||
"""Minimal coordinator stub for WeatherSensor and platform setup."""
|
||||
|
||||
def __init__(
|
||||
self, data: dict[str, Any] | None = None, *, config: Any | None = None
|
||||
) -> None:
|
||||
def __init__(self, data: dict[str, Any] | None = None, *, options: dict[str, Any] | None = None) -> None:
|
||||
self.data = data if data is not None else {}
|
||||
self.config = config
|
||||
# WeatherSensor.__init__ reads coordinator.config.options for the dev-log flag.
|
||||
self.config = SimpleNamespace(options=options if options is not None else {})
|
||||
self.ecowitt_bridge = _EcowittBridgeStub()
|
||||
|
||||
|
||||
class _HealthCoordinatorStub:
|
||||
"""Stand-in for HealthCoordinator (health diagnostic sensors subscribe to it)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.data: dict[str, Any] = {}
|
||||
|
||||
|
||||
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=_HealthCoordinatorStub(), # 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():
|
||||
# Use a very small hass-like object; sensor platform uses only `hass.data`.
|
||||
# Sensor platform setup only forwards hass to health_sensor.async_setup_entry,
|
||||
# which ignores it, and add_new_sensors deletes it. A tiny stub is enough.
|
||||
class _Hass:
|
||||
def __init__(self) -> None:
|
||||
self.data: dict[str, Any] = {}
|
||||
|
|
@ -58,11 +84,6 @@ def hass():
|
|||
return _Hass()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_entry() -> _ConfigEntryStub:
|
||||
return _ConfigEntryStub(entry_id="test_entry_id", options={})
|
||||
|
||||
|
||||
def _capture_add_entities():
|
||||
captured: list[Any] = []
|
||||
|
||||
|
|
@ -72,207 +93,118 @@ def _capture_add_entities():
|
|||
return captured, _add_entities
|
||||
|
||||
|
||||
def _weather_keys(captured: list[Any]) -> set[str]:
|
||||
return {e.entity_description.key for e in captured if isinstance(e, WeatherSensor)}
|
||||
|
||||
|
||||
# --- _auto_enable_derived_sensors ------------------------------------------
|
||||
|
||||
|
||||
def test_auto_enable_derived_sensors_wind_azimut():
|
||||
requested = {WIND_DIR}
|
||||
expanded = _auto_enable_derived_sensors(requested)
|
||||
expanded = _auto_enable_derived_sensors({WIND_DIR})
|
||||
assert WIND_DIR in expanded
|
||||
assert WIND_AZIMUT in expanded
|
||||
|
||||
|
||||
def test_auto_enable_derived_sensors_heat_index():
|
||||
requested = {OUTSIDE_TEMP, OUTSIDE_HUMIDITY}
|
||||
expanded = _auto_enable_derived_sensors(requested)
|
||||
expanded = _auto_enable_derived_sensors({OUTSIDE_TEMP, OUTSIDE_HUMIDITY})
|
||||
assert HEAT_INDEX in expanded
|
||||
|
||||
|
||||
def test_auto_enable_derived_sensors_chill_index():
|
||||
requested = {OUTSIDE_TEMP, WIND_SPEED}
|
||||
expanded = _auto_enable_derived_sensors(requested)
|
||||
expanded = _auto_enable_derived_sensors({OUTSIDE_TEMP, WIND_SPEED})
|
||||
assert CHILL_INDEX in expanded
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sensor_async_setup_entry_no_coordinator_is_noop(hass, config_entry):
|
||||
# No entry dict created by integration yet; async_setup_entry should be defensive and no-op.
|
||||
captured, add_entities = _capture_add_entities()
|
||||
|
||||
await async_setup_entry(hass, config_entry, add_entities)
|
||||
|
||||
assert captured == []
|
||||
# --- async_setup_entry -----------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sensor_async_setup_entry_stores_callback_and_descriptions_even_if_no_sensors_to_load(
|
||||
hass, config_entry
|
||||
):
|
||||
# Prepare runtime entry data and coordinator like integration does.
|
||||
hass.data.setdefault("sws12500", {})
|
||||
hass.data["sws12500"][config_entry.entry_id] = {
|
||||
ENTRY_COORDINATOR: _CoordinatorStub()
|
||||
}
|
||||
|
||||
async def test_setup_stores_callback_and_descriptions_even_without_sensors_to_load(hass):
|
||||
entry, coordinator, runtime = _make_entry()
|
||||
captured, add_entities = _capture_add_entities()
|
||||
|
||||
# No SENSORS_TO_LOAD set -> early return, but it should still store callback + descriptions.
|
||||
await async_setup_entry(hass, config_entry, add_entities)
|
||||
await async_setup_entry(hass, entry, add_entities)
|
||||
|
||||
entry_data = hass.data["sws12500"][config_entry.entry_id]
|
||||
assert entry_data[ENTRY_ADD_ENTITIES] is add_entities
|
||||
assert isinstance(entry_data[ENTRY_DESCRIPTIONS], dict)
|
||||
assert captured == []
|
||||
# Callback + description map persisted for dynamic entity creation.
|
||||
assert runtime.add_sensor_entities is add_entities
|
||||
assert isinstance(runtime.sensor_descriptions, dict)
|
||||
# Ecowitt bridge wired up even though there are no sensors to load yet.
|
||||
assert coordinator.ecowitt_bridge.add_entities_cb is add_entities
|
||||
# No weather sensors created (only health diagnostics, which we ignore here).
|
||||
assert _weather_keys(captured) == set()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sensor_async_setup_entry_selects_weather_api_descriptions_when_wslink_disabled(
|
||||
hass, config_entry
|
||||
):
|
||||
hass.data.setdefault("sws12500", {})
|
||||
hass.data["sws12500"][config_entry.entry_id] = {
|
||||
ENTRY_COORDINATOR: _CoordinatorStub()
|
||||
}
|
||||
async def test_setup_selects_weather_api_descriptions_when_wslink_disabled(hass):
|
||||
entry, _coordinator, runtime = _make_entry(options={WSLINK: False})
|
||||
_captured, add_entities = _capture_add_entities()
|
||||
|
||||
captured, add_entities = _capture_add_entities()
|
||||
await async_setup_entry(hass, entry, add_entities)
|
||||
|
||||
# Explicitly disabled WSLINK
|
||||
config_entry.options[WSLINK] = False
|
||||
|
||||
await async_setup_entry(hass, config_entry, add_entities)
|
||||
|
||||
descriptions = hass.data["sws12500"][config_entry.entry_id][ENTRY_DESCRIPTIONS]
|
||||
assert set(descriptions.keys()) == {d.key for d in SENSOR_TYPES_WEATHER_API}
|
||||
assert captured == []
|
||||
assert set(runtime.sensor_descriptions.keys()) == {d.key for d in SENSOR_TYPES_WEATHER_API}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sensor_async_setup_entry_selects_wslink_descriptions_when_wslink_enabled(
|
||||
hass, config_entry
|
||||
):
|
||||
hass.data.setdefault("sws12500", {})
|
||||
hass.data["sws12500"][config_entry.entry_id] = {
|
||||
ENTRY_COORDINATOR: _CoordinatorStub()
|
||||
}
|
||||
async def test_setup_selects_wslink_descriptions_when_wslink_enabled(hass):
|
||||
entry, _coordinator, runtime = _make_entry(options={WSLINK: True})
|
||||
_captured, add_entities = _capture_add_entities()
|
||||
|
||||
captured, add_entities = _capture_add_entities()
|
||||
await async_setup_entry(hass, entry, add_entities)
|
||||
|
||||
config_entry.options[WSLINK] = True
|
||||
|
||||
await async_setup_entry(hass, config_entry, add_entities)
|
||||
|
||||
descriptions = hass.data["sws12500"][config_entry.entry_id][ENTRY_DESCRIPTIONS]
|
||||
assert set(descriptions.keys()) == {d.key for d in SENSOR_TYPES_WSLINK}
|
||||
assert captured == []
|
||||
assert set(runtime.sensor_descriptions.keys()) == {d.key for d in SENSOR_TYPES_WSLINK}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sensor_async_setup_entry_adds_requested_entities_and_auto_enables_derived(
|
||||
hass, config_entry
|
||||
):
|
||||
hass.data.setdefault("sws12500", {})
|
||||
coordinator = _CoordinatorStub()
|
||||
hass.data["sws12500"][config_entry.entry_id] = {ENTRY_COORDINATOR: coordinator}
|
||||
|
||||
captured, add_entities = _capture_add_entities()
|
||||
|
||||
# Request WIND_DIR, OUTSIDE_TEMP, OUTSIDE_HUMIDITY, WIND_SPEED -> should auto-add derived keys too.
|
||||
config_entry.options[WSLINK] = False
|
||||
config_entry.options[SENSORS_TO_LOAD] = [
|
||||
WIND_DIR,
|
||||
OUTSIDE_TEMP,
|
||||
OUTSIDE_HUMIDITY,
|
||||
WIND_SPEED,
|
||||
]
|
||||
|
||||
await async_setup_entry(hass, config_entry, add_entities)
|
||||
|
||||
# We should have at least those requested + derived in the added entities.
|
||||
keys_added = {
|
||||
e.entity_description.key for e in captured if isinstance(e, WeatherSensor)
|
||||
}
|
||||
assert WIND_DIR in keys_added
|
||||
assert OUTSIDE_TEMP in keys_added
|
||||
assert OUTSIDE_HUMIDITY in keys_added
|
||||
assert WIND_SPEED in keys_added
|
||||
|
||||
# Derived:
|
||||
assert WIND_AZIMUT in keys_added
|
||||
assert HEAT_INDEX in keys_added
|
||||
assert CHILL_INDEX in keys_added
|
||||
|
||||
|
||||
def test_add_new_sensors_is_noop_when_domain_missing(hass, config_entry):
|
||||
called = False
|
||||
|
||||
def add_entities(_entities: list[Any]) -> None:
|
||||
nonlocal called
|
||||
called = True
|
||||
|
||||
# No hass.data["sws12500"] at all.
|
||||
add_new_sensors(hass, config_entry, keys=["anything"])
|
||||
|
||||
assert called is False
|
||||
|
||||
|
||||
def test_add_new_sensors_is_noop_when_entry_missing(hass, config_entry):
|
||||
hass.data["sws12500"] = {}
|
||||
called = False
|
||||
|
||||
def add_entities(_entities: list[Any]) -> None:
|
||||
nonlocal called
|
||||
called = True
|
||||
|
||||
add_new_sensors(hass, config_entry, keys=["anything"])
|
||||
|
||||
assert called is False
|
||||
|
||||
|
||||
def test_add_new_sensors_is_noop_when_callback_or_descriptions_missing(
|
||||
hass, config_entry
|
||||
):
|
||||
hass.data["sws12500"] = {
|
||||
config_entry.entry_id: {ENTRY_COORDINATOR: _CoordinatorStub()}
|
||||
}
|
||||
called = False
|
||||
|
||||
def add_entities(_entities: list[Any]) -> None:
|
||||
nonlocal called
|
||||
called = True
|
||||
|
||||
# Missing ENTRY_ADD_ENTITIES + ENTRY_DESCRIPTIONS -> no-op.
|
||||
add_new_sensors(hass, config_entry, keys=["anything"])
|
||||
|
||||
assert called is False
|
||||
|
||||
|
||||
def test_add_new_sensors_ignores_unknown_keys(hass, config_entry):
|
||||
hass.data["sws12500"] = {
|
||||
config_entry.entry_id: {
|
||||
ENTRY_COORDINATOR: _CoordinatorStub(),
|
||||
ENTRY_ADD_ENTITIES: MagicMock(),
|
||||
ENTRY_DESCRIPTIONS: {}, # nothing known
|
||||
async def test_setup_adds_requested_entities_and_auto_enables_derived(hass):
|
||||
entry, _coordinator, _runtime = _make_entry(
|
||||
options={
|
||||
WSLINK: False,
|
||||
SENSORS_TO_LOAD: [WIND_DIR, OUTSIDE_TEMP, OUTSIDE_HUMIDITY, WIND_SPEED],
|
||||
}
|
||||
}
|
||||
)
|
||||
captured, add_entities = _capture_add_entities()
|
||||
|
||||
add_new_sensors(hass, config_entry, keys=["unknown_key"])
|
||||
await async_setup_entry(hass, entry, add_entities)
|
||||
|
||||
hass.data["sws12500"][config_entry.entry_id][ENTRY_ADD_ENTITIES].assert_not_called()
|
||||
keys_added = _weather_keys(captured)
|
||||
# Requested.
|
||||
assert {WIND_DIR, OUTSIDE_TEMP, OUTSIDE_HUMIDITY, WIND_SPEED} <= keys_added
|
||||
# Derived.
|
||||
assert {WIND_AZIMUT, HEAT_INDEX, CHILL_INDEX} <= keys_added
|
||||
|
||||
|
||||
def test_add_new_sensors_adds_known_keys(hass, config_entry):
|
||||
coordinator = _CoordinatorStub()
|
||||
# --- add_new_sensors -------------------------------------------------------
|
||||
|
||||
|
||||
def test_add_new_sensors_is_noop_when_callback_missing(hass):
|
||||
entry, _coordinator, runtime = _make_entry()
|
||||
# Platform not set up yet -> no stored callback.
|
||||
assert runtime.add_sensor_entities is None
|
||||
|
||||
# Must not raise.
|
||||
add_new_sensors(hass, entry, keys=["anything"])
|
||||
|
||||
|
||||
def test_add_new_sensors_ignores_unknown_keys(hass):
|
||||
entry, _coordinator, runtime = _make_entry()
|
||||
add_entities = MagicMock()
|
||||
runtime.add_sensor_entities = add_entities
|
||||
runtime.sensor_descriptions = {} # nothing known
|
||||
|
||||
# Use one known description from the weather API list.
|
||||
add_new_sensors(hass, entry, keys=["unknown_key"])
|
||||
|
||||
add_entities.assert_not_called()
|
||||
|
||||
|
||||
def test_add_new_sensors_adds_known_keys(hass):
|
||||
entry, _coordinator, runtime = _make_entry()
|
||||
add_entities = MagicMock()
|
||||
known_desc = SENSOR_TYPES_WEATHER_API[0]
|
||||
runtime.add_sensor_entities = add_entities
|
||||
runtime.sensor_descriptions = {known_desc.key: known_desc}
|
||||
|
||||
hass.data["sws12500"] = {
|
||||
config_entry.entry_id: {
|
||||
ENTRY_COORDINATOR: coordinator,
|
||||
ENTRY_ADD_ENTITIES: add_entities,
|
||||
ENTRY_DESCRIPTIONS: {known_desc.key: known_desc},
|
||||
}
|
||||
}
|
||||
|
||||
add_new_sensors(hass, config_entry, keys=[known_desc.key])
|
||||
add_new_sensors(hass, entry, keys=[known_desc.key])
|
||||
|
||||
add_entities.assert_called_once()
|
||||
(entities_arg,) = add_entities.call_args.args
|
||||
|
|
|
|||
|
|
@ -70,11 +70,13 @@ def test_t9_keys_are_remapped() -> None:
|
|||
|
||||
|
||||
def test_connection_gated_sensors_definition() -> None:
|
||||
assert CONNECTION_GATED_SENSORS == {"t9cn": [HCHO, VOC, T9_BATTERY]}
|
||||
# The T9 HCHO/VOC probe is gated by its own connection flag. (Multi-channel
|
||||
# CH2-CH8 probes have their own gates too; we only assert the T9 one here.)
|
||||
assert CONNECTION_GATED_SENSORS["t9cn"] == [HCHO, VOC, T9_BATTERY]
|
||||
|
||||
|
||||
def test_t9_battery_is_non_binary_only() -> None:
|
||||
assert BATTERY_NON_BINARY == [T9_BATTERY]
|
||||
assert BATTERY_NON_BINARY == (T9_BATTERY,)
|
||||
# the 0-5 / percentage battery must not be treated as a binary low/normal one
|
||||
assert T9_BATTERY not in BATTERY_LIST
|
||||
|
||||
|
|
@ -84,7 +86,7 @@ def test_voc_level_map_is_complete_and_ordered() -> None:
|
|||
assert set(VOC_LEVEL_MAP) == {1, 2, 3, 4, 5}
|
||||
assert set(VOC_LEVEL_MAP.values()) == set(VOCLevel)
|
||||
assert VOC_LEVEL_MAP[1] is VOCLevel.UNHEALTHY
|
||||
assert VOC_LEVEL_MAP[5] is VOCLevel.EXCELENT
|
||||
assert VOC_LEVEL_MAP[5] is VOCLevel.EXCELLENT
|
||||
assert [member.value for member in VOCLevel] == [
|
||||
"unhealthy",
|
||||
"poor",
|
||||
|
|
@ -109,7 +111,7 @@ def test_voc_level_to_text_handles_empty(empty) -> None:
|
|||
("2", VOCLevel.POOR),
|
||||
("3", VOCLevel.MODERATE),
|
||||
("4", VOCLevel.GOOD),
|
||||
("5", VOCLevel.EXCELENT),
|
||||
("5", VOCLevel.EXCELLENT),
|
||||
(3, VOCLevel.MODERATE),
|
||||
],
|
||||
)
|
||||
|
|
@ -184,8 +186,8 @@ def test_hcho_entity_description(wslink_descriptions) -> None:
|
|||
assert description.device_class is SensorDeviceClass.VOLATILE_ORGANIC_COMPOUNDS_PARTS
|
||||
assert description.native_unit_of_measurement == CONCENTRATION_PARTS_PER_BILLION
|
||||
assert description.state_class is SensorStateClass.MEASUREMENT
|
||||
# value_fn is a pass-through (typing.cast is a no-op at runtime; HA coerces the str)
|
||||
assert description.value_fn("57") == "57"
|
||||
# HCHO is a numeric ppb concentration, so value_fn coerces to int.
|
||||
assert description.value_fn("57") == 57
|
||||
|
||||
|
||||
def test_voc_entity_description(wslink_descriptions) -> None:
|
||||
|
|
|
|||
|
|
@ -5,8 +5,6 @@ from types import SimpleNamespace
|
|||
from typing import Any, Callable
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from custom_components.sws12500.const import DOMAIN
|
||||
from custom_components.sws12500.sensor import WeatherSensor
|
||||
|
||||
|
|
@ -33,7 +31,9 @@ class _CoordinatorStub:
|
|||
self, data: dict[str, Any] | None = None, *, config: Any | None = None
|
||||
):
|
||||
self.data = data if data is not None else {}
|
||||
self.config = config
|
||||
# WeatherSensor.__init__ reads coordinator.config.options for the dev-log flag,
|
||||
# so default to a config with empty options when the test doesn't supply one.
|
||||
self.config = config if config is not None else SimpleNamespace(options={})
|
||||
|
||||
|
||||
def test_native_value_prefers_value_from_data_fn_success():
|
||||
|
|
|
|||
Loading…
Reference in New Issue