1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
|
"""Tests for Tibber Data API."""
from typing import Any
from unittest.mock import AsyncMock, MagicMock
import pytest
from tibber.data_api import TibberDataAPI, TibberDevice
@pytest.fixture
def data_api() -> TibberDataAPI:
"""Provide a TibberDataAPI instance with a mocked websession."""
websession = MagicMock()
return TibberDataAPI(
access_token="test-token",
timeout=10,
websession=websession,
user_agent="test-agent",
)
@pytest.mark.asyncio
async def test_get_homes_uses_data_api(data_api: TibberDataAPI, monkeypatch: pytest.MonkeyPatch) -> None:
"""Verify homes are fetched via the REST API wrapper."""
mock_response = {"homes": [{"id": "home-1"}, {"id": "home-2"}]}
mocked_make_request = AsyncMock(return_value=mock_response)
monkeypatch.setattr(data_api, "_make_request", mocked_make_request)
homes = await data_api.get_homes()
mocked_make_request.assert_awaited_once_with("GET", "/v1/homes")
assert homes == mock_response["homes"]
@pytest.mark.asyncio
async def test_get_devices_for_home_returns_raw_devices(
data_api: TibberDataAPI,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Ensure device list is returned as received from the API."""
home_id = "home-1"
mock_devices = {
"devices": [
{
"id": "device-1",
"info": {"name": "Device 1", "brand": "Brand", "model": "Model"},
"capabilities": [],
},
{
"id": "device-2",
"info": {"name": "Device 2", "brand": "Brand", "model": "Model"},
"capabilities": [],
},
],
}
mocked_make_request = AsyncMock(return_value=mock_devices)
monkeypatch.setattr(data_api, "_make_request", mocked_make_request)
devices = await data_api.get_devices_for_home(home_id)
mocked_make_request.assert_awaited_once_with("GET", f"/v1/homes/{home_id}/devices")
assert devices == mock_devices["devices"]
@pytest.mark.asyncio
async def test_get_device_returns_tibber_device(data_api: TibberDataAPI, monkeypatch: pytest.MonkeyPatch) -> None:
"""Confirm a detailed device request produces a TibberDevice instance."""
home_id = "home-1"
device_id = "device-1"
device_payload = {
"id": device_id,
"externalId": "external-id",
"info": {"name": "Device 1", "brand": "Brand", "model": "Model"},
"capabilities": [],
}
mocked_make_request = AsyncMock(return_value=device_payload)
monkeypatch.setattr(data_api, "_make_request", mocked_make_request)
device = await data_api.get_device(home_id, device_id)
mocked_make_request.assert_awaited_once_with("GET", f"/v1/homes/{home_id}/devices/{device_id}")
assert isinstance(device, TibberDevice)
assert device.id == device_id
assert device.external_id == "external-id"
assert device.name == "Device 1"
assert device.brand == "Brand"
assert device.model == "Model"
assert device.home_id == home_id
@pytest.mark.asyncio
async def test_get_all_devices_flattens_device_lists(
data_api: TibberDataAPI,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""get_all_devices should merge devices from all homes keyed by device id."""
homes_payload = {"homes": [{"id": "home-1"}, {"id": "home-2"}]}
devices_payload = {
"home-1": {
"devices": [
{"id": "device-1", "info": {"name": "D1", "brand": "B", "model": "M"}, "capabilities": []},
],
},
"home-2": {
"devices": [
{"id": "device-2", "info": {"name": "D2", "brand": "B", "model": "M"}, "capabilities": []},
],
},
}
device_detail_payload = {
"device-1": {
"id": "device-1",
"externalId": "ext-1",
"info": {"name": "D1", "brand": "B", "model": "M"},
"capabilities": [],
},
"device-2": {
"id": "device-2",
"externalId": "ext-2",
"info": {"name": "D2", "brand": "B", "model": "M"},
"capabilities": [],
},
}
async def fake_make_request(
_method: str,
endpoint: str,
_params: dict[str, Any] | None = None,
_retry: int = 3,
) -> dict[str, Any]:
if endpoint == "/v1/homes":
return homes_payload
if endpoint.endswith("/devices"):
home_id = endpoint.split("/")[3]
return devices_payload[home_id]
device_id = endpoint.split("/")[-1]
return device_detail_payload[device_id]
monkeypatch.setattr(data_api, "_make_request", AsyncMock(side_effect=fake_make_request))
devices = await data_api.get_all_devices()
assert set(devices.keys()) == {"device-1", "device-2"}
assert all(isinstance(value, TibberDevice) for value in devices.values())
assert {value.home_id for value in devices.values()} == {"home-1", "home-2"}
def test_tibber_device_properties_match_payload() -> None:
"""Ensure TibberDevice exposes expected basic properties."""
device_data = {
"id": "test-device-id",
"externalId": "test-external-id",
"info": {"name": "Test Device", "brand": "Test Brand", "model": "Test Model"},
"capabilities": [
{"id": "sensor-1", "unit": "W", "value": 10, "description": "power usage"},
],
}
device = TibberDevice(device_data, home_id="home-1")
assert device.id == "test-device-id"
assert device.external_id == "test-external-id"
assert device.name == "Test Device"
assert device.brand == "Test Brand"
assert device.model == "Test Model"
assert device.home_id == "home-1"
assert [sensor.id for sensor in device.sensors] == ["sensor-1"]
|
