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
|
import pytest
from openleadr import OpenADRClient
from openleadr import enums
def test_trailing_slash_on_vtn_url():
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost/')
assert client.vtn_url == 'http://localhost'
def test_wrong_handler_supplied(caplog):
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost')
client.add_handler('non_existant', print)
assert ("'handler' must be either on_event or on_update_event") in [rec.message for rec in caplog.records]
def test_invalid_report_name(caplog):
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost')
with pytest.raises(ValueError):
client.add_report(callback=print,
resource_id='myresource',
measurement='voltage',
report_name='non_existant')
# assert (f"non_existant is not a valid report_name. Valid options are "
# f"{', '.join(enums.REPORT_NAME.values)}",
# " or any name starting with 'x-'.") in [rec.message for rec in caplog.records]
def test_invalid_reading_type():
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost')
with pytest.raises(ValueError):
client.add_report(callback=print,
resource_id='myresource',
measurement='voltage',
reading_type='non_existant')
# assert (f"non_existant is not a valid reading_type. Valid options are "
# f"{', '.join(enums.READING_TYPE.values)}",
# " or any name starting with 'x-'.") in [rec.message for rec in caplog.records]
def test_invalid_report_type():
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost')
with pytest.raises(ValueError):
client.add_report(callback=print,
resource_id='myresource',
measurement='voltage',
report_type='non_existant')
# assert (f"non_existant is not a valid report_type. Valid options are "
# f"{', '.join(enums.REPORT_TYPE.values)}",
# " or any name starting with 'x-'.") in [rec.message for rec in caplog.records]
def test_invalid_data_collection_mode():
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost')
with pytest.raises(ValueError):
client.add_report(callback=print,
resource_id='myresource',
measurement='voltage',
data_collection_mode='non_existant')
# assert ("The data_collection_mode should be 'incremental' or 'full'.") in [rec.message for rec in caplog.records]
def test_invalid_scale():
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost')
with pytest.raises(ValueError):
client.add_report(callback=print,
resource_id='myresource',
measurement='voltage',
scale='non_existant')
def test_add_report_without_specifier_id():
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost')
client.add_report(callback=print,
resource_id='myresource1',
measurement='voltage')
client.add_report(callback=print,
resource_id='myresource2',
measurement='voltage')
assert len(client.reports) == 1
async def wrong_sig(param1):
pass
def test_add_report_with_invalid_callback_signature():
client = OpenADRClient(ven_name='myven', vtn_url='http://localhost')
with pytest.raises(TypeError):
client.add_report(callback=wrong_sig,
data_collection_mode='full',
resource_id='myresource1',
measurement='voltage')
|
