6080961227

Committed 05 Sep 2023 06:23AM UTC coverage: 97.468% (-1.1%) from 98.543%

Build # 6080961227

Build Type

Pull #142

github-actions

Committed by

Shutgun

Commit Message

Fix TCH in tests

Pull Request Pull Request #142: Fix new ruff findings

Run Details

14 of 14 new or added lines in 3 files covered. (100.0%)

539 of 553 relevant lines covered (97.47%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.08

/devolo_plc_api/device.py

"""Representation of your devolo device."""
from __future__ import annotations

import asyncio
import logging
from contextlib import suppress
from datetime import date
from ipaddress import ip_address, ip_network
from struct import unpack_from
from typing import TYPE_CHECKING, cast

from httpx import AsyncClient
from ifaddr import get_adapters
from zeroconf import DNSQuestionType, ServiceInfo, ServiceStateChange, Zeroconf
from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf

from .device_api import SERVICE_TYPE as DEVICEAPI, DeviceApi
from .exceptions import DeviceNotFound
from .plcnet_api import DEVICES_WITHOUT_PLCNET, SERVICE_TYPE as PLCNETAPI, PlcNetApi
from .zeroconf import ZeroconfServiceInfo

if TYPE_CHECKING:
    from types import TracebackType

    from typing_extensions import Self


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 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.
    """

    MDNS_TIMEOUT = 300

    def __init__(
        self,
        ip: str,
        zeroconf_instance: AsyncZeroconf | Zeroconf | None = None,
    ) -> None:
        """Initialize the device."""
        self.ip = ip
        self.mac = ""
        self.mt_number = "0"
        self.product = ""
        self.technology = ""
        self.serial_number = "0"

        self.device: DeviceApi | None = None
        self.plcnet: PlcNetApi | None = None

        self._background_tasks: set[asyncio.Task] = set()
        self._browser: AsyncServiceBrowser | None = None
        self._connected = False
        self._info: dict[str, ZeroconfServiceInfo] = {PLCNETAPI: ZeroconfServiceInfo(), DEVICEAPI: ZeroconfServiceInfo()}
        self._logger = logging.getLogger(f"{self.__class__.__module__}.{self.__class__.__name__}")
        self._multicast = False
        self._password = ""
        self._session_instance: AsyncClient | None = None
        self._zeroconf_instance = zeroconf_instance
        logging.captureWarnings(capture=True)

        self._loop: asyncio.AbstractEventLoop
        self._session: AsyncClient
        self._zeroconf: AsyncZeroconf

    def __del__(self) -> None:
        """Warn user, if the connection was not properly closed."""
        if self._connected and self._session_instance is None:
            self._logger.warning("Please disconnect properly from the device.")

    async def __aenter__(self) -> Self:
        """Connect to a device asynchronously when entering a context manager."""
        await self.async_connect()
        return self

    async def __aexit__(
        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
    ) -> None:
        """Disconnect to a device asynchronously when entering a context manager."""
        await self.async_disconnect()

    def __enter__(self) -> Self:
        """Connect to a device synchronously when leaving a context manager."""
        self.connect()
        return self

    def __exit__(
        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
    ) -> None:
        """Disconnect to a device synchronously when leaving a context manager."""
        self.disconnect()

    @property
    def firmware_date(self) -> date:
        """Date the firmware was built."""
        return date.fromisoformat(self._info[DEVICEAPI].properties.get("FirmwareDate", "1970-01-01")[:10])

    @property
    def firmware_version(self) -> str:
        """Firmware version currently installed."""
        return self._info[DEVICEAPI].properties.get("FirmwareVersion", "")

    @property
    def hostname(self) -> str:
        """mDNS hostname of the device."""  # noqa: D403
        return self._info[DEVICEAPI].hostname

    @property
    def password(self) -> str:
        """The currently set device password."""
        return self._password

    @password.setter
    def password(self, password: str) -> None:
        """Change the currently set device password."""
        self._password = password
        if self.device:
            self.device.password = password
        if self.plcnet:
            self.plcnet.password = password

    async def async_connect(self, session_instance: AsyncClient | None = None) -> None:
        """
        Connect to a device asynchronous.

        :param: session_instance: Session client instance to be potentially reused.
        """
        self._session_instance = session_instance
        self._session = self._session_instance or AsyncClient()
        if not self._zeroconf_instance:
            self._zeroconf = AsyncZeroconf(interfaces=await self._get_relevant_interfaces())
        elif isinstance(self._zeroconf_instance, Zeroconf):
            self._zeroconf = AsyncZeroconf(zc=self._zeroconf_instance)
        else:
            self._zeroconf = self._zeroconf_instance
        await self._get_zeroconf_info()
        if not self._info[DEVICEAPI].properties and not self._info[PLCNETAPI].properties:
            await self._retry_zeroconf_info()
        if not self.device and not self.plcnet:
            raise DeviceNotFound(self.ip)
        self._connected = True

    def connect(self) -> None:
        """Connect to a device synchronous."""
        self._loop = asyncio.get_event_loop()
        self._loop.run_until_complete(self.async_connect())

    async def async_disconnect(self) -> None:
        """Disconnect from a device asynchronous."""
        if self._connected:
            if self._browser:
                await self._browser.async_cancel()
            if not self._zeroconf_instance:
                await self._zeroconf.async_close()
            if not self._session_instance:
                await self._session.aclose()
            self._connected = False

    def disconnect(self) -> None:
        """Disconnect from a device synchronous."""
        self._loop.run_until_complete(self.async_disconnect())

    async def _get_relevant_interfaces(self) -> list[str]:
        """Get the IP address of the relevant interface to reduce traffic."""
        interface: list[str] = []
        for adapter in get_adapters():
            interface.extend(
                cast(str, ip.ip)
                for ip in adapter.ips
                if ip.is_IPv4 and ip_address(self.ip) in ip_network(f"{ip.ip}/{ip.network_prefix}", strict=False)
            )
            interface.extend(
                cast(str, ip.ip[0])
                for ip in adapter.ips
                if ip.is_IPv6 and ip_address(self.ip) in ip_network(f"{ip.ip[0]}/{ip.network_prefix}", strict=False)
            )
        return interface

    async def _get_device_info(self) -> None:
        """Get information from the devolo Device API."""
        service_type = DEVICEAPI
        if self._info[service_type].properties:
            self.mt_number = self._info[service_type].properties.get("MT", "0")
            self.product = self._info[service_type].properties.get("Product", "")
            self.serial_number = self._info[service_type].properties["SN"]
            self.device = DeviceApi(
                ip=str(ip_address(self._info[service_type].address)),
                session=self._session,
                info=self._info[service_type],
            )
            self.device.password = self.password

    async def _get_plcnet_info(self) -> None:
        """Get information from the devolo PlcNet API."""
        service_type = PLCNETAPI
        if self._info[service_type].properties:
            self.mac = self._info[service_type].properties["PlcMacAddress"]
            self.technology = self._info[service_type].properties.get("PlcTechnology", "")
            self.plcnet = PlcNetApi(
                ip=str(ip_address(self._info[service_type].address)),
                session=self._session,
                info=self._info[service_type],
            )
            self.plcnet.password = self.password

    async def _get_zeroconf_info(self) -> None:
        """Browse for the desired mDNS service types and query them."""
        service_types = [DEVICEAPI, PLCNETAPI]
        counter = 0
        self._logger.debug("Browsing for %s", service_types)
        addr = None if self._multicast else self.ip
        question_type = DNSQuestionType.QM if self._multicast else DNSQuestionType.QU
        if self._browser:
            await self._browser.async_cancel()
            self._browser = None
        self._browser = AsyncServiceBrowser(
            zeroconf=self._zeroconf.zeroconf,
            type_=service_types,
            handlers=[self._state_change],
            addr=addr,
            question_type=question_type,
        )
        while (
            not self._info[DEVICEAPI].properties
            or not self._info[PLCNETAPI].properties
            and self.mt_number not in DEVICES_WITHOUT_PLCNET
        ) and counter < self.MDNS_TIMEOUT:
            counter += 1
            await asyncio.sleep(0.01)

    async def _retry_zeroconf_info(self) -> None:
        """Retry getting the zeroconf info using multicast."""
        self._logger.debug("Having trouble getting results via unicast messages. Switching to multicast for this device.")
        self._multicast = True
        await self._get_zeroconf_info()

    def _state_change(self, zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange) -> None:
        """Evaluate the query result."""
        if state_change == ServiceStateChange.Removed:
            return
        task = asyncio.create_task(self._get_service_info(zeroconf, service_type, name))
        self._background_tasks.add(task)
        task.add_done_callback(self._background_tasks.remove)

    async def _get_service_info(self, zeroconf: Zeroconf, service_type: str, name: str) -> None:
        """Get service information, if IP matches."""
        service_info = AsyncServiceInfo(service_type, name)
        update = {
            DEVICEAPI: self._get_device_info,
            PLCNETAPI: self._get_plcnet_info,
        }
        with suppress(RuntimeError):
            if not self._multicast:
                await service_info.async_request(zeroconf, timeout=1000, question_type=DNSQuestionType.QU, addr=self.ip)
            else:
                await service_info.async_request(zeroconf, timeout=1000, question_type=DNSQuestionType.QM)

        if not service_info.addresses or self.ip not in service_info.parsed_addresses():
            return  # No need to continue, if there are no relevant service information

        self._logger.debug("Updating service info of %s for %s", service_type, service_info.server_key)
        if info := self.info_from_service(service_info):
            self._info[service_type] = info
            await update[service_type]()

    @staticmethod
    def info_from_service(service_info: ServiceInfo) -> ZeroconfServiceInfo | None:
        """Return prepared info from mDNS entries."""
        properties = {}
        total_length = len(service_info.text)
        offset = 0
        while offset < total_length:
            (parsed_length,) = 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

        return ZeroconfServiceInfo(
            address=service_info.addresses[0],
            hostname=service_info.server or "",
            port=service_info.port,
            properties=properties,
        )

1	"""Representation of your devolo device."""
2	from __future__ import annotations	1✔
3
4	import asyncio	1✔
5	import logging	1✔
6	from contextlib import suppress	1✔
7	from datetime import date	1✔
8	from ipaddress import ip_address, ip_network	1✔
9	from struct import unpack_from	1✔
10	from typing import TYPE_CHECKING, cast	1✔
11
12	from httpx import AsyncClient	1✔
13	from ifaddr import get_adapters	1✔
14	from zeroconf import DNSQuestionType, ServiceInfo, ServiceStateChange, Zeroconf	1✔
15	from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf	1✔
16
17	from .device_api import SERVICE_TYPE as DEVICEAPI, DeviceApi	1✔
18	from .exceptions import DeviceNotFound	1✔
19	from .plcnet_api import DEVICES_WITHOUT_PLCNET, SERVICE_TYPE as PLCNETAPI, PlcNetApi	1✔
20	from .zeroconf import ZeroconfServiceInfo	1✔
21
22	if TYPE_CHECKING:	1✔
23	from types import TracebackType	×
24
25	from typing_extensions import Self	×
26
27
28	class Device:	1✔
29	"""
30	Representing object for your devolo PLC device. It stores all properties and functionalities discovered during setup.
31
32	:param ip: IP address of the device to communicate with.
33	:param plcnetapi: Reuse externally gathered data for the plcnet API
34	:param deviceapi: Reuse externally gathered data for the device API
35	:param zeroconf_instance: Zeroconf instance to be potentially reused.
36	"""
37
38	MDNS_TIMEOUT = 300	1✔
39
40	def __init__(	1✔
41	self,
42	ip: str,
43	zeroconf_instance: AsyncZeroconf \| Zeroconf \| None = None,
44	) -> None:
45	"""Initialize the device."""
46	self.ip = ip	1✔
47	self.mac = ""	1✔
48	self.mt_number = "0"	1✔
49	self.product = ""	1✔
50	self.technology = ""	1✔
51	self.serial_number = "0"	1✔
52
53	self.device: DeviceApi \| None = None	1✔
54	self.plcnet: PlcNetApi \| None = None	1✔
55
56	self._background_tasks: set[asyncio.Task] = set()	1✔
57	self._browser: AsyncServiceBrowser \| None = None	1✔
58	self._connected = False	1✔
59	self._info: dict[str, ZeroconfServiceInfo] = {PLCNETAPI: ZeroconfServiceInfo(), DEVICEAPI: ZeroconfServiceInfo()}	1✔
60	self._logger = logging.getLogger(f"{self.__class__.__module__}.{self.__class__.__name__}")	1✔
61	self._multicast = False	1✔
62	self._password = ""	1✔
63	self._session_instance: AsyncClient \| None = None	1✔
64	self._zeroconf_instance = zeroconf_instance	1✔
65	logging.captureWarnings(capture=True)	1✔
66
67	self._loop: asyncio.AbstractEventLoop	1✔
68	self._session: AsyncClient	1✔
69	self._zeroconf: AsyncZeroconf	1✔
70
71	def __del__(self) -> None:	1✔
72	"""Warn user, if the connection was not properly closed."""
73	if self._connected and self._session_instance is None:	1✔
74	self._logger.warning("Please disconnect properly from the device.")	1✔
75
76	async def __aenter__(self) -> Self:	1✔
77	"""Connect to a device asynchronously when entering a context manager."""
78	await self.async_connect()	1✔
79	return self	1✔
80
81	async def __aexit__(	1✔
82	self, exc_type: type[BaseException] \| None, exc_val: BaseException \| None, exc_tb: TracebackType \| None
83	) -> None:
84	"""Disconnect to a device asynchronously when entering a context manager."""
85	await self.async_disconnect()	1✔
86
87	def __enter__(self) -> Self:	1✔
88	"""Connect to a device synchronously when leaving a context manager."""
89	self.connect()	1✔
90	return self	1✔
91
92	def __exit__(	1✔
93	self, exc_type: type[BaseException] \| None, exc_val: BaseException \| None, exc_tb: TracebackType \| None
94	) -> None:
95	"""Disconnect to a device synchronously when leaving a context manager."""
96	self.disconnect()	1✔
97
98	@property	1✔
99	def firmware_date(self) -> date:	1✔
100	"""Date the firmware was built."""
101	return date.fromisoformat(self._info[DEVICEAPI].properties.get("FirmwareDate", "1970-01-01")[:10])	1✔
102
103	@property	1✔
104	def firmware_version(self) -> str:	1✔
105	"""Firmware version currently installed."""
106	return self._info[DEVICEAPI].properties.get("FirmwareVersion", "")	1✔
107
108	@property	1✔
109	def hostname(self) -> str:	1✔
110	"""mDNS hostname of the device.""" # noqa: D403
111	return self._info[DEVICEAPI].hostname	1✔
112
113	@property	1✔
114	def password(self) -> str:	1✔
115	"""The currently set device password."""
116	return self._password	1✔
117
118	@password.setter	1✔
119	def password(self, password: str) -> None:	1✔
120	"""Change the currently set device password."""
121	self._password = password	1✔
122	if self.device:	1✔
123	self.device.password = password	1✔
124	if self.plcnet:	1✔
125	self.plcnet.password = password	1✔
126
127	async def async_connect(self, session_instance: AsyncClient \| None = None) -> None:	1✔
128	"""
129	Connect to a device asynchronous.
130
131	:param: session_instance: Session client instance to be potentially reused.
132	"""
133	self._session_instance = session_instance	1✔
134	self._session = self._session_instance or AsyncClient()	1✔
135	if not self._zeroconf_instance:	1✔
136	self._zeroconf = AsyncZeroconf(interfaces=await self._get_relevant_interfaces())	1✔
137	elif isinstance(self._zeroconf_instance, Zeroconf):	×
138	self._zeroconf = AsyncZeroconf(zc=self._zeroconf_instance)	×
139	else:
140	self._zeroconf = self._zeroconf_instance	×
141	await self._get_zeroconf_info()	1✔
142	if not self._info[DEVICEAPI].properties and not self._info[PLCNETAPI].properties:	1✔
143	await self._retry_zeroconf_info()	1✔
144	if not self.device and not self.plcnet:	1✔
145	raise DeviceNotFound(self.ip)	1✔
146	self._connected = True	1✔
147
148	def connect(self) -> None:	1✔
149	"""Connect to a device synchronous."""
150	self._loop = asyncio.get_event_loop()	1✔
151	self._loop.run_until_complete(self.async_connect())	1✔
152
153	async def async_disconnect(self) -> None:	1✔
154	"""Disconnect from a device asynchronous."""
155	if self._connected:	1✔
156	if self._browser:	1✔
157	await self._browser.async_cancel()	1✔
158	if not self._zeroconf_instance:	1✔
159	await self._zeroconf.async_close()	1✔
160	if not self._session_instance:	1✔
161	await self._session.aclose()	1✔
162	self._connected = False	1✔
163
164	def disconnect(self) -> None:	1✔
165	"""Disconnect from a device synchronous."""
166	self._loop.run_until_complete(self.async_disconnect())	1✔
167
168	async def _get_relevant_interfaces(self) -> list[str]:	1✔
169	"""Get the IP address of the relevant interface to reduce traffic."""
170	interface: list[str] = []	1✔
171	for adapter in get_adapters():	1✔
172	interface.extend(	1✔
173	cast(str, ip.ip)
174	for ip in adapter.ips
175	if ip.is_IPv4 and ip_address(self.ip) in ip_network(f"{ip.ip}/{ip.network_prefix}", strict=False)
176	)
177	interface.extend(	1✔
178	cast(str, ip.ip[0])
179	for ip in adapter.ips
180	if ip.is_IPv6 and ip_address(self.ip) in ip_network(f"{ip.ip[0]}/{ip.network_prefix}", strict=False)
181	)
182	return interface	1✔
183
184	async def _get_device_info(self) -> None:	1✔
185	"""Get information from the devolo Device API."""
186	service_type = DEVICEAPI	1✔
187	if self._info[service_type].properties:	1✔
188	self.mt_number = self._info[service_type].properties.get("MT", "0")	1✔
189	self.product = self._info[service_type].properties.get("Product", "")	1✔
190	self.serial_number = self._info[service_type].properties["SN"]	1✔
191	self.device = DeviceApi(	1✔
192	ip=str(ip_address(self._info[service_type].address)),
193	session=self._session,
194	info=self._info[service_type],
195	)
196	self.device.password = self.password	1✔
197
198	async def _get_plcnet_info(self) -> None:	1✔
199	"""Get information from the devolo PlcNet API."""
200	service_type = PLCNETAPI	1✔
201	if self._info[service_type].properties:	1✔
202	self.mac = self._info[service_type].properties["PlcMacAddress"]	1✔
203	self.technology = self._info[service_type].properties.get("PlcTechnology", "")	1✔
204	self.plcnet = PlcNetApi(	1✔
205	ip=str(ip_address(self._info[service_type].address)),
206	session=self._session,
207	info=self._info[service_type],
208	)
209	self.plcnet.password = self.password	1✔
210
211	async def _get_zeroconf_info(self) -> None:	1✔
212	"""Browse for the desired mDNS service types and query them."""
213	service_types = [DEVICEAPI, PLCNETAPI]	1✔
214	counter = 0	1✔
215	self._logger.debug("Browsing for %s", service_types)	1✔
216	addr = None if self._multicast else self.ip	1✔
217	question_type = DNSQuestionType.QM if self._multicast else DNSQuestionType.QU	1✔
218	if self._browser:	1✔
219	await self._browser.async_cancel()	1✔
220	self._browser = None	1✔
221	self._browser = AsyncServiceBrowser(	1✔
222	zeroconf=self._zeroconf.zeroconf,
223	type_=service_types,
224	handlers=[self._state_change],
225	addr=addr,
226	question_type=question_type,
227	)
228	while (	1✔
229	not self._info[DEVICEAPI].properties
230	or not self._info[PLCNETAPI].properties
231	and self.mt_number not in DEVICES_WITHOUT_PLCNET
232	) and counter < self.MDNS_TIMEOUT:
233	counter += 1	1✔
234	await asyncio.sleep(0.01)	1✔
235
236	async def _retry_zeroconf_info(self) -> None:	1✔
237	"""Retry getting the zeroconf info using multicast."""
238	self._logger.debug("Having trouble getting results via unicast messages. Switching to multicast for this device.")	1✔
239	self._multicast = True	1✔
240	await self._get_zeroconf_info()	1✔
241
242	def _state_change(self, zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange) -> None:	1✔
243	"""Evaluate the query result."""
244	if state_change == ServiceStateChange.Removed:	1✔
245	return	1✔
246	task = asyncio.create_task(self._get_service_info(zeroconf, service_type, name))	1✔
247	self._background_tasks.add(task)	1✔
248	task.add_done_callback(self._background_tasks.remove)	1✔
249
250	async def _get_service_info(self, zeroconf: Zeroconf, service_type: str, name: str) -> None:	1✔
251	"""Get service information, if IP matches."""
252	service_info = AsyncServiceInfo(service_type, name)	1✔
253	update = {	1✔
254	DEVICEAPI: self._get_device_info,
255	PLCNETAPI: self._get_plcnet_info,
256	}
257	with suppress(RuntimeError):	1✔
258	if not self._multicast:	1✔
259	await service_info.async_request(zeroconf, timeout=1000, question_type=DNSQuestionType.QU, addr=self.ip)	1✔
260	else:
261	await service_info.async_request(zeroconf, timeout=1000, question_type=DNSQuestionType.QM)	1✔
262
263	if not service_info.addresses or self.ip not in service_info.parsed_addresses():	1✔
264	return # No need to continue, if there are no relevant service information	1✔
265
266	self._logger.debug("Updating service info of %s for %s", service_type, service_info.server_key)	1✔
267	if info := self.info_from_service(service_info):	1✔
268	self._info[service_type] = info	1✔
269	await update[service_type]()	1✔
270
271	@staticmethod	1✔
272	def info_from_service(service_info: ServiceInfo) -> ZeroconfServiceInfo \| None:	1✔
273	"""Return prepared info from mDNS entries."""
274	properties = {}	1✔
275	total_length = len(service_info.text)	1✔
276	offset = 0	1✔
277	while offset < total_length:	1✔
278	(parsed_length,) = unpack_from("!B", service_info.text, offset)	1✔
279	key_value = service_info.text[offset + 1 : offset + 1 + parsed_length].decode("UTF-8").split("=")	1✔
280	properties[key_value[0]] = key_value[1]	1✔
281	offset += parsed_length + 1	1✔
282
283	return ZeroconfServiceInfo(	1✔
284	address=service_info.addresses[0],
285	hostname=service_info.server or "",
286	port=service_info.port,
287	properties=properties,
288	)

2Fake / devolo_plc_api / 6080961227

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous