diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..4eaf559 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,12 @@ +# To get started with Dependabot version updates, you'll need to specify which +# package ecosystems to update and where the package manifests are located. +# Please see the documentation for all configuration options: +# https://help.github.com/github/administering-a-repository/configuration-options-for-dependency-updates + +version: 2 +updates: + - package-ecosystem: "pip" + directory: "/" # Location of package manifests + schedule: + interval: "daily" + target-branch: "development" diff --git a/README.md b/README.md index 99c7472..2f1dfd4 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,27 @@ pytest ## Usage -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). +All features we currently support on device basis 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). + +If you don't know the IP addresses of your devices, you can discover them. You will get a dictionary with the device's serial number as key. The connections to the devices will be already established, but you will have to take about disconnecting. + +```python +from devolo_plc_api.network import async_discover_network + +devices = await async_discover_network() +# Do your magic +await asyncio.gather(*[device.async_disconnect() for device in devices.values()]) +``` + +Or in a synchronous setup: + +```python +from devolo_plc_api.network import discover_network + +devices = discover_network() +# Do your magic +[device.disconnect() for device in devices.values()] +``` ## Supported device diff --git a/devolo_plc_api/__init__.py b/devolo_plc_api/__init__.py index d3ec452..493f741 100644 --- a/devolo_plc_api/__init__.py +++ b/devolo_plc_api/__init__.py @@ -1 +1 @@ -__version__ = "0.2.0" +__version__ = "0.3.0" diff --git a/devolo_plc_api/clients/protobuf.py b/devolo_plc_api/clients/protobuf.py index 64c17e3..9decc99 100644 --- a/devolo_plc_api/clients/protobuf.py +++ b/devolo_plc_api/clients/protobuf.py @@ -21,8 +21,9 @@ def __init__(self): self._loop = asyncio.get_running_loop() self._logger = logging.getLogger(self.__class__.__name__) + self.password: str + self._ip: str - self._password: str self._path: str self._port: int self._session: AsyncClient @@ -50,7 +51,7 @@ async def _async_get(self, sub_url: str, timeout: float = TIMEOUT) -> Response: url = f"{self.url}{sub_url}" self._logger.debug("Getting from %s", url) try: - return await self._session.get(url, auth=DigestAuth(self._user, self._password), timeout=timeout) + return await self._session.get(url, auth=DigestAuth(self._user, self.password), timeout=timeout) except TypeError: raise DevicePasswordProtected("The used password is wrong.") from None @@ -59,7 +60,7 @@ async def _async_post(self, sub_url: str, content: bytes, timeout: float = TIMEO url = f"{self.url}{sub_url}" self._logger.debug("Posting to %s", url) try: - return await self._session.post(url, auth=DigestAuth(self._user, self._password), content=content, timeout=timeout) + return await self._session.post(url, auth=DigestAuth(self._user, self.password), content=content, timeout=timeout) except TypeError: raise DevicePasswordProtected("The used password is wrong.") from None diff --git a/devolo_plc_api/device.py b/devolo_plc_api/device.py index c429b8a..2915020 100644 --- a/devolo_plc_api/device.py +++ b/devolo_plc_api/device.py @@ -1,34 +1,42 @@ import asyncio +import ipaddress import logging -import socket import struct from datetime import date from typing import Dict, Optional import httpx -from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf +from zeroconf import ServiceBrowser, ServiceInfo, ServiceStateChange, Zeroconf from .device_api.deviceapi import DeviceApi from .exceptions.device import DeviceNotFound from .plcnet_api.plcnetapi import PlcNetApi +EMPTY_INFO: Dict = { + "properties": {} +} + class Device: """ Representing object for your devolo PLC device. It stores all properties and functionalities discovered during setup. :param ip: IP address of the device to communicate with. - :param password: Password of the Web-UI, if it is protected + :param plcnetapi: Reuse externally gathered data for the plcnet API + :param deviceapi: Reuse externally gathered data for the device API :param zeroconf_instance: Zeroconf instance to be potentially reused. """ - def __init__(self, ip: str, password: Optional[str] = None, zeroconf_instance: Optional[Zeroconf] = None): + def __init__(self, + ip: str, + plcnetapi: Optional[Dict] = None, + deviceapi: Optional[Dict] = None, + zeroconf_instance: Optional[Zeroconf] = None): self.firmware_date = date.fromtimestamp(0) self.firmware_version = "" self.ip = ip self.mac = "" self.mt_number = 0 - self.password = password self.product = "" self.technology = "" self.serial_number = 0 @@ -37,33 +45,71 @@ def __init__(self, ip: str, password: Optional[str] = None, zeroconf_instance: O self.plcnet = None self._info: Dict = { - "_dvl-plcnetapi._tcp.local.": {}, - "_dvl-deviceapi._tcp.local.": {}, + "_dvl-plcnetapi._tcp.local.": plcnetapi or EMPTY_INFO, + "_dvl-deviceapi._tcp.local.": deviceapi or EMPTY_INFO, } self._logger = logging.getLogger(self.__class__.__name__) + self._password = "" self._zeroconf_instance = zeroconf_instance + logging.captureWarnings(True) self._loop: asyncio.AbstractEventLoop self._session: httpx.AsyncClient self._zeroconf: Zeroconf + def __del__(self): + if self._loop.is_running(): + self._logger.warning("Please disconnect properly from the device.") + async def __aenter__(self): - self._loop = asyncio.get_running_loop() - await self._setup_device() + await self.async_connect() return self async def __aexit__(self, exc_type, exc_val, exc_tb): - if not self._zeroconf_instance: - self._zeroconf.close() - await self._session.aclose() + await self.async_disconnect() def __enter__(self): - self._loop = asyncio.new_event_loop() - self._loop.run_until_complete(self._setup_device()) + self.connect() return self def __exit__(self, exc_type, exc_val, exc_tb): - self._loop.run_until_complete(self.__aexit__(exc_type, exc_val, exc_tb)) + self.disconnect() + + @property + def password(self) -> str: + """ The currently set device password. """ + return self._password + + @password.setter + def password(self, password: str): + """ Change the currently set device password. """ + self._password = password + if self.device: + self.device.password = password + + async def async_connect(self): + """ Connect to a device asynchronous. """ + self._loop = asyncio.get_running_loop() + self._session = httpx.AsyncClient() + self._zeroconf = self._zeroconf_instance or Zeroconf() + await asyncio.gather(self._get_device_info(), self._get_plcnet_info()) + if not self.device and not self.plcnet: + raise DeviceNotFound(f"The device {self.ip} did not answer.") + + def connect(self): + """ Connect to a device synchronous. """ + self._loop = asyncio.new_event_loop() + self._loop.run_until_complete(self.async_connect()) + + async def async_disconnect(self): + """ Disconnect from a device asynchronous. """ + if not self._zeroconf_instance: + self._zeroconf.close() + await self._session.aclose() + + def disconnect(self): + """ Disconnect from a device asynchronous. """ + self._loop.run_until_complete(self.async_disconnect()) self._loop.close() async def _get_device_info(self): @@ -74,13 +120,13 @@ async def _get_device_info(self): except asyncio.TimeoutError: return - self.firmware_date = date.fromisoformat(self._info[service_type].get("FirmwareDate", "1970-01-01")) - self.firmware_version = self._info[service_type].get("FirmwareVersion", "") - self.serial_number = self._info[service_type].get("SN", 0) - self.mt_number = self._info[service_type].get("MT", 0) - self.product = self._info[service_type].get("Product", "") + self.firmware_date = date.fromisoformat(self._info[service_type]["properties"].get("FirmwareDate", "1970-01-01")) + self.firmware_version = self._info[service_type]["properties"].get("FirmwareVersion", "") + self.serial_number = self._info[service_type]["properties"].get("SN", 0) + self.mt_number = self._info[service_type]["properties"].get("MT", 0) + self.product = self._info[service_type]["properties"].get("Product", "") - self.device = DeviceApi(ip=self.ip, session=self._session, info=self._info[service_type], password=self.password) + self.device = DeviceApi(ip=self.ip, session=self._session, info=self._info[service_type]) async def _get_plcnet_info(self): """ Get information from the devolo PlcNet API. """ @@ -90,41 +136,52 @@ async def _get_plcnet_info(self): except asyncio.TimeoutError: return - self.mac = self._info[service_type]['PlcMacAddress'] - self.technology = self._info[service_type].get("PlcTechnology", "") + self.mac = self._info[service_type]["properties"]["PlcMacAddress"] + self.technology = self._info[service_type]["properties"].get("PlcTechnology", "") self.plcnet = PlcNetApi(ip=self.ip, session=self._session, info=self._info[service_type]) async def _get_zeroconf_info(self, service_type: str): """ Browse for the desired mDNS service types and query them. """ + if self._info[service_type]["properties"]: + return # No need to continue, if device info already exist + self._logger.debug("Browsing for %s", service_type) - browser = ServiceBrowser(self._zeroconf, service_type, [self._state_change]) - while not self._info[service_type]: + browser = ServiceBrowser(self._zeroconf, service_type, [self._state_change], addr=self.ip) + while not self._info[service_type]["properties"]: await asyncio.sleep(0.1) browser.cancel() - async def _setup_device(self): - """ Setup device. """ - self._session = httpx.AsyncClient() - self._zeroconf = self._zeroconf_instance or Zeroconf() - await asyncio.gather(self._get_device_info(), self._get_plcnet_info()) - if not self.device and not self.plcnet: - raise DeviceNotFound(f"The device {self.ip} did not answer.") - def _state_change(self, zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange): """ Evaluate the query result. """ service_info = zeroconf.get_service_info(service_type, name) - if service_info and state_change is ServiceStateChange.Added and \ - self.ip in [socket.inet_ntoa(address) for address in service_info.addresses]: - self._logger.debug("Adding service info of %s", service_type) - self._info[service_type]['Port'] = service_info.port + if service_info is None: + return # No need to continue, if there are no service information - # 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 - while offset < total_length: - parsed_length, = struct.unpack_from("!B", service_info.text, offset) - key_value = service_info.text[offset + 1:offset + 1 + parsed_length].decode("UTF-8").split("=") - self._info[service_type][key_value[0]] = key_value[1] - offset += parsed_length + 1 + if state_change is ServiceStateChange.Added: + self._logger.debug("Adding service info of %s", service_type) + self._info[service_type] = self.info_from_service(service_info) + + @staticmethod + def info_from_service(service_info: ServiceInfo) -> Optional[Dict]: + """ Return prepared info from mDNS entries. """ + properties = {} + if not service_info.addresses: + return None # No need to continue, if there is no IP address to contact the device + + total_length = len(service_info.text) + offset = 0 + while offset < total_length: + parsed_length, = struct.unpack_from("!B", service_info.text, offset) + key_value = service_info.text[offset + 1:offset + 1 + parsed_length].decode("UTF-8").split("=") + properties[key_value[0]] = key_value[1] + offset += parsed_length + 1 + + address = service_info.addresses[0] + + return { + "address": str(ipaddress.ip_address(address)), + "port": service_info.port, + "properties": properties, + } diff --git a/devolo_plc_api/device_api/deviceapi.py b/devolo_plc_api/device_api/deviceapi.py index da3d930..9c834e5 100644 --- a/devolo_plc_api/device_api/deviceapi.py +++ b/devolo_plc_api/device_api/deviceapi.py @@ -1,4 +1,4 @@ -from typing import Callable, Dict, Optional +from typing import Callable, Dict from httpx import AsyncClient @@ -16,21 +16,21 @@ class DeviceApi(Protobuf): :param ip: IP address of the device to communicate with :param session: HTTP client session :param info: Information collected from the mDNS query - :param password: Password of the device """ - def __init__(self, ip: str, session: AsyncClient, info: Dict, password: Optional[str]): + def __init__(self, ip: str, session: AsyncClient, info: Dict): super().__init__() + self._ip = ip - self._port = info['Port'] + self._path = info["properties"]["Path"] + self._port = info["port"] self._session = session - self._path = info['Path'] - self._version = info['Version'] self._user = "devolo" - self._password = password or "" + self._version = info["properties"]["Version"] - features = info.get("Features", "") - self.features = features.split(",") if features else ['reset', 'update', 'led', 'intmtg'] + features = info["properties"].get("Features", "") + self.features = features.split(",") if features else ["reset", "update", "led", "intmtg"] + self.password = "" def _feature(feature: str): # type: ignore # pylint: disable=no-self-argument """ Decorator to filter unsupported features before querying the device. """ diff --git a/devolo_plc_api/network/__init__.py b/devolo_plc_api/network/__init__.py new file mode 100644 index 0000000..c58f3c2 --- /dev/null +++ b/devolo_plc_api/network/__init__.py @@ -0,0 +1,51 @@ +import asyncio +import time +from typing import Dict + +from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf + +from ..device import Device + +_devices: Dict[str, + Device] = {} + + +async def async_discover_network() -> Dict[str, Device]: + """ + Discover all devices that expose the devolo device API via mDNS asynchronous. + + :return: Devices accessible via serial number. + """ + browser = ServiceBrowser(Zeroconf(), "_dvl-deviceapi._tcp.local.", [_add]) + await asyncio.sleep(3) + browser.cancel() + await asyncio.gather(*[device.async_connect() for device in _devices.values()]) + return _devices + + +def discover_network() -> Dict[str, Device]: + """ + Discover devices that expose the devolo device API via mDNS synchronous. + + :return: Devices accessible via serial number. + """ + browser = ServiceBrowser(Zeroconf(), "_dvl-deviceapi._tcp.local.", [_add]) + time.sleep(3) + browser.cancel() + for device in _devices.values(): + device.connect() + return _devices + + +def _add(zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange): + """" Create a device object to each matching device. """ + if state_change is ServiceStateChange.Added: + service_info = zeroconf.get_service_info(service_type, name) + if service_info is None: + return + + info = Device.info_from_service(service_info) + if info is None: + return + + _devices[info["properties"]["SN"]] = Device(ip=info["address"], deviceapi=info, zeroconf_instance=zeroconf) diff --git a/devolo_plc_api/plcnet_api/plcnetapi.py b/devolo_plc_api/plcnet_api/plcnetapi.py index c6c07af..1633638 100644 --- a/devolo_plc_api/plcnet_api/plcnetapi.py +++ b/devolo_plc_api/plcnet_api/plcnetapi.py @@ -19,14 +19,16 @@ class PlcNetApi(Protobuf): def __init__(self, ip: str, session: AsyncClient, info: Dict): super().__init__() + self._ip = ip - self._mac = info['PlcMacAddress'] - self._path = info['Path'] - self._port = info['Port'] + self._mac = info["properties"]["PlcMacAddress"] + self._path = info["properties"]["Path"] + self._port = info["port"] self._session = session - self._version = info['Version'] self._user = "" # PLC API is not password protected. - self._password = "" # PLC API is not password protected. + self._version = info["properties"]["Version"] + + self.password = "" # PLC API is not password protected. async def async_get_network_overview(self) -> dict: """ diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index e1ee690..d094d6b 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,6 +4,18 @@ 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.3.0] - 02.12.2020 + +### Added + +- If API data is discovered externally, it can be reused +- The devices can be accessed without context manager +- If the network topology is unknown, it can be discovered now + +### Changed + +- **BREAKING**: The device password must be specified by setting an attribute now + ## [v0.2.0] - 14.09.2020 ### Added diff --git a/example_async.py b/example_async.py index e2ad754..8097dc6 100644 --- a/example_async.py +++ b/example_async.py @@ -11,7 +11,9 @@ async def run(): - async with Device(ip=IP, password=PASSWORD) as dpa: + async with Device(ip=IP) as dpa: + # Set the password + dpa.password = PASSWORD # Get LED settings of the device. The state might be LED_ON or LED_OFF. # {'state': 'LED_ON'} diff --git a/example_sync.py b/example_sync.py index c21139f..35035ea 100644 --- a/example_sync.py +++ b/example_sync.py @@ -9,7 +9,9 @@ def run(): - with Device(ip=IP, password=PASSWORD) as dpa: + with Device(ip=IP) as dpa: + # Set the password + dpa.password = PASSWORD # Get LED settings of the device. The state might be LED_ON or LED_OFF. # {'state': 'LED_ON'} diff --git a/tests/fixtures/device.py b/tests/fixtures/device.py index c7f9ab8..6842c08 100644 --- a/tests/fixtures/device.py +++ b/tests/fixtures/device.py @@ -1,25 +1,35 @@ from copy import deepcopy +from unittest.mock import Mock import pytest from zeroconf import Zeroconf +try: + from unittest.mock import AsyncMock +except ImportError: + from asynctest import CoroutineMock as AsyncMock + from devolo_plc_api.device import Device -from ..mocks.mock_zeroconf import MockZeroconf +from ..mocks.mock_zeroconf import MockServiceBrowser, MockZeroconf @pytest.fixture() -def mock_device(request): +def mock_device(mocker, request): device = Device(ip=request.cls.ip) device._info = deepcopy(request.cls.device_info) - device._session = None - device._zeroconf = None + device._loop = Mock() + device._loop.is_running = lambda: False + device._session = Mock() + device._session.aclose = AsyncMock() + device._zeroconf = Mock() + device._zeroconf.close = lambda: None return device @pytest.fixture() def mock_service_browser(mocker): - mocker.patch("zeroconf.ServiceBrowser.__init__", return_value=None) + mocker.patch("zeroconf.ServiceBrowser.__init__", MockServiceBrowser.__init__) mocker.patch("zeroconf.ServiceBrowser.cancel", return_value=None) diff --git a/tests/fixtures/device_api.py b/tests/fixtures/device_api.py index 5cc1ecd..69cb0f0 100644 --- a/tests/fixtures/device_api.py +++ b/tests/fixtures/device_api.py @@ -18,8 +18,8 @@ def device_api(request, feature): patch("devolo_plc_api.clients.protobuf.Protobuf._async_post", new=AsyncMock(return_value=Response)), \ patch("asyncio.get_running_loop", asyncio.new_event_loop): asyncio.new_event_loop() - request.cls.device_info["_dvl-deviceapi._tcp.local."]["Features"] = feature - yield DeviceApi(request.cls.ip, AsyncClient(), request.cls.device_info["_dvl-deviceapi._tcp.local."], "password") + request.cls.device_info["_dvl-deviceapi._tcp.local."]["properties"]["Features"] = feature + yield DeviceApi(request.cls.ip, AsyncClient(), request.cls.device_info["_dvl-deviceapi._tcp.local."]) @pytest.fixture() diff --git a/tests/mocks/mock_devices.py b/tests/mocks/mock_devices.py new file mode 100644 index 0000000..8aa26b0 --- /dev/null +++ b/tests/mocks/mock_devices.py @@ -0,0 +1,10 @@ +import json +import pathlib + +file = pathlib.Path(__file__).parent / ".." / "test_data.json" +with file.open("r") as fh: + test_data = json.load(fh) + + +def state_change(self): + self._info = test_data["device_info"] diff --git a/tests/mocks/mock_zeroconf.py b/tests/mocks/mock_zeroconf.py index 34857eb..aa060fa 100644 --- a/tests/mocks/mock_zeroconf.py +++ b/tests/mocks/mock_zeroconf.py @@ -16,3 +16,9 @@ def get_service_info(self, service_type, name): service_info.addresses = [socket.inet_aton(test_data['ip'])] service_info.text = b"\x09new=value" return service_info + + +class MockServiceBrowser: + + def __init__(self, zc, st, sc, addr=None): + sc[0]() diff --git a/tests/stubs/protobuf.py b/tests/stubs/protobuf.py index d96b1d6..8e203ba 100644 --- a/tests/stubs/protobuf.py +++ b/tests/stubs/protobuf.py @@ -17,7 +17,8 @@ def __init__(self): self._loop = asyncio.new_event_loop() self._ip = test_data["ip"] self._port = 14791 - self._path = test_data["device_info"]["_dvl-plcnetapi._tcp.local."]["Path"] - self._version = test_data["device_info"]["_dvl-plcnetapi._tcp.local."]["Version"] + self._path = test_data["device_info"]["_dvl-plcnetapi._tcp.local."]["properties"]["Path"] + self._version = test_data["device_info"]["_dvl-plcnetapi._tcp.local."]["properties"]["Version"] self._user = "user" - self._password = "password" + + self.password = "password" diff --git a/tests/test_data.json b/tests/test_data.json index 796683d..9b29edc 100644 --- a/tests/test_data.json +++ b/tests/test_data.json @@ -1,22 +1,26 @@ { "device_info": { "_dvl-deviceapi._tcp.local.": { - "FirmwareDate": "2020-06-29", - "FirmwareVersion": "5.5.1", - "SN": "1234567890123456", - "MT": "2730", - "Product": "dLAN pro 1200+ WiFi ac", - "Path": "1234567890abcdef", - "Port": 80, - "Version": "v0", - "Features": "wifi1" + "port": 80, + "properties": { + "FirmwareDate": "2020-06-29", + "FirmwareVersion": "5.5.1", + "SN": "1234567890123456", + "MT": "2730", + "Product": "dLAN pro 1200+ WiFi ac", + "Path": "1234567890abcdef", + "Version": "v0", + "Features": "wifi1" + } }, "_dvl-plcnetapi._tcp.local.": { - "Path": "1234567890abcdef", - "PlcMacAddress": "AABBCCDDEEFF", - "PlcTechnology": "hpav", - "Port": 80, - "Version": "v0" + "port": 80, + "properties": { + "Path": "1234567890abcdef", + "PlcMacAddress": "AABBCCDDEEFF", + "PlcTechnology": "hpav", + "Version": "v0" + } } }, "ip": "192.168.0.10" diff --git a/tests/test_device.py b/tests/test_device.py index 1ea5b0d..de86ada 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -1,33 +1,81 @@ import asyncio from datetime import date -from unittest.mock import patch +from unittest.mock import Mock, patch import pytest -from zeroconf import ServiceBrowser, ServiceStateChange +from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf try: from unittest.mock import AsyncMock except ImportError: from asynctest import CoroutineMock as AsyncMock +from devolo_plc_api.device import EMPTY_INFO from devolo_plc_api.device_api.deviceapi import DeviceApi from devolo_plc_api.exceptions.device import DeviceNotFound from devolo_plc_api.plcnet_api.plcnetapi import PlcNetApi +from .mocks.mock_devices import state_change + class TestDevice: + @pytest.mark.parametrize("feature", [""]) + def test_set_password(self, mock_device, device_api): + mock_device.device = device_api + mock_device.password = "super_secret" + assert mock_device.device.password == "super_secret" + + @pytest.mark.asyncio + async def test_async_connect(self, mocker, mock_device): + with patch("devolo_plc_api.device.Device._get_device_info", new=AsyncMock()), \ + patch("devolo_plc_api.device.Device._get_plcnet_info", new=AsyncMock()): + spy_device_info = mocker.spy(mock_device, "_get_device_info") + spy_plcnet_info = mocker.spy(mock_device, "_get_plcnet_info") + mock_device.device = object + mock_device.plcnet = object + await mock_device.async_connect() + assert spy_device_info.call_count == 1 + assert spy_plcnet_info.call_count == 1 + + @pytest.mark.asyncio + async def test_async_connect_not_found(self, mock_device): + with patch("devolo_plc_api.device.Device._get_device_info", new=AsyncMock()), \ + patch("devolo_plc_api.device.Device._get_plcnet_info", new=AsyncMock()), \ + pytest.raises(DeviceNotFound): + await mock_device.async_connect() + + def test_connect(self, mocker, mock_device): + with patch("devolo_plc_api.device.Device.async_connect", new=AsyncMock()): + spy_connect = mocker.spy(mock_device, "async_connect") + mock_device.connect() + assert spy_connect.call_count == 1 + + @pytest.mark.asyncio + async def test_async_disconnect(self, mocker, mock_device): + spy_zeroconf = mocker.spy(mock_device._zeroconf, "close") + spy_session = mocker.spy(mock_device._session, "aclose") + await mock_device.async_disconnect() + assert spy_zeroconf.call_count == 1 + assert spy_session.call_count == 1 + + def test_disconnect(self, mocker, mock_device): + with patch("devolo_plc_api.device.Device.async_disconnect", new=AsyncMock()): + spy_connect = mocker.spy(mock_device, "async_disconnect") + mock_device.disconnect() + assert spy_connect.call_count == 1 + @pytest.mark.asyncio @pytest.mark.usefixtures("mock_device_api") async def test__get_device_info(self, mock_device): with patch("devolo_plc_api.device.Device._get_zeroconf_info", new=AsyncMock()): device_info = self.device_info['_dvl-deviceapi._tcp.local.'] await mock_device._get_device_info() - assert mock_device.firmware_date == date.fromisoformat(device_info['FirmwareDate']) - assert mock_device.firmware_version == device_info['FirmwareVersion'] - assert mock_device.serial_number == device_info['SN'] - assert mock_device.mt_number == device_info['MT'] - assert mock_device.product == device_info['Product'] + assert mock_device.firmware_date == date.fromisoformat(device_info["properties"]["FirmwareDate"]) + assert mock_device.firmware_version == device_info["properties"]["FirmwareVersion"] + assert mock_device.serial_number == device_info["properties"]["SN"] + assert mock_device.mt_number == device_info["properties"]["MT"] + assert mock_device.product == device_info["properties"]["Product"] assert type(mock_device.device) == DeviceApi @pytest.mark.asyncio @@ -42,10 +90,10 @@ async def test__get_device_info_timeout(self, mock_device): @pytest.mark.usefixtures("mock_plcnet_api") async def test__get_plcnet_info(self, mock_device): with patch("devolo_plc_api.device.Device._get_zeroconf_info", new=AsyncMock()): - device_info = self.device_info['_dvl-plcnetapi._tcp.local.'] + device_info = self.device_info["_dvl-plcnetapi._tcp.local."] await mock_device._get_plcnet_info() - assert mock_device.mac == device_info['PlcMacAddress'] - assert mock_device.technology == device_info['PlcTechnology'] + assert mock_device.mac == device_info["properties"]["PlcMacAddress"] + assert mock_device.technology == device_info["properties"]["PlcTechnology"] assert type(mock_device.plcnet) == PlcNetApi @pytest.mark.asyncio @@ -59,39 +107,37 @@ async def test__get_plcnet_info_timeout(self, mock_device): @pytest.mark.asyncio @pytest.mark.usefixtures("mock_service_browser") async def test__get_zeroconf_info(self, mocker, mock_device): - spy_cancel = mocker.spy(ServiceBrowser, "cancel") - spy_sleep = mocker.spy(asyncio, "sleep") - await mock_device._get_zeroconf_info("_dvl-plcnetapi._tcp.local.") - assert spy_cancel.call_count == 1 - assert spy_sleep.call_count == 0 + with patch("devolo_plc_api.device.Device._state_change", state_change): + mock_device._info["_dvl-plcnetapi._tcp.local."] = EMPTY_INFO + spy_cancel = mocker.spy(ServiceBrowser, "cancel") + await mock_device._get_zeroconf_info("_dvl-plcnetapi._tcp.local.") + assert spy_cancel.call_count == 1 @pytest.mark.asyncio - async def test__setup_device(self, mocker, mock_device): - with patch("devolo_plc_api.device.Device._get_device_info", new=AsyncMock()), \ - patch("devolo_plc_api.device.Device._get_plcnet_info", new=AsyncMock()): - spy_device_info = mocker.spy(mock_device, "_get_device_info") - spy_plcnet_info = mocker.spy(mock_device, "_get_plcnet_info") - mock_device.device = object - mock_device.plcnet = object - await mock_device._setup_device() - assert spy_device_info.call_count == 1 - assert spy_plcnet_info.call_count == 1 + @pytest.mark.usefixtures("mock_service_browser") + async def test__get_zeroconf_info_device_info_exists(self, mocker, mock_device): + spy_cancel = mocker.spy(ServiceBrowser, "cancel") + await mock_device._get_zeroconf_info("_dvl-plcnetapi._tcp.local.") + assert spy_cancel.call_count == 0 - @pytest.mark.asyncio - async def test__setup_device_not_found(self, mock_device): - with patch("devolo_plc_api.device.Device._get_device_info", new=AsyncMock()), \ - patch("devolo_plc_api.device.Device._get_plcnet_info", new=AsyncMock()), \ - pytest.raises(DeviceNotFound): - await mock_device._setup_device() + def test__state_change_no_service_info(self, mocker, mock_device): + with patch("zeroconf.Zeroconf.get_service_info", return_value=None): + service_type = "_dvl-plcnetapi._tcp.local." + spy_service_info = mocker.spy(mock_device, "info_from_service") + mock_device._state_change(Zeroconf(), service_type, service_type, ServiceStateChange.Added) + assert spy_service_info.call_count == 0 - @pytest.mark.asyncio def test__state_change_added(self, mock_device, mock_zeroconf): service_type = "_dvl-plcnetapi._tcp.local." mock_device._state_change(mock_zeroconf, service_type, service_type, ServiceStateChange.Added) - assert mock_device._info[service_type]['new'] == "value" + assert mock_device._info[service_type]["properties"]["new"] == "value" - @pytest.mark.asyncio def test__state_change_removed(self, mock_device, mock_zeroconf): service_type = "_dvl-plcnetapi._tcp.local." mock_device._state_change(mock_zeroconf, service_type, service_type, ServiceStateChange.Removed) assert "new" not in mock_device._info[service_type] + + def test_info_from_service_no_address(self, mock_device): + service_info = Mock() + service_info.addresses = None + assert mock_device.info_from_service(service_info) is None diff --git a/tests/test_network.py b/tests/test_network.py new file mode 100644 index 0000000..e6285b4 --- /dev/null +++ b/tests/test_network.py @@ -0,0 +1,67 @@ +from unittest.mock import Mock, patch + +import pytest +from zeroconf import ServiceStateChange, Zeroconf + +try: + from unittest.mock import AsyncMock +except ImportError: + from asynctest import CoroutineMock as AsyncMock + +import devolo_plc_api.network as network +from devolo_plc_api.device import Device + + +class TestNetwork: + + @pytest.mark.asyncio + async def test_async_discover_network(self, mocker): + device = { + "1234567890123456": Device(ip="123.123.123.123") + } + with patch("zeroconf.Zeroconf", new=Mock()), \ + patch("asyncio.sleep", new=AsyncMock()), \ + patch("devolo_plc_api.device.Device.async_connect", new=AsyncMock()): + network._devices = device + spy_connect = mocker.spy(Device, "async_connect") + discovered = await network.async_discover_network() + assert spy_connect.call_count == 1 + assert discovered == device + + def test_discover_network(self, mocker): + device = { + "1234567890123456": Device(ip="123.123.123.123") + } + with patch("zeroconf.ServiceBrowser", new=Mock()), \ + patch("time.sleep", new=Mock()), \ + patch("devolo_plc_api.device.Device.connect", new=Mock()): + network._devices = device + spy_connect = mocker.spy(Device, "connect") + discovered = network.discover_network() + assert spy_connect.call_count == 1 + assert discovered == device + + def test__add(self, mocker): + service_info = { + "properties": { + "SN": "1234567890123456" + }, + "address": "123.123.123.123", + } + with patch("zeroconf.Zeroconf.get_service_info", return_value="service_info"), \ + patch("devolo_plc_api.device.Device.info_from_service", return_value=service_info): + network._add(Zeroconf(), "_dvl-deviceapi._tcp.local.", "_dvl-deviceapi._tcp.local.", ServiceStateChange.Added) + assert "1234567890123456" in network._devices + + def test__add_no_device(self, mocker): + with patch("zeroconf.Zeroconf.get_service_info", return_value=None): + spy_info = mocker.spy(Device, "info_from_service") + network._add(Zeroconf(), "_dvl-deviceapi._tcp.local.", "_dvl-deviceapi._tcp.local.", ServiceStateChange.Added) + assert spy_info.call_count == 0 + + def test__add_no_info(self, mocker): + with patch("zeroconf.Zeroconf.get_service_info", return_value="service_info"), \ + patch("devolo_plc_api.device.Device.info_from_service", return_value=None): + spy_device = mocker.spy(Device, "__init__") + network._add(Zeroconf(), "_dvl-deviceapi._tcp.local.", "_dvl-deviceapi._tcp.local.", ServiceStateChange.Added) + assert spy_device.call_count == 0 diff --git a/tests/test_profobuf.py b/tests/test_profobuf.py index c509285..8869101 100644 --- a/tests/test_profobuf.py +++ b/tests/test_profobuf.py @@ -14,8 +14,8 @@ def test_attribute_error(self, mock_protobuf): def test_url(self, request, mock_protobuf): ip = request.cls.ip - path = request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Path'] - version = request.cls.device_info['_dvl-plcnetapi._tcp.local.']['Version'] + path = request.cls.device_info["_dvl-plcnetapi._tcp.local."]["properties"]["Path"] + version = request.cls.device_info["_dvl-plcnetapi._tcp.local."]["properties"]["Version"] assert mock_protobuf.url == f"http://{ip}:14791/{path}/{version}/" @pytest.mark.asyncio