service_netatmo/tests/test_service.py

from pathlib import Path

from netatmo_service import create_app
from netatmo_service.data import extract_samples
from netatmo_service.netatmo import NetatmoClient, TokenStore


class FakeClient:
    configured = False


class FakeStore:
    def ensure_all(self): pass
    def last(self, name, field): return {"rrd": name, "data_point": field, "timestamp": 1, "value": 2.0}
    def graph(self, name, period):
        if name == "missing": raise KeyError(name)
        return b"\x89PNG\r\n"


def app():
    return create_app({"TESTING": True, "START_COLLECTOR": False, "RRD_STORE": FakeStore(), "NETATMO_CLIENT": FakeClient()})


def test_routes():
    client = app().test_client()
    assert client.get("/").status_code == 200
    assert client.get("/last/outdoor/humidity").json["value"] == 2.0
    assert client.get("/graph/wind/2weeks").mimetype == "image/png"
    assert client.get("/graph/windyear").status_code == 200


def test_url_prefix_mounts_all_routes():
    prefixed = create_app({
        "TESTING": True,
        "START_COLLECTOR": False,
        "URL_PREFIX": "/w",
        "RRD_STORE": FakeStore(),
        "NETATMO_CLIENT": FakeClient(),
    }).test_client()
    response = prefixed.get("/w/")
    assert response.status_code == 200
    assert b'/w/static/dashboard.css' in response.data
    assert b'/w/graph/outdoor/day' in response.data
    assert prefixed.get("/w/health").status_code == 200
    assert prefixed.get("/w/last/outdoor/humidity").status_code == 200
    assert prefixed.get("/w/graph/wind/year").status_code == 200
    assert prefixed.get("/").status_code == 404


def test_extract_sample_payload():
    namespace = {}
    exec(Path("payload_dict.py").read_text(), namespace)
    names = {name: name.title() for name in ("outdoor", "wind", "rain", "bedroom", "study", "living")}
    samples = extract_samples(namespace["payload"], names)
    assert samples["outdoor"].values["pressure"] == 1004.8
    assert samples["outdoor"].values["temperature"] == 18.3
    assert samples["pressure"].values["pressure"] == 1004.8
    assert samples["wind"].values["angle"] == 236.0
    assert samples["rain"].values["sum_rain_24"] == 9.8
    assert samples["rain"].values["sum_rain_1"] == 1.9200000000000002
    assert "bedroom" not in samples  # unreachable module has no dashboard_data


def test_rotated_refresh_token_is_persisted(tmp_path):
    token_file = tmp_path / "tokens.json"
    config = {
        "NETATMO_CLIENT_ID": "client",
        "NETATMO_CLIENT_SECRET": "secret",
        "NETATMO_REFRESH_TOKEN": "old-refresh",
        "NETATMO_ACCESS_TOKEN": "",
        "NETATMO_TOKEN_FILE": token_file,
        "NETATMO_DEVICE_ID": "",
        "NETATMO_TOKEN_URL": "https://example.test/token",
        "NETATMO_STATIONS_URL": "https://example.test/stations",
    }
    client = NetatmoClient(config)
    client._request = lambda request: {
        "access_token": "new-access",
        "refresh_token": "new-refresh",
        "expires_in": 3600,
    }
    client._refresh()

    stored = TokenStore(token_file).load()
    assert stored["access_token"] == "new-access"
    assert stored["refresh_token"] == "new-refresh"
    assert token_file.stat().st_mode & 0o777 == 0o600

    restarted = NetatmoClient(config)
    assert restarted.access_token == "new-access"
    assert restarted.refresh_token == "new-refresh"