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
SWS-12500-custom-component/tests/test_pocasi_ecowitt.py

404 lines
14 KiB
Python
Raw Blame History

"""Forwarding an Ecowitt payload to Pocasi Meteo.
Their integration notes are specific, and every detail here is load-bearing:
- Ecowitt is accepted **only as a POST** in the Ecowitt protocol; the payload is
forwarded verbatim rather than translated into PWS field names.
- The endpoint is ``/ws_ecowitt/ws.php`` (not the PWS one).
- The account credentials go in the **query string**, and the password parameter is
``PAS`` - not ``PASS`` or ``PASSWORD`` as on the PWS endpoint.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import timedelta
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock
import pytest
from custom_components.sws12500.const import (
DEFAULT_URL,
POCASI_CZ_API_ID,
POCASI_CZ_API_KEY,
POCASI_CZ_ECOWITT_URL,
POCASI_CZ_ENABLED,
POCASI_CZ_LOGGER_ENABLED,
POCASI_CZ_SEND_INTERVAL,
POCASI_CZ_URL,
)
from custom_components.sws12500.pocasti_cz import PocasiPush
from homeassistant.util import dt as dt_util
# A representative Ecowitt POST body, including the metadata a station really sends.
ECOWITT_PAYLOAD: dict[str, str] = {
"PASSKEY": "A1B2C3D4E5F6",
"stationtype": "GW1000B_V1.6.8",
"dateutc": "2026-07-26 01:09:02",
"tempinf": "53.4",
"humidityin": "76",
"baromrelin": "28.792",
"baromabsin": "28.792",
"tempf": "43.2",
"humidity": "40",
"dewpointf": "20.3",
"winddir": "119",
"windspeedmph": "6.9",
"windgustmph": "9.7",
"solarradiation": "7.0",
"uv": "0",
"dailyrainin": "3.55",
"hourlyrainin": "2.08",
"model": "GW1000",
"freq": "868M",
}
@dataclass(slots=True)
class _FakeResponse:
status: int = 200
text_value: str = ""
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
@dataclass
class _RecordingSession:
"""Records how the request was made: verb, url, query params and body."""
response: _FakeResponse = field(default_factory=_FakeResponse)
calls: list[dict[str, Any]] = field(default_factory=list)
def get(self, url: str, *, params: dict[str, Any] | None = None, **kw: Any):
self.calls.append(
{
"verb": "GET",
"url": url,
"params": dict(params or {}),
"body": kw.get("data"),
"timeout": kw.get("timeout"),
}
)
return self.response
def post(self, url: str, *, params: dict[str, Any] | None = None, data: Any = None, **kw: Any):
self.calls.append(
{
"verb": "POST",
"url": url,
"params": dict(params or {}),
"body": data,
"timeout": kw.get("timeout"),
}
)
return self.response
def _make_entry(*, api_id: str = "myid", api_key: str = "mykey") -> Any:
return SimpleNamespace(
options={
POCASI_CZ_API_ID: api_id,
POCASI_CZ_API_KEY: api_key,
POCASI_CZ_ENABLED: True,
POCASI_CZ_LOGGER_ENABLED: False,
POCASI_CZ_SEND_INTERVAL: 30,
},
entry_id="entry",
)
@pytest.fixture
def hass():
return SimpleNamespace()
@pytest.fixture(autouse=True)
def notify(monkeypatch):
"""Capture the disable notification; the stub `hass` cannot serve the real one."""
created = MagicMock()
monkeypatch.setattr(
"custom_components.sws12500.pocasti_cz.persistent_notification.async_create", created
)
return created
@pytest.fixture
def pusher(monkeypatch, hass):
"""A PocasiPush wired to a recording session and allowed to send immediately."""
session = _RecordingSession()
monkeypatch.setattr(
"custom_components.sws12500.pocasti_cz.async_get_clientsession",
lambda _h: session,
)
monkeypatch.setattr("custom_components.sws12500.pocasti_cz.anonymize", lambda d: d)
pp = PocasiPush(hass, _make_entry())
pp.next_update = dt_util.utcnow() - timedelta(seconds=1)
return pp, session
# ---------------------------------------------------------------------------
# Request shape
# ---------------------------------------------------------------------------
async def test_ecowitt_is_posted_not_get(pusher) -> None:
"""Their server accepts the Ecowitt protocol only via POST."""
pp, session = pusher
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
assert len(session.calls) == 1
assert session.calls[0]["verb"] == "POST"
async def test_ecowitt_uses_its_own_endpoint(pusher) -> None:
pp, session = pusher
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
assert session.calls[0]["url"] == f"{POCASI_CZ_URL}{POCASI_CZ_ECOWITT_URL}"
assert session.calls[0]["url"] != f"{POCASI_CZ_URL}{DEFAULT_URL}"
async def test_credentials_go_in_the_query_string_as_id_and_pas(pusher) -> None:
"""`PAS`, not `PASS`/`PASSWORD` - the PWS spelling is rejected here."""
pp, session = pusher
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
params = session.calls[0]["params"]
assert params == {"ID": "myid", "PAS": "mykey"}
assert "PASS" not in params
assert "PASSWORD" not in params
async def test_station_payload_is_forwarded_verbatim(pusher) -> None:
"""A full forward: what the station sent is what the server receives."""
pp, session = pusher
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
assert session.calls[0]["body"] == ECOWITT_PAYLOAD
async def test_credentials_are_not_injected_into_the_body(pusher) -> None:
"""Only the query string carries our credentials; the body stays the station's."""
pp, session = pusher
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
body = session.calls[0]["body"]
for key in ("ID", "PAS", "PASSWORD", "wsid", "wspw"):
assert key not in body
async def test_ecowitt_post_is_bounded_by_a_timeout(pusher) -> None:
"""The POST is awaited inside the station's own request, so it must be bounded.
Home Assistant's shared session sets no timeout, leaving aiohttp's 5 minute
default - long past the point where the station gives up waiting for us.
"""
from custom_components.sws12500.const import FORWARD_TIMEOUT
pp, session = pusher
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
timeout = session.calls[0]["timeout"]
assert timeout is not None
assert timeout.total == FORWARD_TIMEOUT
async def test_caller_payload_is_not_mutated(pusher) -> None:
"""The same dict is handed to Windy and to the coordinator afterwards."""
pp, _ = pusher
original = dict(ECOWITT_PAYLOAD)
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
assert ECOWITT_PAYLOAD == original
# ---------------------------------------------------------------------------
# The PWS / WSLink modes keep their existing shape
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
("mode", "path", "id_key", "pw_key"),
[("WU", DEFAULT_URL, "ID", "PASSWORD"), ("WSLINK", "/data/upload.php", "wsid", "wspw")],
)
async def test_legacy_modes_still_use_get_with_inline_credentials(pusher, mode, path, id_key, pw_key) -> None:
pp, session = pusher
await pp.push_data_to_server({"tempf": "43.2"}, mode)
call = session.calls[0]
assert call["verb"] == "GET"
assert call["url"] == f"{POCASI_CZ_URL}{path}"
assert call["params"][id_key] == "myid"
assert call["params"][pw_key] == "mykey"
# ---------------------------------------------------------------------------
# Shared behaviour still applies to the new mode
# ---------------------------------------------------------------------------
async def test_ecowitt_respects_the_send_interval(monkeypatch, hass) -> None:
session = _RecordingSession()
monkeypatch.setattr(
"custom_components.sws12500.pocasti_cz.async_get_clientsession",
lambda _h: session,
)
pp = PocasiPush(hass, _make_entry())
pp.next_update = dt_util.utcnow() + timedelta(seconds=999)
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
assert session.calls == []
assert pp.last_status == "rate_limited_local"
@pytest.mark.parametrize("missing", ["api_id", "api_key"])
async def test_ecowitt_needs_credentials(monkeypatch, hass, missing) -> None:
session = _RecordingSession()
monkeypatch.setattr(
"custom_components.sws12500.pocasti_cz.async_get_clientsession",
lambda _h: session,
)
pp = PocasiPush(hass, _make_entry(**{missing: ""}))
pp.next_update = dt_util.utcnow() - timedelta(seconds=1)
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
assert session.calls == []
assert pp.last_status == "config_error"
async def test_ecowitt_auth_error_disables_forwarding(monkeypatch, hass) -> None:
session = _RecordingSession(response=_FakeResponse(status=401))
monkeypatch.setattr(
"custom_components.sws12500.pocasti_cz.async_get_clientsession",
lambda _h: session,
)
monkeypatch.setattr("custom_components.sws12500.pocasti_cz.anonymize", lambda d: d)
entry = _make_entry()
async def _apply(_hass, _entry, key, value):
entry.options[key] = value
return True
monkeypatch.setattr("custom_components.sws12500.pocasti_cz.update_options", MagicMock(side_effect=_apply))
pp = PocasiPush(hass, entry)
pp.next_update = dt_util.utcnow() - timedelta(seconds=1)
await pp.push_data_to_server(ECOWITT_PAYLOAD, "ECOWITT")
assert pp.last_status == "auth_error"
assert pp.enabled is False
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
def test_pas_is_masked_for_logging() -> None:
"""`PAS` is a credential; the debug log must not print it."""
from custom_components.sws12500.utils import anonymize
assert anonymize({"ID": "myid", "PAS": "mykey"}) == {"ID": "***", "PAS": "***"}
# ---------------------------------------------------------------------------
# Windy takes the same payload translated to the field names it accepts
#
# Windy has no Ecowitt endpoint, so `baromrelin`, `dewpointf`, `tempinf`,
# `humidityin` and `hourlyrainin` would not be understood there.
# ---------------------------------------------------------------------------
def test_ecowitt_to_windy_renames_the_diverging_fields() -> None:
from custom_components.sws12500.utils import remap_ecowitt_to_windy
translated = remap_ecowitt_to_windy(ECOWITT_PAYLOAD)
assert translated["dewptf"] == "20.3"
assert translated["baromin"] == "28.792"
assert translated["indoortempf"] == "53.4"
assert translated["indoorhumidity"] == "76"
assert translated["rainin"] == "2.08" # WU `rainin` is the past hour
# The Ecowitt spellings must be gone, not merely duplicated.
for gone in ("baromrelin", "tempinf", "humidityin", "hourlyrainin", "dewpointf"):
assert gone not in translated
def test_ecowitt_to_windy_keeps_the_shared_names() -> None:
from custom_components.sws12500.utils import remap_ecowitt_to_windy
translated = remap_ecowitt_to_windy(ECOWITT_PAYLOAD)
for same in ("tempf", "humidity", "windspeedmph", "windgustmph", "winddir", "solarradiation", "dailyrainin"):
assert translated[same] == ECOWITT_PAYLOAD[same]
def test_ecowitt_to_windy_drops_device_metadata() -> None:
"""Allowlist: station metadata has no meaning upstream and must not be forwarded."""
from custom_components.sws12500.utils import remap_ecowitt_to_windy
translated = remap_ecowitt_to_windy(ECOWITT_PAYLOAD)
for junk in ("PASSKEY", "stationtype", "model", "freq", "baromabsin"):
assert junk not in translated
def test_ecowitt_to_windy_ignores_unknown_future_fields() -> None:
"""A denylist would forward whatever a future firmware starts sending."""
from custom_components.sws12500.utils import remap_ecowitt_to_windy
translated = remap_ecowitt_to_windy({**ECOWITT_PAYLOAD, "some_new_sensor_v9": "42"})
assert "some_new_sensor_v9" not in translated
def test_ecowitt_to_windy_handles_a_partial_payload() -> None:
from custom_components.sws12500.utils import remap_ecowitt_to_windy
assert remap_ecowitt_to_windy({"tempf": "43.2"}) == {"tempf": "43.2"}
assert remap_ecowitt_to_windy({}) == {}
def test_ecowitt_to_windy_keeps_the_windy_uv_spelling() -> None:
"""Windy documents the UV index as lowercase `uv`; WU spells it `UV`.
The WSLink converter already emits `uv` on the path that demonstrably works, so
this table has to follow Windy rather than WU - otherwise Ecowitt users silently
lose the UV index upstream. See `test_both_windy_converters_agree_on_uv`.
"""
from custom_components.sws12500.utils import remap_ecowitt_to_windy
translated = remap_ecowitt_to_windy(ECOWITT_PAYLOAD)
assert translated["uv"] == "0"
assert "UV" not in translated
def test_ecowitt_to_windy_produces_names_the_wu_pipeline_knows() -> None:
"""The translated names must be the ones a real PWS station sends.
That is what makes the Windy forward work: it is the already-proven path, so the
output is pinned against the integration's own WU parser rather than guessed. `uv`
is the documented exception - Windy's own spelling, see the table's comment.
"""
from custom_components.sws12500.const import REMAP_ITEMS
from custom_components.sws12500.utils import remap_ecowitt_to_windy
translated = remap_ecowitt_to_windy(ECOWITT_PAYLOAD)
unknown = set(translated) - set(REMAP_ITEMS) - {"dateutc", "uv"}
assert not unknown, f"field names neither protocol defines: {sorted(unknown)}"
Reference in New Issue View Git Blame Copy Permalink