1
1
from unittest .mock import patch
2
2
3
3
import pytest
4
- from asynctest import CoroutineMock
5
4
from google .protobuf .json_format import MessageToDict
6
5
from httpx import AsyncClient , Client , Response
7
6
7
+ try :
8
+ from unittest .mock import AsyncMock
9
+ except ImportError :
10
+ from asynctest import CoroutineMock as AsyncMock
11
+
8
12
from devolo_plc_api .device_api .deviceapi import DeviceApi
9
13
from devolo_plc_api .device_api .devolo_idl_proto_deviceapi_ledsettings_pb2 import LedSettingsGet , LedSettingsSetResponse
10
14
from devolo_plc_api .device_api .devolo_idl_proto_deviceapi_wifinetwork_pb2 import (
@@ -31,8 +35,8 @@ def test_unsupported_feature(self, request):
31
35
async def test_async_get_led_setting (self , request ):
32
36
led_setting_get = LedSettingsGet ()
33
37
34
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = CoroutineMock (return_value = Response )), \
35
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = led_setting_get .SerializeToString ())):
38
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = AsyncMock (return_value = Response )), \
39
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = led_setting_get .SerializeToString ())):
36
40
device_api = DeviceApi (request .cls .ip ,
37
41
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
38
42
AsyncClient (),
@@ -68,8 +72,8 @@ def test_get_led_setting(self, request):
68
72
async def test_async_set_led_setting (self , request ):
69
73
led_setting_set = LedSettingsSetResponse ()
70
74
71
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_post" , new = CoroutineMock (return_value = Response )), \
72
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = led_setting_set .SerializeToString ())):
75
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_post" , new = AsyncMock (return_value = Response )), \
76
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = led_setting_set .SerializeToString ())):
73
77
device_api = DeviceApi (request .cls .ip ,
74
78
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
75
79
AsyncClient (),
@@ -99,8 +103,8 @@ def test_set_led_setting(self, request):
99
103
async def test_async_check_firmware_available (self , request ):
100
104
firmware_available = UpdateFirmwareCheck ()
101
105
102
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = CoroutineMock (return_value = Response )), \
103
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = firmware_available .SerializeToString ())):
106
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = AsyncMock (return_value = Response )), \
107
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = firmware_available .SerializeToString ())):
104
108
device_api = DeviceApi (request .cls .ip ,
105
109
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
106
110
AsyncClient (),
@@ -136,8 +140,8 @@ def test_check_firmware_available(self, request):
136
140
async def test_async_start_firmware_update (self , request ):
137
141
firmware_update = UpdateFirmwareStart ()
138
142
139
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = CoroutineMock (return_value = Response )), \
140
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = firmware_update .SerializeToString ())):
143
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = AsyncMock (return_value = Response )), \
144
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = firmware_update .SerializeToString ())):
141
145
device_api = DeviceApi (request .cls .ip ,
142
146
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
143
147
AsyncClient (),
@@ -167,8 +171,8 @@ def test_start_firmware_update(self, request):
167
171
async def test_async_get_wifi_connected_station (self , request ):
168
172
wifi_connected_stations_get = WifiConnectedStationsGet ()
169
173
170
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = CoroutineMock (return_value = Response )), \
171
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = wifi_connected_stations_get .SerializeToString ())):
174
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = AsyncMock (return_value = Response )), \
175
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = wifi_connected_stations_get .SerializeToString ())):
172
176
device_api = DeviceApi (request .cls .ip ,
173
177
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
174
178
AsyncClient (),
@@ -204,8 +208,8 @@ def test_get_wifi_connected_station(self, request):
204
208
async def test_async_get_wifi_guest_access (self , request ):
205
209
wifi_guest_access_get = WifiGuestAccessGet ()
206
210
207
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = CoroutineMock (return_value = Response )), \
208
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = wifi_guest_access_get .SerializeToString ())):
211
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = AsyncMock (return_value = Response )), \
212
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = wifi_guest_access_get .SerializeToString ())):
209
213
device_api = DeviceApi (request .cls .ip ,
210
214
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
211
215
AsyncClient (),
@@ -241,8 +245,8 @@ def test_get_wifi_guest_access(self, request):
241
245
async def test_async_set_wifi_guest_access (self , request ):
242
246
wifi_guest_access_set = WifiGuestAccessSetResponse ()
243
247
244
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_post" , new = CoroutineMock (return_value = Response )), \
245
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = wifi_guest_access_set .SerializeToString ())):
248
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_post" , new = AsyncMock (return_value = Response )), \
249
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = wifi_guest_access_set .SerializeToString ())):
246
250
device_api = DeviceApi (request .cls .ip ,
247
251
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
248
252
AsyncClient (),
@@ -272,8 +276,8 @@ def test_set_wifi_guest_access(self, request):
272
276
async def test_async_get_wifi_neighbor_access_points (self , request ):
273
277
wifi_neighbor_accesspoints_get = WifiNeighborAPsGet ()
274
278
275
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = CoroutineMock (return_value = Response )), \
276
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = wifi_neighbor_accesspoints_get .SerializeToString ())):
279
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = AsyncMock (return_value = Response )), \
280
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = wifi_neighbor_accesspoints_get .SerializeToString ())):
277
281
device_api = DeviceApi (request .cls .ip ,
278
282
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
279
283
AsyncClient (),
@@ -309,8 +313,8 @@ def test_get_wifi_neighbor_access_points(self, request):
309
313
async def test_async_get_wifi_repeated_access_points (self , request ):
310
314
wifi_repeated_accesspoints_get = WifiRepeatedAPsGet ()
311
315
312
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = CoroutineMock (return_value = Response )), \
313
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = wifi_repeated_accesspoints_get .SerializeToString ())):
316
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = AsyncMock (return_value = Response )), \
317
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = wifi_repeated_accesspoints_get .SerializeToString ())):
314
318
device_api = DeviceApi (request .cls .ip ,
315
319
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
316
320
AsyncClient (),
@@ -346,8 +350,8 @@ def test_get_wifi_repeated_access_points(self, request):
346
350
async def test_async_start_wps (self , request ):
347
351
wps = WifiWpsPbcStart ()
348
352
349
- with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = CoroutineMock (return_value = Response )), \
350
- patch ("httpx.Response.aread" , new = CoroutineMock (return_value = wps .SerializeToString ())):
353
+ with patch ("devolo_plc_api.clients.protobuf.Protobuf._async_get" , new = AsyncMock (return_value = Response )), \
354
+ patch ("httpx.Response.aread" , new = AsyncMock (return_value = wps .SerializeToString ())):
351
355
device_api = DeviceApi (request .cls .ip ,
352
356
request .cls .device_info ['_dvl-deviceapi._tcp.local.' ]['Port' ],
353
357
AsyncClient (),
0 commit comments