Skip to content

Commit f2d1832

Browse files
drc38drc38
committed
tests: create charge point test
1 parent 2b5d7b3 commit f2d1832

File tree

1 file changed

+253
-31
lines changed

1 file changed

+253
-31
lines changed

‎tests/test_charge_point.py

Lines changed: 253 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,49 +1,271 @@
11
"""Implement a test by a simulating a chargepoint."""
22
import asyncio
3-
import logging
3+
from datetime import datetime, timezone # timedelta,
44

5+
from pytest_homeassistant_custom_component.common import MockConfigEntry
56
import websockets
67

7-
from ocpp.v16 import ChargePoint as cp, call
8-
from ocpp.v16.enums import RegistrationStatus
8+
from custom_components.ocpp import async_setup_entry
9+
from custom_components.ocpp.const import DOMAIN
10+
from custom_components.ocpp.enums import ConfigurationKey
11+
from ocpp.routing import on
12+
from ocpp.v16 import ChargePoint as cp, call, call_result
13+
from ocpp.v16.enums import (
14+
Action,
15+
AuthorizationStatus,
16+
AvailabilityStatus,
17+
ConfigurationStatus,
18+
RegistrationStatus,
19+
)
920

10-
logging.basicConfig(level=logging.INFO)
21+
from .const import MOCK_CONFIG_DATA
1122

1223

13-
class ChargePoint(cp):
14-
"""Representation of Charge Point."""
24+
async def test_cms_responses(hass):
25+
"""Test central system responses to a charger."""
26+
# Create a mock entry so we don't have to go through config flow
27+
config_entry = MockConfigEntry(
28+
domain=DOMAIN, data=MOCK_CONFIG_DATA, entry_id="test"
29+
)
30+
assert await async_setup_entry(hass, config_entry)
31+
await hass.async_block_till_done()
1532

16-
async def send_boot_notification(self):
17-
"""Send a boot notification."""
18-
request = call.BootNotificationPayload(
19-
charge_point_model="Optimus", charge_point_vendor="The Mobility House"
33+
async with websockets.connect(
34+
"ws://localhost:9000/CP_1", subprotocols=["ocpp1.6"]
35+
) as ws:
36+
37+
cp = ChargePoint("CP_1", ws)
38+
asyncio.gather(
39+
cp.start(),
40+
cp.send_boot_notification(),
41+
cp.send_start_transaction(),
42+
cp.send_meter_data(),
43+
cp.send_stop_transaction(),
2044
)
2145

22-
response = await self.call(request)
2346

24-
if response.status == RegistrationStatus.accepted:
25-
print("Connected to central system.")
47+
class ChargePoint(cp):
48+
"""Representation of real client Charge Point."""
2649

50+
def __init__(self, id, connection, response_timeout=30):
51+
"""Init extra variables for testing."""
52+
super().__init__(id, connection)
53+
self._transactionId = 0
2754

28-
async def main():
29-
"""Start at main entry point."""
30-
async with websockets.connect(
31-
"ws://localhost:9000/CP_1", subprotocols=["ocpp1.6"]
32-
) as ws:
55+
@on(Action.GetConfiguration)
56+
def on_get_configuration(self, key, **kwargs):
57+
"""Handle a get configuration requests."""
58+
if key == ConfigurationKey.supported_feature_profiles.value:
59+
return call_result.GetConfigurationPayload(
60+
configuration_key=["Core", "FirmwareManagement", "SmartCharging"]
61+
)
62+
if key == ConfigurationKey.heartbeat_interval.value:
63+
return call_result.GetConfigurationPayload(configuration_key="300")
64+
if key == ConfigurationKey.number_of_connectors.value:
65+
return call_result.GetConfigurationPayload(configuration_key="1")
66+
if key == ConfigurationKey.web_socket_ping_interval.value:
67+
return call_result.GetConfigurationPayload(configuration_key="60")
68+
if key == ConfigurationKey.meter_values_sampled_data.value:
69+
return call_result.GetConfigurationPayload(
70+
configuration_key="Energy.Reactive.Import.Register"
71+
)
72+
if key == ConfigurationKey.meter_value_sample_interval.value:
73+
return call_result.GetConfigurationPayload(configuration_key="60")
74+
if key == ConfigurationKey.charging_schedule_allowed_charging_rate_unit.value:
75+
return call_result.GetConfigurationPayload(configuration_key="current")
76+
if key == ConfigurationKey.authorize_remote_tx_requests.value:
77+
return call_result.GetConfigurationPayload(configuration_key="false")
3378

34-
cp = ChargePoint("CP_1", ws)
79+
@on(Action.ChangeConfiguration)
80+
def on_change_configuration(self, **kwargs):
81+
"""Handle a get configuration requests."""
82+
return call_result.GetConfigurationPayload(ConfigurationStatus.accepted)
83+
84+
@on(Action.ChangeAvailability)
85+
def on_change_availability(self, **kwargs):
86+
"""Handle change availability requests."""
87+
return call_result.ChangeAvailabilityPayload(AvailabilityStatus.accepted)
88+
89+
async def send_boot_notification(self):
90+
"""Send a boot notification."""
91+
request = call.BootNotificationPayload(
92+
charge_point_model="Optimus", charge_point_vendor="The Mobility House"
93+
)
94+
resp = await self.call(request)
95+
assert resp.status == RegistrationStatus.accepted
3596

36-
await asyncio.gather(cp.start(), cp.send_boot_notification())
97+
async def send_start_transaction(self):
98+
"""Send a start transaction notification."""
99+
request = call.StartTransactionPayload(
100+
connector_id=1,
101+
id_tag="test_cp",
102+
meter_start=12345,
103+
timestamp=datetime.now(tz=timezone.utc).isoformat,
104+
)
105+
resp = await self.call(request)
106+
self._transactionId = resp.transaction_id
107+
assert resp.id_tag_info["status"] == AuthorizationStatus.accepted.value
37108

109+
async def send_meter_data(self):
110+
"""Send meter data notification."""
111+
request = call.MeterValuesPayload(
112+
connector_id=1,
113+
transaction_id=self._transactionId,
114+
meter_value=[
115+
{
116+
"timestamp": "2021-06-21T16:15:09Z",
117+
"sampledValue": [
118+
{
119+
"value": "1305570.000",
120+
"context": "Sample.Periodic",
121+
"measurand": "Energy.Active.Import.Register",
122+
"location": "Outlet",
123+
"unit": "Wh",
124+
},
125+
{
126+
"value": "0.000",
127+
"context": "Sample.Periodic",
128+
"measurand": "Current.Import",
129+
"location": "Outlet",
130+
"unit": "A",
131+
"phase": "L1",
132+
},
133+
{
134+
"value": "0.000",
135+
"context": "Sample.Periodic",
136+
"measurand": "Current.Import",
137+
"location": "Outlet",
138+
"unit": "A",
139+
"phase": "L2",
140+
},
141+
{
142+
"value": "0.000",
143+
"context": "Sample.Periodic",
144+
"measurand": "Current.Import",
145+
"location": "Outlet",
146+
"unit": "A",
147+
"phase": "L3",
148+
},
149+
{
150+
"value": "16.000",
151+
"context": "Sample.Periodic",
152+
"measurand": "Current.Offered",
153+
"location": "Outlet",
154+
"unit": "A",
155+
},
156+
{
157+
"value": "50.010",
158+
"context": "Sample.Periodic",
159+
"measurand": "Frequency",
160+
"location": "Outlet",
161+
},
162+
{
163+
"value": "0.000",
164+
"context": "Sample.Periodic",
165+
"measurand": "Power.Active.Import",
166+
"location": "Outlet",
167+
"unit": "W",
168+
},
169+
{
170+
"value": "0.000",
171+
"context": "Sample.Periodic",
172+
"measurand": "Power.Active.Import",
173+
"location": "Outlet",
174+
"unit": "W",
175+
"phase": "L1",
176+
},
177+
{
178+
"value": "0.000",
179+
"context": "Sample.Periodic",
180+
"measurand": "Power.Active.Import",
181+
"location": "Outlet",
182+
"unit": "W",
183+
"phase": "L2",
184+
},
185+
{
186+
"value": "0.000",
187+
"context": "Sample.Periodic",
188+
"measurand": "Power.Active.Import",
189+
"location": "Outlet",
190+
"unit": "W",
191+
"phase": "L3",
192+
},
193+
{
194+
"value": "0.000",
195+
"context": "Sample.Periodic",
196+
"measurand": "Power.Factor",
197+
"location": "Outlet",
198+
},
199+
{
200+
"value": "38.500",
201+
"context": "Sample.Periodic",
202+
"measurand": "Temperature",
203+
"location": "Body",
204+
"unit": "Celsius",
205+
},
206+
{
207+
"value": "228.000",
208+
"context": "Sample.Periodic",
209+
"measurand": "Voltage",
210+
"location": "Outlet",
211+
"unit": "V",
212+
"phase": "L1-N",
213+
},
214+
{
215+
"value": "227.000",
216+
"context": "Sample.Periodic",
217+
"measurand": "Voltage",
218+
"location": "Outlet",
219+
"unit": "V",
220+
"phase": "L2-N",
221+
},
222+
{
223+
"value": "229.300",
224+
"context": "Sample.Periodic",
225+
"measurand": "Voltage",
226+
"location": "Outlet",
227+
"unit": "V",
228+
"phase": "L3-N",
229+
},
230+
{
231+
"value": "395.900",
232+
"context": "Sample.Periodic",
233+
"measurand": "Voltage",
234+
"location": "Outlet",
235+
"unit": "V",
236+
"phase": "L1-L2",
237+
},
238+
{
239+
"value": "396.300",
240+
"context": "Sample.Periodic",
241+
"measurand": "Voltage",
242+
"location": "Outlet",
243+
"unit": "V",
244+
"phase": "L2-L3",
245+
},
246+
{
247+
"value": "398.900",
248+
"context": "Sample.Periodic",
249+
"measurand": "Voltage",
250+
"location": "Outlet",
251+
"unit": "V",
252+
"phase": "L3-L1",
253+
},
254+
],
255+
}
256+
],
257+
)
258+
await self.call(request)
259+
# check an error is not thrown
38260

39-
if __name__ == "__main__":
40-
try:
41-
# asyncio.run() is used when running this example with Python 3.7 and
42-
# higher.
43-
asyncio.run(main())
44-
except AttributeError:
45-
# For Python 3.6 a bit more code is required to run the main() task on
46-
# an event loop.
47-
loop = asyncio.get_event_loop()
48-
loop.run_until_complete(main())
49-
loop.close()
261+
async def send_stop_transaction(self):
262+
"""Send a stop transaction notification."""
263+
request = call.StopTransactionPayload(
264+
meter_stop=54321,
265+
timestamp=datetime.now(tz=timezone.utc).isoformat,
266+
transaction_id=self._transactionId,
267+
reason="EVDisconnected",
268+
id_tag="test_cp",
269+
)
270+
resp = await self.call(request)
271+
assert resp.id_tag_info["status"] == AuthorizationStatus.accepted.value

0 commit comments

Comments
 (0)