from __future__ import annotations from dataclasses import dataclass from datetime import timedelta from types import SimpleNamespace from typing import Any, Literal from unittest.mock import AsyncMock, MagicMock from aiohttp import ClientError import pytest from custom_components.sws12500.const import ( DEFAULT_URL, POCASI_CZ_API_ID, POCASI_CZ_API_KEY, POCASI_CZ_ENABLED, POCASI_CZ_LOGGER_ENABLED, POCASI_CZ_MAX_RETRIES, POCASI_CZ_SEND_INTERVAL, POCASI_CZ_UNEXPECTED, POCASI_CZ_URL, POCASI_INVALID_KEY, WSLINK_URL, ) from custom_components.sws12500.pocasti_cz import PocasiPush from homeassistant.util import dt as dt_util @dataclass(slots=True) class _FakeResponse: text_value: str status: int = 200 async def text(self) -> str: return self.text_value 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 = None, exc: Exception | None = None ): self._response = response self._exc = exc self.calls: list[dict[str, Any]] = [] def get(self, url: str, *, params: dict[str, Any] | None = None): self.calls.append({"url": url, "params": dict(params or {})}) if self._exc is not None: raise self._exc assert self._response is not None return self._response def _make_entry( *, api_id: str | None = "id", api_key: str | None = "key", interval: int = 30, logger: bool = False, ) -> Any: options: dict[str, Any] = { POCASI_CZ_SEND_INTERVAL: interval, POCASI_CZ_LOGGER_ENABLED: logger, POCASI_CZ_ENABLED: True, } if api_id is not None: options[POCASI_CZ_API_ID] = api_id if api_key is not None: options[POCASI_CZ_API_KEY] = api_key entry = SimpleNamespace() entry.options = options entry.entry_id = "test_entry_id" return entry def _write_through_update_options(entry: Any) -> AsyncMock: """Mock `update_options` that really mutates the entry, like the real helper. `PocasiPush.enabled` reads the option back, so a mock that only records the call would leave `enabled` reporting the pre-disable value. """ async def _apply(_hass, _entry, key, value): entry.options[key] = value return True return AsyncMock(side_effect=_apply) @pytest.fixture def hass(): # Minimal hass-like object; we patch client session retrieval. return SimpleNamespace() @pytest.mark.asyncio async def test_push_data_to_server_missing_api_id_returns_early(monkeypatch, hass): entry = _make_entry(api_id=None, api_key="key") pp = PocasiPush(hass, entry) session = _FakeSession(response=_FakeResponse("OK")) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) await pp.push_data_to_server({"x": 1}, "WU") assert session.calls == [] @pytest.mark.asyncio async def test_push_data_to_server_missing_api_key_returns_early(monkeypatch, hass): entry = _make_entry(api_id="id", api_key=None) pp = PocasiPush(hass, entry) session = _FakeSession(response=_FakeResponse("OK")) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) await pp.push_data_to_server({"x": 1}, "WU") assert session.calls == [] @pytest.mark.asyncio async def test_push_data_to_server_respects_interval_limit(monkeypatch, hass): entry = _make_entry(interval=30, logger=True) pp = PocasiPush(hass, entry) # Ensure "next_update > now" so it returns early before doing HTTP. pp.next_update = dt_util.utcnow() + timedelta(seconds=999) session = _FakeSession(response=_FakeResponse("OK")) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) await pp.push_data_to_server({"x": 1}, "WU") assert session.calls == [] @pytest.mark.asyncio @pytest.mark.parametrize( "mode,expected_path", [("WU", DEFAULT_URL), ("WSLINK", WSLINK_URL)] ) async def test_push_data_to_server_injects_auth_and_chooses_url( monkeypatch, hass, mode: Literal["WU", "WSLINK"], expected_path: str ): entry = _make_entry(api_id="id", api_key="key") pp = PocasiPush(hass, entry) # Force send now. pp.next_update = dt_util.utcnow() - timedelta(seconds=1) session = _FakeSession(response=_FakeResponse("OK")) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) # Avoid depending on anonymize output; just make it deterministic. monkeypatch.setattr("custom_components.sws12500.pocasti_cz.anonymize", lambda d: d) await pp.push_data_to_server({"temp": 1}, mode) assert len(session.calls) == 1 call = session.calls[0] assert call["url"] == f"{POCASI_CZ_URL}{expected_path}" params = call["params"] if mode == "WU": assert params["ID"] == "id" assert params["PASSWORD"] == "key" else: assert params["wsid"] == "id" assert params["wspw"] == "key" @pytest.mark.asyncio async def test_push_data_to_server_calls_verify_response(monkeypatch, hass): entry = _make_entry() pp = PocasiPush(hass, entry) pp.next_update = dt_util.utcnow() - timedelta(seconds=1) session = _FakeSession(response=_FakeResponse("OK")) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) monkeypatch.setattr("custom_components.sws12500.pocasti_cz.anonymize", lambda d: d) verify = MagicMock(return_value="ok") monkeypatch.setattr(pp, "verify_response", verify) await pp.push_data_to_server({"x": 1}, "WU") verify.assert_called_once_with(200, "OK") @pytest.mark.asyncio @pytest.mark.parametrize("status", [401, 403]) async def test_push_data_to_server_auth_error_disables_feature(monkeypatch, hass, status): """A 401/403 disables resending immediately - credentials will not self-heal.""" entry = _make_entry() pp = PocasiPush(hass, entry) pp.next_update = dt_util.utcnow() - timedelta(seconds=1) session = _FakeSession(response=_FakeResponse("", status=status)) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) monkeypatch.setattr("custom_components.sws12500.pocasti_cz.anonymize", lambda d: d) update_options = _write_through_update_options(entry) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.update_options", update_options ) crit = MagicMock() monkeypatch.setattr("custom_components.sws12500.pocasti_cz._LOGGER.critical", crit) await pp.push_data_to_server({"x": 1}, "WU") crit.assert_called() # Should log invalid key message and disable feature. assert any( POCASI_INVALID_KEY in str(c.args[0]) for c in crit.call_args_list if c.args ) update_options.assert_awaited_once_with(hass, entry, POCASI_CZ_ENABLED, False) assert pp.enabled is False assert pp.last_status == "auth_error" @pytest.mark.asyncio async def test_push_data_to_server_success_logs_when_logger_enabled(monkeypatch, hass): entry = _make_entry(logger=True) pp = PocasiPush(hass, entry) pp.next_update = dt_util.utcnow() - timedelta(seconds=1) # A previous failure must be cleared by a successful send. pp.invalid_response_count = 2 session = _FakeSession(response=_FakeResponse("OK", status=200)) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) monkeypatch.setattr("custom_components.sws12500.pocasti_cz.anonymize", lambda d: d) info = MagicMock() monkeypatch.setattr("custom_components.sws12500.pocasti_cz._LOGGER.info", info) await pp.push_data_to_server({"x": 1}, "WU") info.assert_called() assert pp.last_status == "ok" assert pp.last_error is None assert pp.invalid_response_count == 0 @pytest.mark.asyncio async def test_push_data_to_server_server_error_disables_after_max_retries(monkeypatch, hass): """HTTP 500 is no longer silently reported as success; it counts toward the limit.""" entry = _make_entry() pp = PocasiPush(hass, entry) session = _FakeSession(response=_FakeResponse("", status=500)) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) monkeypatch.setattr("custom_components.sws12500.pocasti_cz.anonymize", lambda d: d) update_options = _write_through_update_options(entry) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.update_options", update_options ) for _ in range(POCASI_CZ_MAX_RETRIES - 1): pp.next_update = dt_util.utcnow() - timedelta(seconds=1) await pp.push_data_to_server({"x": 1}, "WU") assert pp.last_status == "unexpected_response" assert pp.enabled is True update_options.assert_not_awaited() pp.next_update = dt_util.utcnow() - timedelta(seconds=1) await pp.push_data_to_server({"x": 1}, "WU") assert pp.enabled is False update_options.assert_awaited_once_with(hass, entry, POCASI_CZ_ENABLED, False) @pytest.mark.asyncio async def test_push_data_to_server_client_error_increments_and_disables_after_three( monkeypatch, hass ): entry = _make_entry() pp = PocasiPush(hass, entry) update_options = _write_through_update_options(entry) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.update_options", update_options ) crit = MagicMock() monkeypatch.setattr("custom_components.sws12500.pocasti_cz._LOGGER.critical", crit) session = _FakeSession(exc=ClientError("boom")) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) # Force request attempts and exceed invalid count threshold. for _i in range(4): pp.next_update = dt_util.utcnow() - timedelta(seconds=1) await pp.push_data_to_server({"x": 1}, "WU") assert pp.invalid_response_count == 4 # Should disable after >3 update_options.assert_awaited() args = update_options.await_args.args assert args[2] == POCASI_CZ_ENABLED assert args[3] is False # Should log unexpected at least once assert any( POCASI_CZ_UNEXPECTED in str(c.args[0]) for c in crit.call_args_list if c.args ) def test_verify_response_logs_debug_when_logger_enabled(monkeypatch, hass): entry = _make_entry(logger=True) pp = PocasiPush(hass, entry) dbg = MagicMock() monkeypatch.setattr("custom_components.sws12500.pocasti_cz._LOGGER.debug", dbg) assert pp.verify_response(200, "anything") == "ok" dbg.assert_called() @pytest.mark.parametrize( ("status", "expected"), [ (200, "ok"), (204, "ok"), (299, "ok"), (401, "auth_error"), (403, "auth_error"), (400, "unexpected_response"), (404, "unexpected_response"), (500, "unexpected_response"), (503, "unexpected_response"), ], ) def test_verify_response_status_mapping(hass, status, expected): """Every send outcome is derived from the HTTP status, not from the (empty) body.""" pp = PocasiPush(hass, _make_entry()) assert pp.verify_response(status, "") == expected @pytest.mark.asyncio async def test_disable_pocasi_logs_when_option_write_fails(monkeypatch, hass): """A failed option write is logged but still leaves resending off in memory.""" entry = _make_entry() pp = PocasiPush(hass, entry) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.update_options", AsyncMock(return_value=False), ) dbg = MagicMock() monkeypatch.setattr("custom_components.sws12500.pocasti_cz._LOGGER.debug", dbg) await pp._disable_pocasi("because") # `enabled` mirrors the persisted option: if the write failed, forwarding is still # on as far as the config is concerned, and the failure is logged instead. assert pp.enabled is True assert pp.last_error == "because" dbg.assert_called() # --------------------------------------------------------------------------- # Live `enabled` and empty-credential rejection # --------------------------------------------------------------------------- def test_enabled_reads_options_live(hass): """Toggling the option is visible immediately - no reload, no cached copy. `update_listener` deliberately skips the reload when only this flag changes, so a value cached in __init__ would leave the diagnostics sensor permanently stale. """ entry = _make_entry() pp = PocasiPush(hass, entry) assert pp.enabled is True entry.options[POCASI_CZ_ENABLED] = False assert pp.enabled is False entry.options[POCASI_CZ_ENABLED] = True assert pp.enabled is True @pytest.mark.asyncio @pytest.mark.parametrize( ("api_id", "api_key"), [("", "key"), ("id", ""), ("", "")], ids=["empty-id", "empty-key", "both-empty"], ) async def test_empty_credentials_never_reach_the_network(monkeypatch, hass, api_id, api_key): """An empty string is still a `str`, so it must be rejected explicitly. Otherwise a blank configuration sends a request that can only ever be refused. """ entry = _make_entry(api_id=api_id, api_key=api_key) pp = PocasiPush(hass, entry) pp.next_update = dt_util.utcnow() - timedelta(seconds=1) session = _FakeSession(response=_FakeResponse("OK")) monkeypatch.setattr( "custom_components.sws12500.pocasti_cz.async_get_clientsession", lambda _h: session, ) await pp.push_data_to_server({"x": 1}, "WU") assert session.calls == [] assert pp.last_status == "config_error"