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
|
"""Tests for aiopvpc."""
from __future__ import annotations
import logging
from datetime import datetime
import pytest
from aiopvpc.const import (
ALL_SENSORS,
KEY_ADJUSTMENT,
KEY_INJECTION,
KEY_MAG,
KEY_OMIE,
KEY_PVPC,
TARIFFS,
UTC_TZ,
)
from aiopvpc.ha_helpers import get_enabled_sensor_keys, make_sensor_unique_id
from aiopvpc.pvpc_data import BadApiTokenAuthError, PVPCData
from tests.conftest import check_num_datapoints, MockAsyncSession, run_h_step
def test_sensor_unique_ids():
all_sensor_keys = get_enabled_sensor_keys(
using_private_api=False, disabled_sensor_ids=[]
)
assert all_sensor_keys == {KEY_PVPC}
all_sensor_keys = get_enabled_sensor_keys(
using_private_api=True, disabled_sensor_ids=[]
)
assert sorted(all_sensor_keys) == sorted(ALL_SENSORS)
counter_combs = 0
unique_ids = set()
for tariff in TARIFFS:
for key in all_sensor_keys:
unique_ids.add(
make_sensor_unique_id(config_entry_id=tariff, sensor_key=key)
)
counter_combs += 1
assert counter_combs == len(unique_ids)
assert not get_enabled_sensor_keys(
using_private_api=True, disabled_sensor_ids=list(unique_ids)
)
@pytest.mark.asyncio
async def test_disable_sensors():
start = datetime(2024, 3, 9, 19, tzinfo=UTC_TZ)
mock_session = MockAsyncSession()
sensor_keys = ALL_SENSORS
assert len(sensor_keys) == 5
pvpc_data = PVPCData(
session=mock_session,
tariff="2.0TD",
data_source="esios",
api_token="test-token",
sensor_keys=sensor_keys,
)
api_data = None
start, api_data = await run_h_step(mock_session, pvpc_data, api_data, start)
assert mock_session.call_count == 10
check_num_datapoints(api_data, sensor_keys, 24)
pvpc_data.update_active_sensors(KEY_PVPC, enabled=False)
pvpc_data.update_active_sensors(KEY_OMIE, enabled=False)
pvpc_data.update_active_sensors(KEY_MAG, enabled=False)
pvpc_data.update_active_sensors(KEY_MAG, enabled=False)
pvpc_data.update_active_sensors(KEY_ADJUSTMENT, enabled=False)
start, api_data = await run_h_step(mock_session, pvpc_data, api_data, start)
assert mock_session.call_count == 11
check_num_datapoints(api_data, sensor_keys, 24)
logging.error(api_data.sensors.keys())
pvpc_data.update_active_sensors(KEY_INJECTION, enabled=False)
start, api_data = await run_h_step(mock_session, pvpc_data, api_data, start)
assert mock_session.call_count == 11
check_num_datapoints(api_data, sensor_keys, 24)
start, api_data = await run_h_step(mock_session, pvpc_data, api_data, start)
assert mock_session.call_count == 11
check_num_datapoints(api_data, sensor_keys, 24)
start, api_data = await run_h_step(
mock_session, pvpc_data, api_data, start, should_fail=True
)
assert mock_session.call_count == 11
# check_num_datapoints(api_data, sensor_keys, 0)
pvpc_data.update_active_sensors(KEY_INJECTION, enabled=True)
pvpc_data.update_active_sensors(KEY_PVPC, enabled=True)
start, api_data = await run_h_step(
mock_session, pvpc_data, api_data, start, should_fail=True
)
assert mock_session.call_count == 13
pvpc_data.update_active_sensors(KEY_INJECTION, enabled=True)
pvpc_data.update_active_sensors(KEY_PVPC, enabled=True)
pvpc_data.update_active_sensors(KEY_MAG, enabled=True)
pvpc_data.update_active_sensors(KEY_OMIE, enabled=True)
await run_h_step(mock_session, pvpc_data, api_data, start, should_fail=True)
assert mock_session.call_count == 17
@pytest.mark.asyncio
async def test_check_api_token():
start = datetime(2024, 3, 9, 19, tzinfo=UTC_TZ)
mock_session = MockAsyncSession(status=401)
pvpc_data = PVPCData(session=mock_session)
token_ok = await pvpc_data.check_api_token(start, "bad_token")
assert not token_ok
assert mock_session.call_count == 1
with pytest.raises(BadApiTokenAuthError):
await pvpc_data.async_update_all(None, start)
assert mock_session.call_count == 2
mock_session_ok = MockAsyncSession()
pvpc_data_ok = PVPCData(
session=mock_session_ok,
data_source="esios",
api_token="good_token",
)
token_ok = await pvpc_data_ok.check_api_token(start)
assert token_ok
assert mock_session_ok.call_count == 1
|
