test_interfaces_client.py

#  Copyright 2019-2021, 2024  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
from collections.abc import Iterator
from contextlib import contextmanager
from unittest.mock import AsyncMock
from unittest.mock import call

from wpa_supplicant import config
from wpa_supplicant.client import interfaces


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

	def setUp(self) -> None:
		self.client = client = interfaces.InterfaceClient()
		client.sock = AsyncMock()
		client.sock.send.return_value = None

	@contextmanager
	def subTest(self, *args: object, reset: list[AsyncMock] = [], **kwargs: object) -> Iterator[None]:
		with super().subTest(*args, **kwargs):
			try:
				yield
			finally:
				for mock in reset:
					mock.reset_mock()

	async def test_scan(self) -> None:
		"""
		Check that a scan command waits for a notification then terminates correctly
		"""
		async with self.client as client:
			client.sock.receive.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)

	async def test_set_network(self) -> None:
		"""
		Check that set_network sends values to the daemon and raises TypeError for bad types
		"""
		async with self.client as client:
			client.sock.receive.return_value = b"OK"
			send = client.sock.send

			with self.subTest("good string", reset=[send]):
				await client.set_network("0", "ssid", "example")
				send.assert_called_once_with(b'SET_NETWORK 0 ssid "example"')

			with self.subTest("good int", reset=[send]):
				await client.set_network("0", "priority", 1)
				send.assert_called_once_with(b"SET_NETWORK 0 priority 1")

			with self.subTest("good enum", reset=[send]):
				await client.set_network("0", "key_mgmt", config.KeyManagementValue.WPA_PSK)  # type: ignore
				send.assert_called_once_with(b"SET_NETWORK 0 key_mgmt WPA-PSK")

			with \
					self.subTest("int not a string", reset=[send]), \
					self.assertRaises(TypeError):
				await client.set_network("0", "ssid", 1)

			with \
					self.subTest("string not an int", reset=[send]), \
					self.assertRaises(TypeError):
				await client.set_network("0", "priority", "example")

			with \
					self.subTest("string not an enum", reset=[send]), \
					self.assertRaises(TypeError):
				await client.set_network("0", "key_mgmt", "example")

			with \
					self.subTest("int not an enum", reset=[send]), \
					self.assertRaises(TypeError):
				await client.set_network("0", "key_mgmt", 1)

	async def test_add_network(self) -> None:
		"""
		Check that add_network adds a new network and configures it
		"""
		async with self.client as client:
			send = client.sock.send
			client.sock.receive.side_effect = [
				b"0",
				b"OK",
				b"OK",
			]

			await client.add_network({"ssid": "example"})

			send.assert_has_calls(
				[
					call(b"ADD_NETWORK"),
					call(b'SET_NETWORK 0 ssid "example"'),
					call(b"ENABLE_NETWORK 0"),
				],
			)