Complete typing across the code base (f9a479c0) · Commits · Dom Sekotill / wpa-supplicant-client

setup.cfg

+2 −2

Original line number	Diff line number	Diff line
		@@ -151,8 +151,8 @@ ignore =
		; Prefer B950 implementation
		E501

		;[ multiple statements on one line (colon) ]
		E701
		;[ multiple statements on one line (%s) ]
		E701 E704

		;[ unexpected number of spaces at start of statement line ]
		;[ unexpected number of tabs and spaces at start of statement line ]

tests/unit/test_master_client.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -70,8 +70,8 @@ class InterfaceMethodsTests(unittest.TestCase):
		)
		# fmt: on

		self.assertListEqual(
		await client.list_interfaces(), ["enp0s0", "enp1s0", "wlp2s0"],
		self.assertEqual(
		await client.list_interfaces(), {"enp0s0", "enp1s0", "wlp2s0"},
		)

		client.sock.send.assert_called_once_with(b"INTERFACES")

wpa_supplicant/client/base.py

+25 −2

Original line number	Diff line number	Diff line
		@@ -31,7 +31,9 @@ from typing import Dict
		from typing import Optional
		from typing import Tuple
		from typing import Type
		from typing import TypeVar
		from typing import Union
		from typing import overload

		import anyio
		from anyio.abc import SocketStream
		@@ -41,6 +43,7 @@ from .._anyio import connect_unix_datagram
		from ..types import PathLike
		from . import consts

		T = TypeVar('T')
		EventInfo = Tuple['EventPriority', str, Optional[str]]

		# 128kB (actual max size slightly less than this)
		@@ -135,14 +138,34 @@ class BaseClient:
		if self.sock:
		await self.sock.aclose()

		@overload
		async def send_command(
		self,
		message: str,
		*args: str,
		separator: str = consts.SEPARATOR_TAB,
		expect: str = consts.RESPONSE_OK,
		convert: Optional[Callable[[Any], Any]] = None,
		) -> Any:
		convert: Callable[[str], T],
		) -> T: ...

		@overload
		async def send_command(
		self,
		message: str,
		*args: str,
		separator: str = consts.SEPARATOR_TAB,
		expect: str = consts.RESPONSE_OK,
		convert: None = None,
		) -> None: ...

		async def send_command(
		self,
		message: str,
		*args: str,
		separator: str = consts.SEPARATOR_TAB,
		expect: str = consts.RESPONSE_OK,
		convert: Optional[Callable[[str], T]] = None,
		) -> Optional[T]:
		"""
		Send a message and await a response

wpa_supplicant/client/interfaces.py

+5 −5

Original line number	Diff line number	Diff line
		# Copyright 2019 Dom Sekotill <dom.sekotill@kodo.org.uk>
		# Copyright 2019-2021 Dom Sekotill <dom.sekotill@kodo.org.uk>
		#
		# Licensed under the Apache License, Version 2.0 (the "License");
		# you may not use this file except in compliance with the License.
		@@ -18,8 +18,8 @@ Interfaces control client class

		from itertools import count
		from typing import Any
		from typing import AsyncGenerator
		from typing import Dict
		from typing import Iterable

		from .. import config
		from .. import util
		@@ -36,11 +36,11 @@ class InterfaceClient(BaseClient):

		name = None

		async def connect(self, path: PathLike):
		async def connect(self, path: PathLike) -> None:
		await super().connect(path)
		self.name = await self.send_command(consts.COMMAND_IFNAME, convert=str)

		async def scan(self) -> Iterable[StringMap]:
		async def scan(self) -> AsyncGenerator[StringMap, None]:
		"""
		Iteratively produces the details of all detectable IEEE 802.11 BSS

		@@ -65,7 +65,7 @@ class InterfaceClient(BaseClient):
		await self.send_command(consts.COMMAND_ENABLE_NETWORK, netid)
		return int(netid)

		async def set_network(self, netid: str, variable: str, value: Any):
		async def set_network(self, netid: str, variable: str, value: Any) -> None:
		"""Set a network configuration option"""
		if not isinstance(value, config.get_type(variable)):
		raise TypeError(f"Wrong type for {variable}: {value!r}")

wpa_supplicant/client/master.py

+18 −7

Original line number	Diff line number	Diff line
		# Copyright 2019 Dom Sekotill <dom.sekotill@kodo.org.uk>
		# Copyright 2019-2021 Dom Sekotill <dom.sekotill@kodo.org.uk>
		#
		# Licensed under the Apache License, Version 2.0 (the "License");
		# you may not use this file except in compliance with the License.
		@@ -32,30 +32,38 @@ class MasterClient(BaseClient):

		ctrl_dir = None

		async def connect(self, path: PathLike):
		async def connect(self, path: PathLike) -> None:
		if not isinstance(path, pathlib.Path):
		path = pathlib.Path(path)
		await super().connect(path)
		self.ctrl_dir = path.parent

		async def list_interfaces(self) -> Set:
		async def list_interfaces(self) -> Set[str]:
		"""
		Return a set of the interfaces currently managed by the daemon
		"""
		return await self.send_command(
		consts.COMMAND_INTERFACES, convert=lambda x: x.splitlines(),
		consts.COMMAND_INTERFACES, convert=lambda x: set(x.splitlines()),
		)

		async def add_interface(self, ifname: str, driver: str = "", driver_param: str = ""):
		async def add_interface(self, ifname: str, driver: str = "", driver_param: str = "") -> None:
		"""
		Add a network interface to the daemon's control interfaces
		"""
		if self.ctrl_dir:
		ctrl_iface = f"DIR={self.ctrl_dir} GROUP={self.ctrl_dir.group()}"
		else:
		# RuntimeError should be raised by send_command() as connect() does not appear
		# to have been called; set ctrl_iface to any string
		ctrl_iface = ""
		await self.send_command(
		consts.COMMAND_INTERFACE_ADD, ifname, "", driver, ctrl_iface, driver_param,
		)
		assert self.ctrl_dir is not None, \
		"RuntimeError should be raised for sends on unconnected clients; " \
		"or connect() may not have set ctrl_dir"

		async def remove_interface(self, ifname: str): # pragma: no cover
		async def remove_interface(self, ifname: str) -> None: # pragma: no cover
		"""
		Remove a network interface from the daemon's control
		"""
		@@ -70,6 +78,9 @@ class MasterClient(BaseClient):
		"""
		if ifname not in await self.list_interfaces():
		await self.add_interface(ifname)
		assert self.ctrl_dir is not None, \
		"RuntimeError should be raised for sends on unconnected clients; " \
		"or connect() may not have set ctrl_dir"
		client = InterfaceClient(logger=self.logger)
		await client.connect(self.ctrl_dir.joinpath(ifname).as_posix())
		return client