diff --git a/.github/workflows/pythonpublish.yml b/.github/workflows/pythonpublish.yml new file mode 100644 index 0000000..a2f1288 --- /dev/null +++ b/.github/workflows/pythonpublish.yml @@ -0,0 +1,26 @@ +name: Upload Python Package + +on: + release: + types: [created] + +jobs: + deploy: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + - name: Set up Python + uses: actions/setup-python@v1 + with: + python-version: "3.7" + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install setuptools wheel twine + - name: Build and publish + env: + TWINE_USERNAME: ${{ secrets.PYPI_USERNAME }} + TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }} + run: | + python setup.py sdist bdist_wheel + twine upload dist/* diff --git a/README.md b/README.md index d81db22..b6a10a0 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@ # devolo PLC API [![GitHub Workflow Status](https://img.shields.io/github/workflow/status/2Fake/devolo_plc_api/Python%20package)](https://github.com/2Fake/devolo_plc_api/actions?query=workflow%3A%22Python+package%22) +[![PyPI - Downloads](https://img.shields.io/pypi/dd/devolo-plc-api)](https://pypi.org/project/devolo-plc-api/) [![Code Climate maintainability](https://img.shields.io/codeclimate/maintainability/2Fake/devolo_plc_api)](https://codeclimate.com/github/2Fake/devolo_plc_api) [![Coverage Status](https://coveralls.io/repos/github/2Fake/devolo_plc_api/badge.svg?branch=development)](https://coveralls.io/github/2Fake/devolo_plc_api?branch=development) @@ -59,4 +60,23 @@ python setup.py test ## Usage -All features we currently support are shown in our [example.py](https://github.com/2Fake/devolo_plc_api/blob/master/example.py) +All features we currently support are shown in our examples. If you want to use the package asynchronously, please check [example_async.py](https://github.com/2Fake/devolo_plc_api/blob/master/example_async.py). If you want to use it synchronously, please check [example_sync.py](https://github.com/2Fake/devolo_plc_api/blob/master/example_sync.py). + +## Supported device + +The following devolo devices were queried with at least one call to verify functionality: + +* Magic 2 WiFi next +* Magic 2 WiFi 2-1 +* Magic 2 LAN triple +* Magic 2 DinRail +* Magic 2 LAN 1-1 +* Magic 1 WiFi mini +* Magic 1 WiFi 2-1 +* Magic 1 LAN 1-1 +* dLAN 1200+ WiFi ac +* dLAN 550+ Wifi +* dLAN 550 WiFi +* dLAN 500 WiFi + +However, other devices might work, some might have a limited functionality. Also firmware version will matter. If you discover something weird, [we want to know](https://github.com/2Fake/devolo_plc_api/issues). diff --git a/devolo_plc_api/__init__.py b/devolo_plc_api/__init__.py index 3dc1f76..d3ec452 100644 --- a/devolo_plc_api/__init__.py +++ b/devolo_plc_api/__init__.py @@ -1 +1 @@ -__version__ = "0.1.0" +__version__ = "0.2.0" diff --git a/devolo_plc_api/device.py b/devolo_plc_api/device.py index a94b7ad..af2fc49 100644 --- a/devolo_plc_api/device.py +++ b/devolo_plc_api/device.py @@ -83,6 +83,7 @@ async def _get_device_info(self): self.product = self._info[service_type].get("Product", "") self.device = DeviceApi(ip=self.ip, + port=self._info[service_type]['Port'], session=self._session, path=self._info[service_type]['Path'], version=self._info[service_type]['Version'], @@ -101,6 +102,7 @@ async def _get_plcnet_info(self): self.technology = self._info[service_type].get("PlcTechnology", "") self.plcnet = PlcNetApi(ip=self.ip, + port=self._info[service_type]['Port'], session=self._session, path=self._info[service_type]['Path'], version=self._info[service_type]['Version'], @@ -121,6 +123,8 @@ def _state_change(self, zeroconf: Zeroconf, service_type: str, name: str, state_ self.ip in [socket.inet_ntoa(address) for address in service_info.addresses]: self._logger.debug(f"Adding service info of {service_type}") + self._info[service_type]['Port'] = service_info.port + # The answer is a byte string, that concatenates key-value pairs with their length as two byte hex value. total_length = len(service_info.text) offset = 0 diff --git a/devolo_plc_api/device_api/deviceapi.py b/devolo_plc_api/device_api/deviceapi.py index b9a18a7..00f3b36 100644 --- a/devolo_plc_api/device_api/deviceapi.py +++ b/devolo_plc_api/device_api/deviceapi.py @@ -7,6 +7,7 @@ from ..exceptions.feature import FeatureNotSupported from . import devolo_idl_proto_deviceapi_wifinetwork_pb2 from . import devolo_idl_proto_deviceapi_ledsettings_pb2 +from . import devolo_idl_proto_deviceapi_updatefirmware_pb2 class DeviceApi(Protobuf): @@ -20,13 +21,13 @@ class DeviceApi(Protobuf): :param features: Feature, the device has """ - def __init__(self, ip: str, session: Client, path: str, version: str, features: str, password: str): + def __init__(self, ip: str, port: int, session: Client, path: str, version: str, features: str, password: str): self._ip = ip - self._port = 14791 + self._port = port self._session = session self._path = path self._version = version - self._features = features.split(",") if features else [] + self._features = features.split(",") if features else ['reset', 'update', 'led', 'intmtg'] self._user = "devolo" self._password = password self._logger = logging.getLogger(self.__class__.__name__) @@ -45,8 +46,12 @@ def wrapper(self, *args, **kwargs): @_feature("led") - async def async_get_led_setting(self): - """ Get LED setting asynchronously. """ + async def async_get_led_setting(self) -> dict: + """ + Get LED setting asynchronously. This feature only works on devices, that announce the led feature. + + return: LED settings + """ led_setting = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsGet() response = await self._async_get("LedSettingsGet") led_setting.FromString(await response.aread()) @@ -54,7 +59,11 @@ async def async_get_led_setting(self): @_feature("led") def get_led_setting(self): - """ Get LED setting synchronously. """ + """ + Get LED setting synchronously. This feature only works on devices, that announce the led feature. + + return: LED settings + """ led_setting = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsGet() response = self._get("LedSettingsGet") led_setting.FromString(response.read()) @@ -62,7 +71,12 @@ def get_led_setting(self): @_feature("led") async def async_set_led_setting(self, enable: bool) -> bool: - """ Set LED setting asynchronously. """ + """ + Set LED setting asynchronously. This feature only works on devices, that announce the led feature. + + :param enable: True to enable the LEDs, False to disable the LEDs + :return: True, if LED state was successfully changed, otherwise False + """ led_setting = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsSet() led_setting.state = int(not enable) query = await self._async_post("LedSettingsSet", data=led_setting.SerializeToString()) @@ -72,7 +86,12 @@ async def async_set_led_setting(self, enable: bool) -> bool: @_feature("led") def set_led_setting(self, enable: bool) -> bool: - """ Set LED setting synchronously. """ + """ + Set LED setting synchronously. This feature only works on devices, that announce the led feature. + + :param enable: True to enable the LEDs, False to disable the LEDs + :return: True, if LED state was successfully changed, otherwise False + """ led_setting = devolo_idl_proto_deviceapi_ledsettings_pb2.LedSettingsSet() led_setting.state = int(not enable) query = self._post("LedSettingsSet", data=led_setting.SerializeToString()) @@ -80,9 +99,66 @@ def set_led_setting(self, enable: bool) -> bool: response.FromString(query.read()) return bool(not response.result) + + @_feature("update") + async def async_check_firmware_available(self) -> dict: + """ + Check asynchronously, if a firmware update is available for the device. + + :return: Result and new firmware version, if newer one is available + """ + update_firmware_check = devolo_idl_proto_deviceapi_updatefirmware_pb2.UpdateFirmwareCheck() + response = await self._async_get("UpdateFirmwareCheck") + update_firmware_check.ParseFromString(await response.aread()) + return self._message_to_dict(update_firmware_check) + + @_feature("update") + def check_firmware_available(self) -> dict: + """ + Check synchronously, if a firmware update is available for the device. + + :return: Result and new firmware version, if newer one is available + """ + update_firmware_check = devolo_idl_proto_deviceapi_updatefirmware_pb2.UpdateFirmwareCheck() + response = self._get("UpdateFirmwareCheck") + update_firmware_check.ParseFromString(response.read()) + return self._message_to_dict(update_firmware_check) + + @_feature("update") + async def async_start_firmware_update(self) -> bool: + """ + Start firmware update asynchronously, if a firmware update is available for the device. Important: The response does + not tell you anything about the success of the update itself. + + :return: True, if the firmware update was started, False if there is no update + """ + update_firmware = devolo_idl_proto_deviceapi_updatefirmware_pb2.UpdateFirmwareStart() + response = await self._async_get("UpdateFirmwareStart") + update_firmware.FromString(await response.aread()) + return bool(not update_firmware.result) + + @_feature("update") + def start_firmware_update(self) -> bool: + """ + Start firmware update synchronously, if a firmware update is available for the device. Important: The response does + not tell you anything about the success of the update itself. + + :return: True, if the firmware update was started, False if there is no update + """ + update_firmware = devolo_idl_proto_deviceapi_updatefirmware_pb2.UpdateFirmwareStart() + response = self._get("UpdateFirmwareStart") + update_firmware.FromString(response.read()) + return bool(not update_firmware.result) + + @_feature("wifi1") async def async_get_wifi_connected_station(self) -> dict: - """ Get wifi stations connected to the device asynchronously. """ + """ + Get wifi stations connected to the device asynchronously. This feature only works on devices, that announce the wifi1 + feature. + + :return: All connected wifi stations including connection rate data + """ wifi_connected_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiConnectedStationsGet() response = await self._async_get("WifiConnectedStationsGet") wifi_connected_proto.ParseFromString(await response.aread()) @@ -90,7 +166,12 @@ async def async_get_wifi_connected_station(self) -> dict: @_feature("wifi1") def get_wifi_connected_station(self) -> dict: - """ Get wifi stations connected to the device synchronously. """ + """ + Get wifi stations connected to the device synchronously. This feature only works on devices, that announce the wifi1 + feature. + + :return: All connected wifi stations including connection rate data + """ wifi_connected_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiConnectedStationsGet() response = self._get("WifiConnectedStationsGet") wifi_connected_proto.ParseFromString(response.read()) @@ -98,7 +179,12 @@ def get_wifi_connected_station(self) -> dict: @_feature("wifi1") async def async_get_wifi_guest_access(self) -> dict: - """ Get details about wifi guest access asynchronously. """ + """ + Get details about wifi guest access asynchronously. This feature only works on devices, that announce the wifi1 + feature. + + :return: Details about the wifi guest access + """ self._logger.debug("Getting wifi guest access") wifi_guest_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiGuestAccessGet() response = await self._async_get("WifiGuestAccessGet") @@ -107,7 +193,12 @@ async def async_get_wifi_guest_access(self) -> dict: @_feature("wifi1") def get_wifi_guest_access(self) -> dict: - """ Get details about wifi guest access synchronously. """ + """ + Get details about wifi guest access synchronously. This feature only works on devices, that announce the wifi1 + feature. + + :return: Details about the wifi guest access + """ self._logger.debug("Getting wifi guest access") wifi_guest_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiGuestAccessGet() response = self._get("WifiGuestAccessGet") @@ -116,7 +207,12 @@ def get_wifi_guest_access(self) -> dict: @_feature("wifi1") async def async_set_wifi_guest_access(self, enable: bool) -> bool: - """ Enable wifi guest access asynchronously. """ + """ + Enable wifi guest access asynchronously. This feature only works on devices, that announce the wifi1 feature. + + :param enable: True to enable, False to disable wifi guest access + :return: True, if the state of the wifi guest access was successfully changed, otherwise False + """ wifi_guest_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiGuestAccessSet() wifi_guest_proto.enable = enable query = await self._async_post("WifiGuestAccessSet", data=wifi_guest_proto.SerializeToString()) @@ -126,7 +222,12 @@ async def async_set_wifi_guest_access(self, enable: bool) -> bool: @_feature("wifi1") def set_wifi_guest_access(self, enable: bool) -> bool: - """ Enable wifi guest access synchronously. """ + """ + Enable wifi guest access synchronously. This feature only works on devices, that announce the wifi1 feature. + + :param enable: True to enable, False to disable wifi guest access + :return: True, if the state of the wifi guest access was successfully changed, otherwise False + """ wifi_guest_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiGuestAccessSet() wifi_guest_proto.enable = enable query = self._post("WifiGuestAccessSet", data=wifi_guest_proto.SerializeToString()) @@ -136,7 +237,12 @@ def set_wifi_guest_access(self, enable: bool) -> bool: @_feature("wifi1") async def async_get_wifi_neighbor_access_points(self) -> dict: - """ Get wifi access point in the neighborhood asynchronously. """ + """ + Get wifi access point in the neighborhood asynchronously. This feature only works on devices, that announce the wifi1 + feature. + + :return: Visible access points in the neighborhood including connection rate data + """ wifi_neighbor_aps = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiNeighborAPsGet() response = await self._async_get("WifiNeighborAPsGet", timeout=15.0) wifi_neighbor_aps.ParseFromString(await response.aread()) @@ -144,7 +250,12 @@ async def async_get_wifi_neighbor_access_points(self) -> dict: @_feature("wifi1") def get_wifi_neighbor_access_points(self) -> dict: - """ Get wifi access point in the neighborhood synchronously. """ + """ + Get wifi access point in the neighborhood synchronously. This feature only works on devices, that announce the wifi1 + feature. + + :return: Visible access points in the neighborhood including connection rate data + """ wifi_neighbor_aps = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiNeighborAPsGet() response = self._get("WifiNeighborAPsGet", timeout=15.0) wifi_neighbor_aps.ParseFromString(response.read()) @@ -152,6 +263,12 @@ def get_wifi_neighbor_access_points(self) -> dict: @_feature("wifi1") async def async_get_wifi_repeated_access_points(self): + """ + Get repeated wifi access point asynchronously. This feature only works on repeater devices, that announce the wifi1 + feature. + + :return: Repeated access points in the neighborhood including connection rate data + """ wifi_connected_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiRepeatedAPsGet() response = await self._async_get("WifiRepeatedAPsGet") wifi_connected_proto.ParseFromString(await response.aread()) @@ -159,7 +276,37 @@ async def async_get_wifi_repeated_access_points(self): @_feature("wifi1") def get_wifi_repeated_access_points(self): + """ + Get repeated wifi access point synchronously. This feature only works on repeater devices, that announce the wifi1 + feature. + + :return: Repeated access points in the neighborhood including connection rate data + """ wifi_connected_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiRepeatedAPsGet() response = self._get("WifiRepeatedAPsGet") wifi_connected_proto.ParseFromString(response.read()) return self._message_to_dict(wifi_connected_proto) + + @_feature("wifi1") + async def async_start_wps(self): + """ + Start WPS push button configuration. + + :return: True, if the WPS was successfully started, otherwise False + """ + wps_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiWpsPbcStart() + response = await self._async_get("WifiWpsPbcStart") + wps_proto.FromString(await response.aread()) + return bool(not wps_proto.result) + + @_feature("wifi1") + def start_wps(self): + """ + Start WPS push button configuration. + + :return: True, if the WPS was successfully started, otherwise False + """ + wps_proto = devolo_idl_proto_deviceapi_wifinetwork_pb2.WifiWpsPbcStart() + response = self._get("WifiWpsPbcStart") + wps_proto.FromString(response.read()) + return bool(not wps_proto.result) diff --git a/devolo_plc_api/device_api/devolo_idl_proto_deviceapi_updatefirmware_pb2.py b/devolo_plc_api/device_api/devolo_idl_proto_deviceapi_updatefirmware_pb2.py new file mode 100644 index 0000000..639bf2b --- /dev/null +++ b/devolo_plc_api/device_api/devolo_idl_proto_deviceapi_updatefirmware_pb2.py @@ -0,0 +1,174 @@ +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: updatefirmware.proto + +import sys +_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1')) +from google.protobuf import descriptor as _descriptor +from google.protobuf import message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor.FileDescriptor( + name='updatefirmware.proto', + package='device.api', + syntax='proto3', + serialized_options=_b('\n\006deviceB\006update'), + serialized_pb=_b('\n\x14updatefirmware.proto\x12\ndevice.api\"\xb9\x01\n\x13UpdateFirmwareCheck\x12\x36\n\x06result\x18\x01 \x01(\x0e\x32&.device.api.UpdateFirmwareCheck.Result\x12\x1c\n\x14new_firmware_version\x18\x02 \x01(\t\"L\n\x06Result\x12\x14\n\x10UPDATE_AVAILABLE\x10\x00\x12\x18\n\x14UPDATE_NOT_AVAILABLE\x10\x01\x12\x12\n\rUNKNOWN_ERROR\x10\xff\x01\"\x99\x01\n\x13UpdateFirmwareStart\x12\x36\n\x06result\x18\x01 \x01(\x0e\x32&.device.api.UpdateFirmwareStart.Result\"J\n\x06Result\x12\x12\n\x0eUPDATE_STARTED\x10\x00\x12\x18\n\x14UPDATE_NOT_AVAILABLE\x10\x01\x12\x12\n\rUNKNOWN_ERROR\x10\xff\x01\x42\x10\n\x06\x64\x65viceB\x06updateb\x06proto3') +) + + + +_UPDATEFIRMWARECHECK_RESULT = _descriptor.EnumDescriptor( + name='Result', + full_name='device.api.UpdateFirmwareCheck.Result', + filename=None, + file=DESCRIPTOR, + values=[ + _descriptor.EnumValueDescriptor( + name='UPDATE_AVAILABLE', index=0, number=0, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='UPDATE_NOT_AVAILABLE', index=1, number=1, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='UNKNOWN_ERROR', index=2, number=255, + serialized_options=None, + type=None), + ], + containing_type=None, + serialized_options=None, + serialized_start=146, + serialized_end=222, +) +_sym_db.RegisterEnumDescriptor(_UPDATEFIRMWARECHECK_RESULT) + +_UPDATEFIRMWARESTART_RESULT = _descriptor.EnumDescriptor( + name='Result', + full_name='device.api.UpdateFirmwareStart.Result', + filename=None, + file=DESCRIPTOR, + values=[ + _descriptor.EnumValueDescriptor( + name='UPDATE_STARTED', index=0, number=0, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='UPDATE_NOT_AVAILABLE', index=1, number=1, + serialized_options=None, + type=None), + _descriptor.EnumValueDescriptor( + name='UNKNOWN_ERROR', index=2, number=255, + serialized_options=None, + type=None), + ], + containing_type=None, + serialized_options=None, + serialized_start=304, + serialized_end=378, +) +_sym_db.RegisterEnumDescriptor(_UPDATEFIRMWARESTART_RESULT) + + +_UPDATEFIRMWARECHECK = _descriptor.Descriptor( + name='UpdateFirmwareCheck', + full_name='device.api.UpdateFirmwareCheck', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='result', full_name='device.api.UpdateFirmwareCheck.result', index=0, + number=1, type=14, cpp_type=8, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR), + _descriptor.FieldDescriptor( + name='new_firmware_version', full_name='device.api.UpdateFirmwareCheck.new_firmware_version', index=1, + number=2, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=_b("").decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + _UPDATEFIRMWARECHECK_RESULT, + ], + serialized_options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=37, + serialized_end=222, +) + + +_UPDATEFIRMWARESTART = _descriptor.Descriptor( + name='UpdateFirmwareStart', + full_name='device.api.UpdateFirmwareStart', + filename=None, + file=DESCRIPTOR, + containing_type=None, + fields=[ + _descriptor.FieldDescriptor( + name='result', full_name='device.api.UpdateFirmwareStart.result', index=0, + number=1, type=14, cpp_type=8, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + _UPDATEFIRMWARESTART_RESULT, + ], + serialized_options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=225, + serialized_end=378, +) + +_UPDATEFIRMWARECHECK.fields_by_name['result'].enum_type = _UPDATEFIRMWARECHECK_RESULT +_UPDATEFIRMWARECHECK_RESULT.containing_type = _UPDATEFIRMWARECHECK +_UPDATEFIRMWARESTART.fields_by_name['result'].enum_type = _UPDATEFIRMWARESTART_RESULT +_UPDATEFIRMWARESTART_RESULT.containing_type = _UPDATEFIRMWARESTART +DESCRIPTOR.message_types_by_name['UpdateFirmwareCheck'] = _UPDATEFIRMWARECHECK +DESCRIPTOR.message_types_by_name['UpdateFirmwareStart'] = _UPDATEFIRMWARESTART +_sym_db.RegisterFileDescriptor(DESCRIPTOR) + +UpdateFirmwareCheck = _reflection.GeneratedProtocolMessageType('UpdateFirmwareCheck', (_message.Message,), dict( + DESCRIPTOR = _UPDATEFIRMWARECHECK, + __module__ = 'updatefirmware_pb2' + # @@protoc_insertion_point(class_scope:device.api.UpdateFirmwareCheck) + )) +_sym_db.RegisterMessage(UpdateFirmwareCheck) + +UpdateFirmwareStart = _reflection.GeneratedProtocolMessageType('UpdateFirmwareStart', (_message.Message,), dict( + DESCRIPTOR = _UPDATEFIRMWARESTART, + __module__ = 'updatefirmware_pb2' + # @@protoc_insertion_point(class_scope:device.api.UpdateFirmwareStart) + )) +_sym_db.RegisterMessage(UpdateFirmwareStart) + + +DESCRIPTOR._options = None +# @@protoc_insertion_point(module_scope) diff --git a/devolo_plc_api/plcnet_api/plcnetapi.py b/devolo_plc_api/plcnet_api/plcnetapi.py index bd27bb8..d7784e9 100644 --- a/devolo_plc_api/plcnet_api/plcnetapi.py +++ b/devolo_plc_api/plcnet_api/plcnetapi.py @@ -18,9 +18,9 @@ class PlcNetApi(Protobuf): :param version: Version of the API to use """ - def __init__(self, ip: str, session: Client, path: str, version: str, mac: str): + def __init__(self, ip: str, port: int, session: Client, path: str, version: str, mac: str): self._ip = ip - self._port = 47219 + self._port = port self._session = session self._path = path self._version = version @@ -51,7 +51,7 @@ def get_network_overview(self) -> dict: self._logger.debug("Getting network overview") network_overview = devolo_idl_proto_plcnetapi_getnetworkoverview_pb2.GetNetworkOverview() response = self._get("GetNetworkOverview") - network_overview.ParseFromString(response.content) + network_overview.ParseFromString(response.read()) return self._message_to_dict(network_overview) async def async_identify_device_start(self): @@ -101,7 +101,7 @@ def identify_device_stop(self): """ identify_device = devolo_idl_proto_plcnetapi_identifydevice_pb2.IdentifyDeviceStop() identify_device.mac_address = self._mac - query = self._async_post("IdentifyDeviceStop", data=identify_device.SerializeToString()) + query = self._post("IdentifyDeviceStop", data=identify_device.SerializeToString()) response = devolo_idl_proto_plcnetapi_identifydevice_pb2.IdentifyDeviceResponse() response.FromString(query.read()) return bool(not response.result) @@ -131,7 +131,7 @@ def set_user_device_name(self, name): set_user_name = devolo_idl_proto_plcnetapi_setuserdevicename_pb2.SetUserDeviceName() set_user_name.mac_address = self._mac set_user_name.user_device_name = name - query = self._async_post("SetUserDeviceName", data=set_user_name.SerializeToString(), timeout=10.0) + query = self._post("SetUserDeviceName", data=set_user_name.SerializeToString(), timeout=10.0) response = devolo_idl_proto_plcnetapi_setuserdevicename_pb2.SetUserDeviceNameResponse() response.FromString(query.read()) return bool(not response.result) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 3a5c0b9..e1ee690 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,11 +4,29 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [v0.2.0] - 14.09.2020 + +### Added + +#### Device API + +- Check for firmware updates +- Start firmware updates +- Start WPS + +### Fixed + +- Port from mDNS query is now used +- Get network overview now also works synchroniously +- Sopping identify device now also works synchroniously +- Set user device name now also works synchroniously + ## [v0.1.0] - 28.08.2020 ### Added #### Device API + - Get LED settings - Set LED settings - Get connected wifi clients @@ -18,6 +36,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Get details about master wifi (repeater only) #### PLC API + - Get details about your powerline network - Start and stop identifying your PLC device - Rename your device diff --git a/example.py b/example.py deleted file mode 100644 index d923b8e..0000000 --- a/example.py +++ /dev/null @@ -1,21 +0,0 @@ -import asyncio - -from devolo_plc_api.device import Device - - -# IP of the device to query -IP = "192.168.0.10" - - -async def run(): - async with Device(IP) as dpa: - - # Get details about wifi guest access - print(await dpa.device.get_wifi_guest_access()) - - # Get PLC data rates - print(await dpa.plcnet.get_network_overview()) - - -if __name__ == "__main__": - asyncio.run(run()) diff --git a/example_async.py b/example_async.py new file mode 100644 index 0000000..8b6e31b --- /dev/null +++ b/example_async.py @@ -0,0 +1,126 @@ +import asyncio + +from devolo_plc_api.device import Device + + +# IP of the device to query +IP = "192.168.0.10" + +# Password, if the device has one. It is the same as the Web-UI has. It no password is set, you can remove the password +# parameter or set it to None. +PASSWORD = "super_secret" + + +async def run(): + async with Device(ip=IP, password=PASSWORD) as dpa: + + # Get LED settings of the device. The state might be LED_ON or LED_OFF. + # {'state': 'LED_ON'} + print(await dpa.device.async_get_led_setting()) + + # Set LED settings of the device. Set enable to True to them turn on, to False to turn them off. + # If the state was changed successfully, True is returned, otherwise False. + print("success" if await dpa.device.async_set_led_setting(enable=True) else "failed") + + # Check for new firmware versions + # {'result': 'UPDATE_NOT_AVAILABLE', 'new_firmware_version': ''} + print(await dpa.device.async_check_firmware_available()) + + # Start firmware update, if new version is available. Important: The response does not tell you anything about the + # success of the update itself. + print("update started" if await dpa.device.async_start_firmware_update() else "no update available") + + # Get details of wifi stations connected to the device: MAC address, access point type (main or guest), wifi band and + # connection rates. + # {'connected_stations': + # [ + # {'mac_address': 'AA:BB:CC:DD:EE:FF', + # 'vap_type': 'WIFI_VAP_MAIN_AP', + # 'band': 'WIFI_BAND_5G', + # 'rx_rate': 87800, + # 'tx_rate': 87800} + # ] + # } + print(await dpa.device.async_get_wifi_connected_station()) + + # Get details about wifi guest access: SSID, Wifi key, state (enabled/disabled) and if time limited, the remaining + # duration. + # {'ssid': 'devolo-guest-930', 'key': 'HMANPGBA', 'enabled': False, 'remaining_duration': 0} + print(await dpa.device.async_get_wifi_guest_access()) + + # Enable or disable the wifi guest access. Set enable to True to it turn on, to False to turn it off. Chaning SSID, + # wifi key or duration is currently not supported. If the state was changed successfully, True is returned, otherwise + # False. + print("success" if await dpa.device.async_set_wifi_guest_access(enable=False) else "failed") + + # Get details about other access points in your neighborhood: MAC address, SSID, wifi band, used channel, signal + # strength in DB and a value from 1 to 5, if you would want to map the signal strenght to a signal bars. + # {'neighbor_aps': + # [ + # {'mac_address': 'AA:BB:CC:DD:EE:FF', + # 'ssid': 'wifi', + # 'band': 'WIFI_BAND_2G', + # 'channel': 1, + # 'signal': -73, + # 'signal_bars': 1} + # ] + # } + print(await dpa.device.async_get_wifi_neighbor_access_points()) + + # Start WPS push button configuration. If WPS was started successfully, True is returned, otherwise False. + print("WPS started" if await dpa.device.async_start_wps() else "WPS start failed") + + + # Get PLC network overview with enriched information like firmware version, + # {'network': + # {'devices': + # [ + # {'product_name': 'devolo dLAN pro 1200+ WiFi ac', + # 'product_id': 'MT2730', + # 'friendly_version': '2.8.0.01', + # 'full_version': 'MAC-QCA7500-2.8.0.30-01-20190705-CS', + # 'user_device_name': '', + # 'mac_address': 'AABBCCDDEEFF', + # 'topology': 'LOCAL', + # 'technology': 'HPAV_PANTHER', + # 'bridged_devices': [], + # 'user_network_name': '', + # 'ipv4_address': '', + # 'attached_to_router': False}, + # {'product_name': 'devolo dLAN 1200+', + # 'product_id': 'MT2639', + # 'friendly_version': '2.8.0.01-1', + # 'full_version': 'MAC-QCA7500-2.8.0.30-01-20190705-CS', + # 'user_device_name': '', + # 'mac_address': 'AABBCCDDEEFF', + # 'topology': 'REMOTE', + # 'technology': 'HPAV_PANTHER', + # 'bridged_devices': [], + # 'attached_to_router': True, + # 'user_network_name': '', + # 'ipv4_address': ''} + # ], + # 'data_rates': + # [ + # {'mac_address_from': 'AABBCCDDEEFF', + # 'mac_address_to': 'AABBCCDDEEFF', + # 'tx_rate': 129.9375, + # 'rx_rate': 124.6875} + # ] + # } + # } + print(await dpa.plcnet.async_get_network_overview()) + + # Identify the device by making the PLC LED blink. This call returns directly with True, if identifing was started + # succcessfully, otherwise False. However, the LED stays blinking for two minutes. + print("success" if await dpa.plcnet.async_identify_device_start() else "failed") + + # Stop identify the device if you don't want to wait for the timeout. + print("success" if await dpa.plcnet.async_identify_device_start() else "failed") + + # Set the user device name. If the name was changed successfully, True is returned, otherwise False. + print("success" if await dpa.plcnet.async_set_user_device_name(name="New name") else "failed") + + +if __name__ == "__main__": + asyncio.run(run()) diff --git a/example_sync.py b/example_sync.py new file mode 100644 index 0000000..c391c1f --- /dev/null +++ b/example_sync.py @@ -0,0 +1,124 @@ +from devolo_plc_api.device import Device + + +# IP of the device to query +IP = "192.168.0.10" + +# Password, if the device has one. It is the same as the Web-UI has. It no password is set, you can remove the password +# parameter or set it to None. +PASSWORD = "super_secret" + + +def run(): + with Device(ip=IP, password=PASSWORD) as dpa: + + # Get LED settings of the device. The state might be LED_ON or LED_OFF. + # {'state': 'LED_ON'} + print(dpa.device.get_led_setting()) + + # Set LED settings of the device. Set enable to True to them turn on, to False to turn them off. + # If the state was changed successfully, True is returned, otherwise False. + print("success" if dpa.device.set_led_setting(enable=True) else "failed") + + # Check for new firmware versions + # {'result': 'UPDATE_NOT_AVAILABLE', 'new_firmware_version': ''} + print(dpa.device.check_firmware_available()) + + # Start firmware update, if new version is available. Important: The response does not tell you anything about the + # success of the update itself. + print("update started" if dpa.device.start_firmware_update() else "no update available") + + # Get details of wifi stations connected to the device: MAC address, access point type (main or guest), wifi band and + # connection rates. + # {'connected_stations': + # [ + # {'mac_address': 'AA:BB:CC:DD:EE:FF', + # 'vap_type': 'WIFI_VAP_MAIN_AP', + # 'band': 'WIFI_BAND_5G', + # 'rx_rate': 87800, + # 'tx_rate': 87800} + # ] + # } + print(dpa.device.get_wifi_connected_station()) + + # Get details about wifi guest access: SSID, Wifi key, state (enabled/disabled) and if time limited, the remaining + # duration. + # {'ssid': 'devolo-guest-930', 'key': 'HMANPGBA', 'enabled': False, 'remaining_duration': 0} + print(dpa.device.get_wifi_guest_access()) + + # Enable or disable the wifi guest access. Set enable to True to it turn on, to False to turn it off. Chaning SSID, + # wifi key or duration is currently not supported. If the state was changed successfully, True is returned, otherwise + # False. + print("success" if dpa.device.set_wifi_guest_access(enable=False) else "failed") + + # Get details about other access points in your neighborhood: MAC address, SSID, wifi band, used channel, signal + # strength in DB and a value from 1 to 5, if you would want to map the signal strenght to a signal bars. + # {'neighbor_aps': + # [ + # {'mac_address': 'AA:BB:CC:DD:EE:FF', + # 'ssid': 'wifi', + # 'band': 'WIFI_BAND_2G', + # 'channel': 1, + # 'signal': -73, + # 'signal_bars': 1} + # ] + # } + print(dpa.device.get_wifi_neighbor_access_points()) + + # Start WPS push button configuration. If WPS was started successfully, True is returned, otherwise False. + print("WPS started" if dpa.device.start_wps() else "WPS start failed") + + + # Get PLC network overview with enriched information like firmware version, + # {'network': + # {'devices': + # [ + # {'product_name': 'devolo dLAN pro 1200+ WiFi ac', + # 'product_id': 'MT2730', + # 'friendly_version': '2.8.0.01', + # 'full_version': 'MAC-QCA7500-2.8.0.30-01-20190705-CS', + # 'user_device_name': '', + # 'mac_address': 'AABBCCDDEEFF', + # 'topology': 'LOCAL', + # 'technology': 'HPAV_PANTHER', + # 'bridged_devices': [], + # 'user_network_name': '', + # 'ipv4_address': '', + # 'attached_to_router': False}, + # {'product_name': 'devolo dLAN 1200+', + # 'product_id': 'MT2639', + # 'friendly_version': '2.8.0.01-1', + # 'full_version': 'MAC-QCA7500-2.8.0.30-01-20190705-CS', + # 'user_device_name': '', + # 'mac_address': 'AABBCCDDEEFF', + # 'topology': 'REMOTE', + # 'technology': 'HPAV_PANTHER', + # 'bridged_devices': [], + # 'attached_to_router': True, + # 'user_network_name': '', + # 'ipv4_address': ''} + # ], + # 'data_rates': + # [ + # {'mac_address_from': 'AABBCCDDEEFF', + # 'mac_address_to': 'AABBCCDDEEFF', + # 'tx_rate': 129.9375, + # 'rx_rate': 124.6875} + # ] + # } + # } + print(dpa.plcnet.get_network_overview()) + + # Identify the device by making the PLC LED blink. This call returns directly with True, if identifing was started + # succcessfully, otherwise False. However, the LED stays blinking for two minutes. + print("success" if dpa.plcnet.identify_device_start() else "failed") + + # Stop identify the device if you don't want to wait for the timeout. + print("success" if dpa.plcnet.identify_device_start() else "failed") + + # Set the user device name. If the name was changed successfully, True is returned, otherwise False. + print("success" if dpa.plcnet.set_user_device_name(name="New name") else "failed") + + +if __name__ == "__main__": + run() diff --git a/tests/fixtures/protobuf.py b/tests/fixtures/protobuf.py index 6f17036..82fa026 100644 --- a/tests/fixtures/protobuf.py +++ b/tests/fixtures/protobuf.py @@ -38,8 +38,3 @@ def mock_wrong_password(mocker): mocker.patch("httpx.Client.get", Client.wrong_password) mocker.patch("httpx.AsyncClient.post", AsyncClient.wrong_password) mocker.patch("httpx.Client.post", Client.wrong_password) - - -@pytest.fixture() -def mock_message_to_dict(mocker): - mocker.patch("google.protobuf.json_format.MessageToDict", return_value=None) diff --git a/tests/test_data.json b/tests/test_data.json index 1a4672e..e845622 100644 --- a/tests/test_data.json +++ b/tests/test_data.json @@ -7,6 +7,7 @@ "MT": "2730", "Product": "dLAN pro 1200+ WiFi ac", "Path": "1234567890abcdef", + "Port": 80, "Version": "v0", "Features": "wifi1" }, @@ -14,6 +15,7 @@ "Path": "1234567890abcdef", "PlcMacAddress": "AABBCCDDEEFF", "PlcTechnology": "hpav", + "Port": 80, "Version": "v0" } }, diff --git a/tests/test_device.py b/tests/test_device.py index 83cf270..1e234ab 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -8,6 +8,7 @@ from devolo_plc_api.device_api.deviceapi import DeviceApi from devolo_plc_api.plcnet_api.plcnetapi import PlcNetApi +from devolo_plc_api.exceptions.device import DeviceNotFound class TestDevice: @@ -37,6 +38,14 @@ async def test__get_device_info(self, mock_device): assert mock_device.product == device_info['Product'] assert type(mock_device.device) == DeviceApi + @pytest.mark.asyncio + @pytest.mark.usefixtures("mock_device_api") + async def test__get_device_info_timeout(self, mock_device): + with patch("devolo_plc_api.device.Device._get_zeroconf_info", new=CoroutineMock()), \ + patch("asyncio.wait_for", new=CoroutineMock(side_effect=asyncio.TimeoutError())), \ + pytest.raises(DeviceNotFound): + await mock_device._get_device_info() + @pytest.mark.asyncio @pytest.mark.usefixtures("mock_plcnet_api") async def test__get_plcnet_info(self, mock_device): @@ -48,6 +57,14 @@ async def test__get_plcnet_info(self, mock_device): assert mock_device.technology == device_info['PlcTechnology'] assert type(mock_device.plcnet) == PlcNetApi + @pytest.mark.asyncio + @pytest.mark.usefixtures("mock_device_api") + async def test__get_plcnet_info_timeout(self, mock_device): + with patch("devolo_plc_api.device.Device._get_zeroconf_info", new=CoroutineMock()), \ + patch("asyncio.wait_for", new=CoroutineMock(side_effect=asyncio.TimeoutError())), \ + pytest.raises(DeviceNotFound): + await mock_device._get_plcnet_info() + @pytest.mark.asyncio @pytest.mark.usefixtures("mock_service_browser") async def test__get_zeroconf_info(self, mocker, mock_device): diff --git a/tests/test_deviceapi.py b/tests/test_deviceapi.py index be4dade..17b510a 100644 --- a/tests/test_deviceapi.py +++ b/tests/test_deviceapi.py @@ -3,19 +3,38 @@ import pytest from asynctest import CoroutineMock from google.protobuf.json_format import MessageToDict -from httpx import AsyncClient, Response +from httpx import AsyncClient, Client, Response from devolo_plc_api.device_api.deviceapi import DeviceApi -from devolo_plc_api.device_api.devolo_idl_proto_deviceapi_ledsettings_pb2 import LedSettingsGet +from devolo_plc_api.device_api.devolo_idl_proto_deviceapi_ledsettings_pb2 import LedSettingsGet, LedSettingsSetResponse +from devolo_plc_api.device_api.devolo_idl_proto_deviceapi_wifinetwork_pb2 import ( + WifiConnectedStationsGet, WifiGuestAccessGet, WifiGuestAccessSetResponse, WifiNeighborAPsGet, WifiRepeatedAPsGet, + WifiWpsPbcStart) +from devolo_plc_api.exceptions.feature import FeatureNotSupported +from devolo_plc_api.device_api.devolo_idl_proto_deviceapi_updatefirmware_pb2 import UpdateFirmwareCheck, UpdateFirmwareStart class TestDeviceApi: + def test_unsupported_feature(self, request): + with pytest.raises(FeatureNotSupported): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "[]", + "password") + device_api.get_led_setting() + @pytest.mark.asyncio async def test_async_get_led_setting(self, request): + led_setting_get = LedSettingsGet() + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ - patch("httpx.Response.aread", new=CoroutineMock(return_value=b"")): + patch("httpx.Response.aread", new=CoroutineMock(return_value=led_setting_get.SerializeToString())): device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], AsyncClient(), request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], @@ -23,6 +42,333 @@ async def test_async_get_led_setting(self, request): "password") led_setting = await device_api.async_get_led_setting() - assert led_setting == MessageToDict(LedSettingsGet(), + assert led_setting == MessageToDict(led_setting_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + def test_get_led_setting(self, request): + led_setting_get = LedSettingsGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=led_setting_get.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "led", + "password") + led_setting = device_api.get_led_setting() + + assert led_setting == MessageToDict(led_setting_get, including_default_value_fields=True, preserving_proto_field_name=True) + + @pytest.mark.asyncio + async def test_async_set_led_setting(self, request): + led_setting_set = LedSettingsSetResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_post", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=led_setting_set.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "led", + "password") + + assert await device_api.async_set_led_setting(True) + + def test_set_led_setting(self, request): + led_setting_set = LedSettingsSetResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._post", return_value=Response), \ + patch("httpx.Response.read", return_value=led_setting_set.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "led", + "password") + + assert device_api.set_led_setting(True) + + @pytest.mark.asyncio + async def test_async_check_firmware_available(self, request): + firmware_available = UpdateFirmwareCheck() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=firmware_available.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "update", + "password") + firmware = await device_api.async_check_firmware_available() + + assert firmware == MessageToDict(firmware_available, + including_default_value_fields=True, + preserving_proto_field_name=True) + + def test_check_firmware_available(self, request): + firmware_available = UpdateFirmwareCheck() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=firmware_available.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "update", + "password") + firmware = device_api.check_firmware_available() + + assert firmware == MessageToDict(firmware_available, + including_default_value_fields=True, + preserving_proto_field_name=True) + + @pytest.mark.asyncio + async def test_async_start_firmware_update(self, request): + firmware_update = UpdateFirmwareStart() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=firmware_update.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "update", + "password") + + assert await device_api.async_start_firmware_update() + + def test_start_firmware_update(self, request): + firmware_update = UpdateFirmwareStart() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=firmware_update.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "update", + "password") + + assert device_api.start_firmware_update() + + @pytest.mark.asyncio + async def test_async_get_wifi_connected_station(self, request): + wifi_connected_stations_get = WifiConnectedStationsGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=wifi_connected_stations_get.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + connected_stations = await device_api.async_get_wifi_connected_station() + + assert connected_stations == MessageToDict(wifi_connected_stations_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + def test_get_wifi_connected_station(self, request): + wifi_connected_stations_get = WifiConnectedStationsGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=wifi_connected_stations_get.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + connected_stations = device_api.get_wifi_connected_station() + + assert connected_stations == MessageToDict(wifi_connected_stations_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + @pytest.mark.asyncio + async def test_async_get_wifi_guest_access(self, request): + wifi_guest_access_get = WifiGuestAccessGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=wifi_guest_access_get.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + wifi_guest_access = await device_api.async_get_wifi_guest_access() + + assert wifi_guest_access == MessageToDict(wifi_guest_access_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + def test_get_wifi_guest_access(self, request): + wifi_guest_access_get = WifiGuestAccessGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=wifi_guest_access_get.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + wifi_guest_access = device_api.get_wifi_guest_access() + + assert wifi_guest_access == MessageToDict(wifi_guest_access_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + @pytest.mark.asyncio + async def test_async_set_wifi_guest_access(self, request): + wifi_guest_access_set = WifiGuestAccessSetResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_post", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=wifi_guest_access_set.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + + assert await device_api.async_set_wifi_guest_access(True) + + def test_set_wifi_guest_access(self, request): + wifi_guest_access_set = WifiGuestAccessSetResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._post", return_value=Response), \ + patch("httpx.Response.read", return_value=wifi_guest_access_set.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + + assert device_api.set_wifi_guest_access(True) + + @pytest.mark.asyncio + async def test_async_get_wifi_neighbor_access_points(self, request): + wifi_neighbor_accesspoints_get = WifiNeighborAPsGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=wifi_neighbor_accesspoints_get.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + wifi_neighbor_access_points = await device_api.async_get_wifi_neighbor_access_points() + + assert wifi_neighbor_access_points == MessageToDict(wifi_neighbor_accesspoints_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + def test_get_wifi_neighbor_access_points(self, request): + wifi_neighbor_access_points_get = WifiNeighborAPsGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=wifi_neighbor_access_points_get.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + wifi_neighbor_access_points = device_api.get_wifi_neighbor_access_points() + + assert wifi_neighbor_access_points == MessageToDict(wifi_neighbor_access_points_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + @pytest.mark.asyncio + async def test_async_get_wifi_repeated_access_points(self, request): + wifi_repeated_accesspoints_get = WifiRepeatedAPsGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=wifi_repeated_accesspoints_get.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + wifi_repeated_access_points = await device_api.async_get_wifi_repeated_access_points() + + assert wifi_repeated_access_points == MessageToDict(wifi_repeated_accesspoints_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + def test_get_wifi_repeated_access_points(self, request): + wifi_repeated_access_points_get = WifiRepeatedAPsGet() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=wifi_repeated_access_points_get.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + wifi_repeated_access_points = device_api.get_wifi_repeated_access_points() + + assert wifi_repeated_access_points == MessageToDict(wifi_repeated_access_points_get, + including_default_value_fields=True, + preserving_proto_field_name=True) + + @pytest.mark.asyncio + async def test_async_start_wps(self, request): + wps = WifiWpsPbcStart() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=wps.SerializeToString())): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + + assert await device_api.async_start_wps() + + def test_start_wps(self, request): + wps = WifiWpsPbcStart() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=wps.SerializeToString()): + device_api = DeviceApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Version'], + "wifi1", + "password") + + assert device_api.start_wps() diff --git a/tests/test_plcnetapi.py b/tests/test_plcnetapi.py new file mode 100644 index 0000000..c3f28e0 --- /dev/null +++ b/tests/test_plcnetapi.py @@ -0,0 +1,136 @@ +from unittest.mock import patch + +import pytest +from asynctest import CoroutineMock +from google.protobuf.json_format import MessageToDict +from httpx import AsyncClient, Client, Response + +from devolo_plc_api.plcnet_api.devolo_idl_proto_plcnetapi_getnetworkoverview_pb2 import GetNetworkOverview +from devolo_plc_api.plcnet_api.devolo_idl_proto_plcnetapi_identifydevice_pb2 import IdentifyDeviceResponse +from devolo_plc_api.plcnet_api.devolo_idl_proto_plcnetapi_setuserdevicename_pb2 import SetUserDeviceNameResponse +from devolo_plc_api.plcnet_api.plcnetapi import PlcNetApi + + +class TestDeviceApi: + + @pytest.mark.asyncio + async def test_async_get_network_overview(self, request): + network_overview = GetNetworkOverview() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_get", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=network_overview.SerializeToString())): + plcnet_api = PlcNetApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + AsyncClient(), + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['PlcMacAddress']) + overview = await plcnet_api.async_get_network_overview() + + assert overview == MessageToDict(network_overview, + including_default_value_fields=True, + preserving_proto_field_name=True) + + def test_get_network_overview(self, request): + network_overview = GetNetworkOverview() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._get", return_value=Response), \ + patch("httpx.Response.read", return_value=network_overview.SerializeToString()): + plcnet_api = PlcNetApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['PlcMacAddress']) + overview = plcnet_api.get_network_overview() + + assert overview == MessageToDict(network_overview, + including_default_value_fields=True, + preserving_proto_field_name=True) + + @pytest.mark.asyncio + async def test_async_identify_device_start(self, request): + identify_device = IdentifyDeviceResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_post", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=identify_device.SerializeToString())): + plcnet_api = PlcNetApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['PlcMacAddress']) + + assert await plcnet_api.async_identify_device_start() + + def test_identify_device_start(self, request): + identify_device = IdentifyDeviceResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._post", return_value=Response), \ + patch("httpx.Response.read", return_value=identify_device.SerializeToString()): + plcnet_api = PlcNetApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['PlcMacAddress']) + + assert plcnet_api.identify_device_start() + + @pytest.mark.asyncio + async def test_async_identify_device_stop(self, request): + identify_device = IdentifyDeviceResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_post", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=identify_device.SerializeToString())): + plcnet_api = PlcNetApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['PlcMacAddress']) + + assert await plcnet_api.async_identify_device_stop() + + def test_identify_device_stop(self, request): + identify_device = IdentifyDeviceResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._post", return_value=Response), \ + patch("httpx.Response.read", return_value=identify_device.SerializeToString()): + plcnet_api = PlcNetApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['PlcMacAddress']) + + assert plcnet_api.identify_device_stop() + + @pytest.mark.asyncio + async def test_async_set_user_device_name(self, request): + user_device_name_set = SetUserDeviceNameResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._async_post", new=CoroutineMock(return_value=Response)), \ + patch("httpx.Response.aread", new=CoroutineMock(return_value=user_device_name_set.SerializeToString())): + plcnet_api = PlcNetApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['PlcMacAddress']) + + assert await plcnet_api.async_set_user_device_name("Test") + + def test_set_user_device_name(self, request): + user_device_name_set = SetUserDeviceNameResponse() + + with patch("devolo_plc_api.clients.protobuf.Protobuf._post", return_value=Response), \ + patch("httpx.Response.read", return_value=user_device_name_set.SerializeToString()): + plcnet_api = PlcNetApi(request.cls.ip, + request.cls.device_info['_dvl-deviceapi._tcp.local.']['Port'], + Client(), + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'], + request.cls.device_info['_dvl-plcnetapi._tcp.local.']['PlcMacAddress']) + + assert plcnet_api.set_user_device_name("Test")