Implement an InterfaceClient.scan method (bdcecc31) · Commits · Dom Sekotill / wpa-supplicant-client

tests/unit/test_interfaces_client.py

0 → 100644

+58 −0

Original line number	Diff line number	Diff line
		# Copyright 2019 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.
		# You may obtain a copy of the License at
		#
		# http://www.apache.org/licenses/LICENSE-2.0
		#
		# Unless required by applicable law or agreed to in writing, software
		# distributed under the License is distributed on an "AS IS" BASIS,
		# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
		# See the License for the specific language governing permissions and
		# limitations under the License.

		"""
		Test cases for wpa_supplicant.client.interfaces.InterfaceClient
		"""

		import unittest

		import anyio

		from tests.unit import anyio as anyio_mock

		from wpa_supplicant.client import interfaces


		class MethodsTests(unittest.TestCase):
		"""
		Tests for InterfaceClient methods
		"""

		def setUp(self):
		async def _make_client():
		return interfaces.InterfaceClient()
		self.client = client = anyio.run(_make_client)
		client.sock = anyio_mock.AsyncMock()
		client.sock.send.side_effect = len

		@anyio_mock.with_anyio()
		async def test_scan(self):
		"""
		Check that a scan command waits for a notification then terminates correctly
		"""
		async with self.client as client:
		client.sock.recv.side_effect = [
		b"OK", # Response to "ATTACH"
		b"OK", # Response to "SCAN"
		b"<3>CTRL-EVENT-SCAN-RESULTS",
		b"good=value\nid=1\n", # Response to "BSS 1"
		b"good=value\nid=2\n", # Response to "BSS 2"
		b"", # Response to "BSS 3"
		b"OK", # Response to "DETACH"
		]

		async for bss in client.scan():
		self.assertIsInstance(bss, dict)
		self.assertIn('good', bss)

wpa_supplicant/client/consts.py

+7 −0

Original line number	Diff line number	Diff line
		@@ -34,6 +34,8 @@ COMMAND_IFNAME = 'IFNAME'
		COMMAND_INTERFACES = 'INTERFACES'
		COMMAND_INTERFACE_ADD = 'INTERFACE_ADD'
		COMMAND_INTERFACE_REMOVE = 'INTERFACE_REMOVE'
		COMMAND_SCAN = 'SCAN'
		COMMAND_BSS = 'BSS'


		# Unvalued and failure responses
		@@ -42,3 +44,8 @@ RESPONSE_OK = 'OK'
		RESPONSE_PONG = 'PONG'
		RESPONSE_FAIL = 'FAIL'
		RESPONSE_UNKNOWN_COMMAND = 'UNKNOWN COMMAND'


		# Control messages

		CTRL_EVENT_SCAN_RESULTS = 'CTRL-EVENT-SCAN-RESULTS'

wpa_supplicant/client/interfaces.py

+19 −0

Original line number	Diff line number	Diff line
		@@ -16,8 +16,12 @@
		Interfaces control client class
		"""

		from itertools import count
		from typing import Iterable

		from . import consts
		from .base import BaseClient
		from .. import util


		class InterfaceClient(BaseClient):
		@@ -30,3 +34,18 @@ class InterfaceClient(BaseClient):
		async def connect(self, path: consts.Path):
		await super().connect(path)
		self.name = await self.send_command(consts.COMMAND_IFNAME, convert=str)

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

		(WiFi Access Points to you and me)
		"""
		async with self.attach():
		await self.send_command(consts.COMMAND_SCAN)
		await self.event(consts.CTRL_EVENT_SCAN_RESULTS)
		for idx in count():
		bss = await self.send_command(consts.COMMAND_BSS, str(idx), convert=util.kv2dict)
		if not bss:
		return
		yield bss

wpa_supplicant/util.py

+8 −1

Original line number	Diff line number	Diff line
		@@ -21,7 +21,7 @@ import os
		import socket
		import tempfile
		from contextlib import suppress
		from typing import Union
		from typing import Dict, Union

		import anyio

		@@ -49,3 +49,10 @@ async def connect_unix_datagram(path: Union[str, os.PathLike]) -> anyio.SocketSt
		raise
		else:
		return sock


		def kv2dict(keyvalues: str) -> Dict[str, str]:
		"""
		Convert a list of line-terminated "key=value" substrings into a dictionary
		"""
		return dict(kv.split('=', 1) for kv in keyvalues.splitlines())