Commit 10d37bda authored by Dom Sekotill's avatar Dom Sekotill
Browse files

Add {add,set}_network methods to InterfaceClient

parent 0f4c52b0
Loading
Loading
Loading
Loading
+1 −0
Original line number Diff line number Diff line
@@ -89,6 +89,7 @@ reports = no
output-format = colorized
indent-string = '\t'
max-attributes = 10
const-rgx = [A-Z][A-Z0-9_]{2,30}$|_{1,2}[a-z][a-z0-9_]{2,30}$

disable =
  bad-continuation,
+65 −0
Original line number Diff line number Diff line
@@ -17,11 +17,13 @@ Test cases for wpa_supplicant.client.interfaces.InterfaceClient
"""

import unittest
from unittest.mock import call

import anyio

from tests.unit import anyio as anyio_mock

from wpa_supplicant import config
from wpa_supplicant.client import interfaces


@@ -57,3 +59,66 @@ class MethodsTests(unittest.TestCase):
			async for bss in client.scan():
				self.assertIsInstance(bss, dict)
				self.assertIn("good", bss)

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

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

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

			with self.subTest("good enum"):
				await client.set_network("0", "key_mgmt", config.KeyManagementValue.WPA_PSK)
				send.assert_called_once_with(b"SET_NETWORK 0 key_mgmt WPA-PSK")
				send.reset_mock()

			with self.subTest("int not a string"), self.assertRaises(TypeError):
				await client.set_network("0", "ssid", 1)
				send.reset_mock()

			with self.subTest("string not an int"), self.assertRaises(TypeError):
				await client.set_network("0", "priority", "example")
				send.reset_mock()

			with self.subTest("string not an enum"), self.assertRaises(TypeError):
				await client.set_network("0", "key_mgmt", "example")
				send.reset_mock()

			with self.subTest("int not an enum"), self.assertRaises(TypeError):
				await client.set_network("0", "key_mgmt", 1)
				send.reset_mock()

	@anyio_mock.with_anyio()
	async def test_add_network(self):
		"""
		Check that add_network adds a new network and configures it
		"""
		async with self.client as client:
			send = client.sock.send
			client.sock.recv.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"),
				]
			)
+1 −1
Original line number Diff line number Diff line
@@ -114,7 +114,7 @@ class BaseClient:
		self,
		message: str,
		*args: str,
		separator: str = "\t",
		separator: str = consts.SEPARATOR_TAB,
		expect: str = consts.RESPONSE_OK,
		convert: Optional[Callable] = None,
	) -> Any:
+12 −0
Original line number Diff line number Diff line
@@ -36,6 +36,18 @@ COMMAND_INTERFACE_ADD = "INTERFACE_ADD"
COMMAND_INTERFACE_REMOVE = "INTERFACE_REMOVE"
COMMAND_SCAN = "SCAN"
COMMAND_BSS = "BSS"
COMMAND_ADD_NETWORK = "ADD_NETWORK"
COMMAND_REMOVE_NETWORK = "REMOVE_NETWORK"
COMMAND_SET_NETWORK = "SET_NETWORK"
COMMAND_GET_NETWORK = "GET_NETWORK"
COMMAND_ENABLE_NETWORK = "ENABLE_NETWORK"
COMMAND_DISABLE_NETWORK = "DISABLE_NETWORK"


# Common separators

SEPARATOR_SPACE = " "
SEPARATOR_TAB = "\t"


# Unvalued and failure responses
+22 −2
Original line number Diff line number Diff line
@@ -17,11 +17,11 @@ Interfaces control client class
"""

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

from . import consts
from .base import BaseClient
from .. import util
from .. import config, util


class InterfaceClient(BaseClient):
@@ -51,3 +51,23 @@ class InterfaceClient(BaseClient):
				if not bss:
					return
				yield bss

	async def add_network(self, configuration: Dict[str, Any]) -> int:
		"""Add a new network configuration"""
		netid = await self.send_command(consts.COMMAND_ADD_NETWORK, convert=str)
		for var, val in configuration.items():
			await self.set_network(netid, var, val)
		await self.send_command(consts.COMMAND_ENABLE_NETWORK, netid)
		return int(netid)

	async def set_network(self, netid: str, variable: str, value: Any):
		"""Set a network configuration option"""
		if not isinstance(value, config.get_type(variable)):
			raise TypeError(f"Wrong type for {variable}: {value!r}")
		await self.send_command(
			consts.COMMAND_SET_NETWORK,
			netid,
			variable,
			f'"{value}"' if isinstance(value, str) else str(value),
			separator=consts.SEPARATOR_SPACE,
		)
Loading