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

import pathlib
import unittest
from unittest.mock import AsyncMock
from unittest.mock import patch

from wpa_supplicant.client import GlobalClient
from wpa_supplicant.client import InterfaceClient


class InterfaceMethodsTests(unittest.IsolatedAsyncioTestCase):
	"""
	Tests for the *_interface(s?) methods
	"""

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

	async def test_connect(self) -> None:
		"""
		Check that connect sets ctrl_dir
		"""
		client1 = GlobalClient()
		client2 = GlobalClient()

		with patch(
			"wpa_supplicant.client.base.BaseClient.connect",
			new_callable=AsyncMock,
		):
			await client1.connect("/tmp/foo/bar")
			await client2.connect(pathlib.Path("/tmp/foo/bar"))

		self.assertIsInstance(client1.ctrl_dir, pathlib.Path)
		self.assertIsInstance(client2.ctrl_dir, pathlib.Path)

		assert client1.ctrl_dir == pathlib.Path("/tmp/foo")
		assert client2.ctrl_dir == pathlib.Path("/tmp/foo")

	async def test_list_interfaces(self) -> None:
		"""
		Check list_interfaces() processes lines of names in a list
		"""
		async with self.client as client:
			# fmt: off
			client.sock.receive.return_value = (
				b"enp0s0\n"
				b"enp1s0\n"
				b"wlp2s0\n"
			)
			# fmt: on

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

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

	async def test_add_interface(self) -> None:
		"""
		Check add_interface() sends the correct arguments
		"""
		async with self.client as client:
			client.ctrl_dir = pathlib.Path("/tmp")
			client.sock.receive.return_value = b"OK"

			assert await client.add_interface("enp1s0", driver="wired") is None

			client.sock.send.assert_called_once()
			args = client.sock.send.call_args[0]
			assert args[0].startswith(b"INTERFACE_ADD enp1s0\t\twired\tDIR=/tmp GROUP=")

	@patch(
		"wpa_supplicant.client.interfaces.InterfaceClient.connect",
		new_callable=AsyncMock,
	)
	async def test_connect_interface(self, connect_mock: AsyncMock) -> None:
		"""
		Check connect_interface() returns a connected InterfaceClient
		"""
		async with self.client as client:
			client.ctrl_dir = pathlib.Path("/tmp")
			client.sock.receive.side_effect = [
				b"enp1s0\n",  # Response to INTERFACES
			]

			ifclient = await client.connect_interface("enp1s0")

			self.assertIsInstance(ifclient, InterfaceClient)
			connect_mock.assert_called_once_with(pathlib.Path("/tmp/enp1s0"))

			# Check only INTERFACES was sent
			client.sock.send.assert_called_once_with(b"INTERFACES")

	@patch(
		"wpa_supplicant.client.interfaces.InterfaceClient.connect",
		new_callable=AsyncMock,
	)
	async def test_connect_interface_with_add(self, connect_mock: AsyncMock) -> None:
		"""
		Check connect_interface() adds the interface when not already managed
		"""
		async with self.client as client:
			client.ctrl_dir = pathlib.Path("/tmp")
			client.sock.receive.side_effect = [
				b"",  # Response to INTERFACES
				b"OK",  # Response to INTERFACE_ADD
			]

			ifclient = await client.connect_interface("enp1s0")

			self.assertIsInstance(ifclient, InterfaceClient)
			connect_mock.assert_called_once_with(pathlib.Path("/tmp/enp1s0"))

			# Check INTERFACE_ADD sent after INTERFACES
			args = client.sock.send.call_args_list
			self.assertTupleEqual(args[0][0], (b"INTERFACES",))
			assert args[1][0][0].startswith(b"INTERFACE_ADD enp1s0\t")

	async def test_unconnected(self) -> None:
		"""
		Check that calling add_interface() on an unconnected client raises RuntimeError

		This is similar to the identically-named test on the base class, but checks there
		are no issues related to the socket directory not being set.
		"""
		client = GlobalClient()

		with self.assertRaises(RuntimeError):
			await client.add_interface("interface")