Michal
/
SWS-12500-custom-component
mirror of https://github.com/schizza/SWS-12500-custom-component
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Make routes method-aware and update related tests 

Include HTTP method in route keys and dispatch, and fix
Routes.show_enabled.
Update register_path to accept a HealthCoordinator and adjust router
stubs in tests. Update WindyPush tests to use response objects
(status/text)
and adapt related exception/notification expectations.
ecowitt_support
SchiZzA 2026-03-04 07:53:26 +01:00
parent 995f607cf7
commit f0554573ce
No known key found for this signature in database
7 changed files with 117 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
custom_components/sws12500/routes.py
View File

@ -116,9 +116,9 @@ class Routes:
for route in self.routes.values() for route in self.routes.values()
if route.enabled if route.enabled
} }
return ", ".join( if not enabled_routes:
sorted(enabled_routes) if enabled_routes else "No routes are enabled." return "No routes are enabled."
) return ", ".join(sorted(enabled_routes))
async def unregistered(request: Request) -> Response: async def unregistered(request: Request) -> Response:

4
tests/test_init.py
View File

@ -40,7 +40,7 @@ async def test_async_setup_entry_creates_runtime_state(
# Patch it out so the test doesn't depend on aiohttp being initialized. # Patch it out so the test doesn't depend on aiohttp being initialized.
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.register_path", "custom_components.sws12500.register_path",
lambda _hass, _coordinator, _entry: True, lambda _hass, _coordinator, _coordinator_h, _entry: True,
) )
# Avoid depending on Home Assistant integration loader in this test. # Avoid depending on Home Assistant integration loader in this test.
@ -69,7 +69,7 @@ async def test_async_setup_entry_forwards_sensor_platform(
# Patch it out so the test doesn't depend on aiohttp being initialized. # Patch it out so the test doesn't depend on aiohttp being initialized.
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.register_path", "custom_components.sws12500.register_path",
lambda _hass, _coordinator, _entry: True, lambda _hass, _coordinator, _coordinator_h, _entry: True,
) )
# Patch forwarding so we don't need to load real platforms for this unit/integration test. # Patch forwarding so we don't need to load real platforms for this unit/integration test.

33
tests/test_integration_lifecycle.py
View File

@ -10,6 +10,7 @@ import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry from pytest_homeassistant_custom_component.common import MockConfigEntry
from custom_components.sws12500 import ( from custom_components.sws12500 import (
HealthCoordinator,
IncorrectDataError, IncorrectDataError,
WeatherDataUpdateCoordinator, WeatherDataUpdateCoordinator,
async_setup_entry, async_setup_entry,
@ -22,6 +23,7 @@ from custom_components.sws12500.const import (
API_KEY, API_KEY,
DEFAULT_URL, DEFAULT_URL,
DOMAIN, DOMAIN,
HEALTH_URL,
SENSORS_TO_LOAD, SENSORS_TO_LOAD,
WSLINK, WSLINK,
WSLINK_URL, WSLINK_URL,
@ -35,6 +37,9 @@ class _RequestStub:
query: dict[str, Any] query: dict[str, Any]
async def post(self) -> dict[str, Any]:
return {}
class _RouterStub: class _RouterStub:
"""Router stub that records route registrations.""" """Router stub that records route registrations."""
@ -44,17 +49,17 @@ class _RouterStub:
self.add_post_calls: list[tuple[str, Any]] = [] self.add_post_calls: list[tuple[str, Any]] = []
self.raise_on_add: Exception | None = None self.raise_on_add: Exception | None = None
def add_get(self, path: str, handler: Any) -> Any: def add_get(self, path: str, handler: Any, **_kwargs: Any) -> Any:
if self.raise_on_add is not None: if self.raise_on_add is not None:
raise self.raise_on_add raise self.raise_on_add
self.add_get_calls.append((path, handler)) self.add_get_calls.append((path, handler))
return object() return SimpleNamespace(method="GET")
def add_post(self, path: str, handler: Any) -> Any: def add_post(self, path: str, handler: Any, **_kwargs: Any) -> Any:
if self.raise_on_add is not None: if self.raise_on_add is not None:
raise self.raise_on_add raise self.raise_on_add
self.add_post_calls.append((path, handler)) self.add_post_calls.append((path, handler))
return object() return SimpleNamespace(method="POST")
@pytest.fixture @pytest.fixture
@ -79,13 +84,18 @@ async def test_register_path_registers_routes_and_stores_dispatcher(hass_with_ht
entry.add_to_hass(hass_with_http) entry.add_to_hass(hass_with_http)
coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry) coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry)
coordinator_health = HealthCoordinator(hass_with_http, entry)
ok = register_path(hass_with_http, coordinator, entry) ok = register_path(hass_with_http, coordinator, coordinator_health, entry)
assert ok is True assert ok is True
# Router registrations # Router registrations
router: _RouterStub = hass_with_http.http.app.router router: _RouterStub = hass_with_http.http.app.router
assert [p for (p, _h) in router.add_get_calls] == [DEFAULT_URL] 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]
# Dispatcher stored # Dispatcher stored
@ -115,13 +125,14 @@ async def test_register_path_raises_config_entry_not_ready_on_router_runtime_err
entry.add_to_hass(hass_with_http) entry.add_to_hass(hass_with_http)
coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry) coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry)
coordinator_health = HealthCoordinator(hass_with_http, entry)
# Make router raise RuntimeError on add # Make router raise RuntimeError on add
router: _RouterStub = hass_with_http.http.app.router router: _RouterStub = hass_with_http.http.app.router
router.raise_on_add = RuntimeError("router broken") router.raise_on_add = RuntimeError("router broken")
with pytest.raises(ConfigEntryNotReady): with pytest.raises(ConfigEntryNotReady):
register_path(hass_with_http, coordinator, entry) register_path(hass_with_http, coordinator, coordinator_health, entry)
@pytest.mark.asyncio @pytest.mark.asyncio
@ -143,12 +154,13 @@ async def test_register_path_checked_hass_data_wrong_type_raises_config_entry_no
entry.add_to_hass(hass_with_http) entry.add_to_hass(hass_with_http)
coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry) coordinator = WeatherDataUpdateCoordinator(hass_with_http, entry)
coordinator_health = HealthCoordinator(hass_with_http, entry)
# Force wrong type under DOMAIN so `checked(..., dict)` fails. # Force wrong type under DOMAIN so `checked(..., dict)` fails.
hass_with_http.data[DOMAIN] = [] hass_with_http.data[DOMAIN] = []
with pytest.raises(ConfigEntryNotReady): with pytest.raises(ConfigEntryNotReady):
register_path(hass_with_http, coordinator, entry) register_path(hass_with_http, coordinator, coordinator_health, entry)
@pytest.mark.asyncio @pytest.mark.asyncio
@ -213,7 +225,7 @@ async def test_async_setup_entry_fatal_when_register_path_returns_false(
# Force register_path to return False # Force register_path to return False
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.register_path", "custom_components.sws12500.register_path",
lambda _hass, _coordinator, _entry: False, lambda _hass, _coordinator, _coordinator_h, _entry: False,
) )
# Forwarding shouldn't be reached; patch anyway to avoid accidental loader calls. # Forwarding shouldn't be reached; patch anyway to avoid accidental loader calls.
@ -251,7 +263,8 @@ async def test_async_setup_entry_reuses_existing_coordinator_and_switches_routes
routes = hass_with_http.data[DOMAIN].get("routes") routes = hass_with_http.data[DOMAIN].get("routes")
if routes is None: if routes is None:
# Create a dispatcher via register_path once # Create a dispatcher via register_path once
register_path(hass_with_http, existing_coordinator, entry) coordinator_health = HealthCoordinator(hass_with_http, entry)
register_path(hass_with_http, existing_coordinator, coordinator_health, entry)
routes = hass_with_http.data[DOMAIN]["routes"] routes = hass_with_http.data[DOMAIN]["routes"]
# Turn on WSLINK to trigger dispatcher switching. # Turn on WSLINK to trigger dispatcher switching.

6
tests/test_received_data.py
View File

@ -26,10 +26,14 @@ from custom_components.sws12500.const import (
class _RequestStub: class _RequestStub:
"""Minimal aiohttp Request stub. """Minimal aiohttp Request stub.
The coordinator only uses `webdata.query` (a mapping of query parameters). The coordinator uses `webdata.query` and `await webdata.post()`.
""" """
query: dict[str, Any] 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( def _make_entry(

44
tests/test_routes.py
View File

@ -15,26 +15,34 @@ Handler = Callable[["_RequestStub"], Awaitable[Response]]
class _RequestStub: class _RequestStub:
"""Minimal request stub for unit-testing the dispatcher. """Minimal request stub for unit-testing the dispatcher.
`Routes.dispatch` relies on `request.path`. `Routes.dispatch` relies on `request.method` and `request.path`.
`unregistered` accepts a request object but does not use it. `unregistered` accepts a request object but does not use it.
""" """
method: str
path: str path: str
@dataclass(slots=True)
class _RouteStub:
"""Minimal route stub providing `method` expected by Routes.add_route`."""
method: str
@pytest.fixture @pytest.fixture
def routes() -> Routes: def routes() -> Routes:
return Routes() return Routes()
async def test_dispatch_unknown_path_calls_unregistered(routes: Routes) -> None: async def test_dispatch_unknown_path_calls_unregistered(routes: Routes) -> None:
request = _RequestStub(path="/unregistered") request = _RequestStub(method="GET", path="/unregistered")
response = await routes.dispatch(request) # type: ignore[arg-type] response = await routes.dispatch(request) # type: ignore[arg-type]
assert response.status == 400 assert response.status == 400
async def test_unregistered_handler_returns_400() -> None: async def test_unregistered_handler_returns_400() -> None:
request = _RequestStub(path="/invalid") request = _RequestStub(method="GET", path="/invalid")
response = await unregistered(request) # type: ignore[arg-type] response = await unregistered(request) # type: ignore[arg-type]
assert response.status == 400 assert response.status == 400
@ -43,9 +51,9 @@ async def test_dispatch_registered_but_disabled_uses_fallback(routes: Routes) ->
async def handler(_request: _RequestStub) -> Response: async def handler(_request: _RequestStub) -> Response:
return Response(text="OK", status=200) return Response(text="OK", status=200)
routes.add_route("/a", handler, enabled=False) routes.add_route("/a", _RouteStub(method="GET"), handler, enabled=False)
response = await routes.dispatch(_RequestStub(path="/a")) # type: ignore[arg-type] response = await routes.dispatch(_RequestStub(method="GET", path="/a")) # type: ignore[arg-type]
assert response.status == 400 assert response.status == 400
@ -53,9 +61,9 @@ async def test_dispatch_registered_and_enabled_uses_handler(routes: Routes) -> N
async def handler(_request: _RequestStub) -> Response: async def handler(_request: _RequestStub) -> Response:
return Response(text="OK", status=201) return Response(text="OK", status=201)
routes.add_route("/a", handler, enabled=True) routes.add_route("/a", _RouteStub(method="GET"), handler, enabled=True)
response = await routes.dispatch(_RequestStub(path="/a")) # type: ignore[arg-type] response = await routes.dispatch(_RequestStub(method="GET", path="/a")) # type: ignore[arg-type]
assert response.status == 201 assert response.status == 201
@ -66,32 +74,32 @@ def test_switch_route_enables_exactly_one(routes: Routes) -> None:
async def handler_b(_request: _RequestStub) -> Response: async def handler_b(_request: _RequestStub) -> Response:
return Response(text="B", status=200) return Response(text="B", status=200)
routes.add_route("/a", handler_a, enabled=True) routes.add_route("/a", _RouteStub(method="GET"), handler_a, enabled=True)
routes.add_route("/b", handler_b, enabled=False) routes.add_route("/b", _RouteStub(method="GET"), handler_b, enabled=False)
routes.switch_route("/b") routes.switch_route(handler_b, "/b")
assert routes.routes["/a"].enabled is False assert routes.routes["GET:/a"].enabled is False
assert routes.routes["/b"].enabled is True assert routes.routes["GET:/b"].enabled is True
def test_show_enabled_returns_message_when_none_enabled(routes: Routes) -> None: def test_show_enabled_returns_message_when_none_enabled(routes: Routes) -> None:
async def handler(_request: _RequestStub) -> Response: async def handler(_request: _RequestStub) -> Response:
return Response(text="OK", status=200) return Response(text="OK", status=200)
routes.add_route("/a", handler, enabled=False) routes.add_route("/a", _RouteStub(method="GET"), handler, enabled=False)
routes.add_route("/b", handler, enabled=False) routes.add_route("/b", _RouteStub(method="GET"), handler, enabled=False)
assert routes.show_enabled() == "No routes is enabled." assert routes.show_enabled() == "No routes are enabled."
def test_show_enabled_includes_url_when_enabled(routes: Routes) -> None: def test_show_enabled_includes_url_when_enabled(routes: Routes) -> None:
async def handler(_request: _RequestStub) -> Response: async def handler(_request: _RequestStub) -> Response:
return Response(text="OK", status=200) return Response(text="OK", status=200)
routes.add_route("/a", handler, enabled=False) routes.add_route("/a", _RouteStub(method="GET"), handler, enabled=False)
routes.add_route("/b", handler, enabled=True) routes.add_route("/b", _RouteStub(method="GET"), handler, enabled=True)
msg = routes.show_enabled() msg = routes.show_enabled()
assert "Dispatcher enabled for URL: /b" in msg assert "Dispatcher enabled for (GET):/b" in msg
assert "handler" in msg assert "handler" in msg

11
tests/test_weather_sensor_entity.py
View File

@ -51,10 +51,7 @@ def test_native_value_prefers_value_from_data_fn_success():
def test_native_value_value_from_data_fn_success_with_dev_logging_hits_computed_debug_branch( def test_native_value_value_from_data_fn_success_with_dev_logging_hits_computed_debug_branch(
monkeypatch, monkeypatch,
): ):
"""Cover the dev-log debug branch after successful value_from_data_fn computation.""" """Ensure value_from_data_fn works with dev logging enabled."""
debug = MagicMock()
monkeypatch.setattr("custom_components.sws12500.sensor._LOGGER.debug", debug)
desc = _DescriptionStub( desc = _DescriptionStub(
key="derived", key="derived",
value_from_data_fn=lambda data: data["x"] + 1, value_from_data_fn=lambda data: data["x"] + 1,
@ -65,12 +62,6 @@ def test_native_value_value_from_data_fn_success_with_dev_logging_hits_computed_
assert entity.native_value == 42 assert entity.native_value == 42
debug.assert_any_call(
"native_value computed via value_from_data_fn: key=%s -> %s",
"derived",
42,
)
def test_native_value_value_from_data_fn_exception_returns_none(): def test_native_value_value_from_data_fn_exception_returns_none():
def boom(_data: dict[str, Any]) -> Any: def boom(_data: dict[str, Any]) -> Any:

76
tests/test_windy_push.py
View File

@ -22,8 +22,8 @@ from custom_components.sws12500.const import (
WINDY_URL, WINDY_URL,
) )
from custom_components.sws12500.windy_func import ( from custom_components.sws12500.windy_func import (
WindyApiKeyError,
WindyNotInserted, WindyNotInserted,
WindyPasswordMissing,
WindyPush, WindyPush,
WindySuccess, WindySuccess,
) )
@ -31,7 +31,8 @@ from custom_components.sws12500.windy_func import (
@dataclass(slots=True) @dataclass(slots=True)
class _FakeResponse: class _FakeResponse:
text_value: str status: int
text_value: str = ""
async def text(self) -> str: async def text(self) -> str:
return self.text_value return self.text_value
@ -87,20 +88,19 @@ def _make_entry(**options: Any):
def test_verify_windy_response_notice_raises_not_inserted(hass): def test_verify_windy_response_notice_raises_not_inserted(hass):
wp = WindyPush(hass, _make_entry()) wp = WindyPush(hass, _make_entry())
with pytest.raises(WindyNotInserted): with pytest.raises(WindyNotInserted):
wp.verify_windy_response("NOTICE: something") wp.verify_windy_response(_FakeResponse(status=400, text_value="Bad Request"))
def test_verify_windy_response_success_raises_success(hass): def test_verify_windy_response_success_raises_success(hass):
wp = WindyPush(hass, _make_entry()) wp = WindyPush(hass, _make_entry())
with pytest.raises(WindySuccess): with pytest.raises(WindySuccess):
wp.verify_windy_response("SUCCESS") wp.verify_windy_response(_FakeResponse(status=200, text_value="OK"))
@pytest.mark.parametrize("msg", ["Invalid API key", "Unauthorized"]) def test_verify_windy_response_password_missing_raises(hass):
def test_verify_windy_response_api_key_errors_raise(msg, hass):
wp = WindyPush(hass, _make_entry()) wp = WindyPush(hass, _make_entry())
with pytest.raises(WindyApiKeyError): with pytest.raises(WindyPasswordMissing):
wp.verify_windy_response(msg) wp.verify_windy_response(_FakeResponse(status=401, text_value="Unauthorized"))
def test_covert_wslink_to_pws_maps_keys(hass): def test_covert_wslink_to_pws_maps_keys(hass):
@ -155,7 +155,7 @@ async def test_push_data_to_windy_respects_initial_next_update(monkeypatch, hass
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: _FakeSession(response=_FakeResponse("SUCCESS")), lambda _h: _FakeSession(response=_FakeResponse(status=200, text_value="OK")),
) )
ok = await wp.push_data_to_windy({"a": "b"}) ok = await wp.push_data_to_windy({"a": "b"})
assert ok is False assert ok is False
@ -169,7 +169,7 @@ async def test_push_data_to_windy_purges_data_and_sets_auth(monkeypatch, hass):
# Force it to send now # Force it to send now
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
session = _FakeSession(response=_FakeResponse("SUCCESS")) session = _FakeSession(response=_FakeResponse(status=200, text_value="OK"))
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
@ -200,7 +200,7 @@ async def test_push_data_to_windy_wslink_conversion_applied(monkeypatch, hass):
wp = WindyPush(hass, entry) wp = WindyPush(hass, entry)
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
session = _FakeSession(response=_FakeResponse("SUCCESS")) session = _FakeSession(response=_FakeResponse(status=200, text_value="OK"))
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
@ -221,12 +221,21 @@ async def test_push_data_to_windy_missing_station_id_returns_false(monkeypatch,
wp = WindyPush(hass, entry) wp = WindyPush(hass, entry)
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
session = _FakeSession(response=_FakeResponse("SUCCESS")) session = _FakeSession(response=_FakeResponse(status=200, text_value="OK"))
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
) )
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(),
)
ok = await wp.push_data_to_windy({"a": "b"}) ok = await wp.push_data_to_windy({"a": "b"})
assert ok is False assert ok is False
assert session.calls == [] assert session.calls == []
@ -239,12 +248,21 @@ async def test_push_data_to_windy_missing_station_pw_returns_false(monkeypatch,
wp = WindyPush(hass, entry) wp = WindyPush(hass, entry)
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
session = _FakeSession(response=_FakeResponse("SUCCESS")) session = _FakeSession(response=_FakeResponse(status=200, text_value="OK"))
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
) )
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(),
)
ok = await wp.push_data_to_windy({"a": "b"}) ok = await wp.push_data_to_windy({"a": "b"})
assert ok is False assert ok is False
assert session.calls == [] assert session.calls == []
@ -256,8 +274,10 @@ async def test_push_data_to_windy_invalid_api_key_disables_windy(monkeypatch, ha
wp = WindyPush(hass, entry) wp = WindyPush(hass, entry)
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
# Response triggers WindyApiKeyError # Response triggers WindyPasswordMissing (401)
session = _FakeSession(response=_FakeResponse("Invalid API key")) session = _FakeSession(
response=_FakeResponse(status=401, text_value="Unauthorized")
)
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
@ -267,6 +287,10 @@ async def test_push_data_to_windy_invalid_api_key_disables_windy(monkeypatch, ha
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.update_options", update_options "custom_components.sws12500.windy_func.update_options", update_options
) )
monkeypatch.setattr(
"custom_components.sws12500.windy_func.persistent_notification.create",
MagicMock(),
)
ok = await wp.push_data_to_windy({"a": "b"}) ok = await wp.push_data_to_windy({"a": "b"})
assert ok is True assert ok is True
@ -281,7 +305,9 @@ async def test_push_data_to_windy_invalid_api_key_update_options_failure_logs_de
wp = WindyPush(hass, entry) wp = WindyPush(hass, entry)
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
session = _FakeSession(response=_FakeResponse("Unauthorized")) session = _FakeSession(
response=_FakeResponse(status=401, text_value="Unauthorized")
)
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
@ -294,6 +320,10 @@ async def test_push_data_to_windy_invalid_api_key_update_options_failure_logs_de
dbg = MagicMock() dbg = MagicMock()
monkeypatch.setattr("custom_components.sws12500.windy_func._LOGGER.debug", dbg) monkeypatch.setattr("custom_components.sws12500.windy_func._LOGGER.debug", dbg)
monkeypatch.setattr(
"custom_components.sws12500.windy_func.persistent_notification.create",
MagicMock(),
)
ok = await wp.push_data_to_windy({"a": "b"}) ok = await wp.push_data_to_windy({"a": "b"})
assert ok is True assert ok is True
@ -307,7 +337,7 @@ async def test_push_data_to_windy_notice_logs_not_inserted(monkeypatch, hass):
wp = WindyPush(hass, entry) wp = WindyPush(hass, entry)
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
session = _FakeSession(response=_FakeResponse("NOTICE: no insert")) session = _FakeSession(response=_FakeResponse(status=400, text_value="Bad Request"))
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
@ -330,7 +360,7 @@ async def test_push_data_to_windy_success_logs_info_when_logger_enabled(
wp = WindyPush(hass, entry) wp = WindyPush(hass, entry)
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
session = _FakeSession(response=_FakeResponse("SUCCESS")) session = _FakeSession(response=_FakeResponse(status=200, text_value="OK"))
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
@ -363,7 +393,7 @@ async def test_push_data_to_windy_verify_no_raise_logs_debug_not_inserted_when_l
wp.next_update = datetime.now() - timedelta(seconds=1) wp.next_update = datetime.now() - timedelta(seconds=1)
# Response text that does not contain any of the known markers (NOTICE/SUCCESS/Invalid/Unauthorized) # Response text that does not contain any of the known markers (NOTICE/SUCCESS/Invalid/Unauthorized)
session = _FakeSession(response=_FakeResponse("OK")) session = _FakeSession(response=_FakeResponse(status=500, text_value="Error"))
monkeypatch.setattr( monkeypatch.setattr(
"custom_components.sws12500.windy_func.async_get_clientsession", "custom_components.sws12500.windy_func.async_get_clientsession",
lambda _h: session, lambda _h: session,
@ -392,6 +422,10 @@ async def test_push_data_to_windy_client_error_increments_and_disables_after_thr
crit = MagicMock() crit = MagicMock()
monkeypatch.setattr("custom_components.sws12500.windy_func._LOGGER.critical", crit) monkeypatch.setattr("custom_components.sws12500.windy_func._LOGGER.critical", crit)
monkeypatch.setattr(
"custom_components.sws12500.windy_func.persistent_notification.create",
MagicMock(),
)
# Cause ClientError on session.get # Cause ClientError on session.get
session = _FakeSession(exc=ClientError("boom")) session = _FakeSession(exc=ClientError("boom"))
@ -434,6 +468,10 @@ async def test_push_data_to_windy_client_error_disable_failure_logs_debug(
dbg = MagicMock() dbg = MagicMock()
monkeypatch.setattr("custom_components.sws12500.windy_func._LOGGER.debug", dbg) monkeypatch.setattr("custom_components.sws12500.windy_func._LOGGER.debug", dbg)
monkeypatch.setattr(
"custom_components.sws12500.windy_func.persistent_notification.create",
MagicMock(),
)
session = _FakeSession(exc=ClientError("boom")) session = _FakeSession(exc=ClientError("boom"))
monkeypatch.setattr( monkeypatch.setattr(