test_base_client.py

#  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.base.BaseClient
"""

import unittest
from unittest import mock

import anyio

from tests.unit import anyio as anyio_mock

from wpa_supplicant import errors
from wpa_supplicant.client import base


@mock.patch('wpa_supplicant.util.connect_unix_datagram', new_callable=anyio_mock.AsyncMock)
@mock.patch('wpa_supplicant.client.base.BaseClient.send_command', new_callable=anyio_mock.AsyncMock)
class ConnectTests(unittest.TestCase):
	"""
	Tests for the connect() method
	"""

	@anyio_mock.with_anyio()
	async def test_connect(self, _, connect_mock):
		"""
		Check connect() calls socket.connect()
		"""
		async with base.BaseClient() as client:
			await client.connect("foo")

		connect_mock.assert_called_once_with("foo")

	@anyio_mock.with_anyio()
	async def test_connect_timeout_1(self, _, connect_mock):
		"""
		Check a socket.connect() delay causes TimeoutError to be raised
		"""
		connect_mock.delay = 2

		async with base.BaseClient() as client:
			with self.assertRaises(TimeoutError):
				await client.connect("foo")

	@anyio_mock.with_anyio()
	async def test_connect_timeout_2(self, send_mock, _):
		"""
		Check a send/recv delay causes a TimeoutError to be raised
		"""
		send_mock.delay = 2

		async with base.BaseClient() as client:
			with self.assertRaises(TimeoutError):
				await client.connect("foo")


class SendMessageTests(unittest.TestCase):
	"""
	Tests for the send_command() method
	"""

	def setUp(self):
		async def _make_client():
			return base.BaseClient()
		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_simple(self):
		"""
		Check that a response is processed after a command
		"""
		async with self.client as client:
			client.sock.recv.return_value = b"OK"
			assert await client.send_command("SOME_COMMAND") is None

	@anyio_mock.with_anyio()
	async def test_simple_expect(self):
		"""
		Check that an alternate expected response is processed
		"""
		async with self.client as client:
			client.sock.recv.return_value = b"PONG"
			assert await client.send_command("PING", expect="PONG") is None

	@anyio_mock.with_anyio()
	async def test_simple_no_expect(self):
		"""
		Check that an unexpected response raises an UnexpectedResponseError
		"""
		async with self.client as client:
			client.sock.recv.return_value = b"DING"
			with self.assertRaises(errors.UnexpectedResponseError):
				await client.send_command("PING")
			with self.assertRaises(errors.UnexpectedResponseError):
				await client.send_command("PING", expect="PONG")

	@anyio_mock.with_anyio()
	async def test_simple_convert(self):
		"""
		Check that a response is passed through a converter if given
		"""
		async with self.client as client:
			client.sock.recv.return_value = b"FOO\nBAR\nBAZ\n"
			self.assertListEqual(
				await client.send_command(
					"SOME_COMMAND",
					convert=lambda x: x.splitlines(),
				),
				["FOO", "BAR", "BAZ"]
			)

	@anyio_mock.with_anyio()
	async def test_simple_convert_over_expect(self):
		"""
		Check that 'convert' overrides 'expect'
		"""
		async with self.client as client:
			client.sock.recv.return_value = b"FOO\nBAR\nBAZ\n"
			self.assertListEqual(
				await client.send_command(
					"SOME_COMMAND",
					convert=lambda x: x.splitlines(),
					expect="PONG",
				),
				["FOO", "BAR", "BAZ"]
			)

	@anyio_mock.with_anyio()
	async def test_simple_fail(self):
		"""
		Check that a response of 'FAIL' causes CommandFailed to be raised
		"""
		async with self.client as client:
			client.sock.recv.return_value = b"FAIL"
			with self.assertRaises(errors.CommandFailed):
				await client.send_command("SOME_COMMAND")

	@anyio_mock.with_anyio()
	async def test_simple_bad_command(self):
		"""
		Check that a response of 'UNKNOWN COMMAND' causes ValueError to be raised
		"""
		async with self.client as client:
			client.sock.recv.return_value = b"UNKNOWN COMMAND"
			with self.assertRaises(ValueError):
				await client.send_command("SOME_COMMAND")

	@anyio_mock.with_anyio()
	async def test_simple_too_large(self):
		"""
		Check that MessageTooLargeError is raised if send() reports an incomplete send
		"""
		async with self.client as client:
			client.sock.send.side_effect = lambda b: len(b) / 2
			with self.assertRaises(errors.MessageTooLargeError):
				await client.send_command("SOME_COMMAND")

	@anyio_mock.with_anyio()
	async def test_interleaved(self):
		"""
		Check that messages are processed alongside replies
		"""
		async with self.client as client:
			client.sock.recv.side_effect = [
				b"<2>SOME-MESSAGE",
				b"<1>SOME-OTHER-MESSAGE with|args",
				b"OK",
				b"<2>SOME-MESSAGE"
			]
			assert await client.send_command("SOME_COMMAND") is None


class EventTests(unittest.TestCase):
	"""
	Tests for the event() method
	"""

	def setUp(self):
		async def _make_client():
			return base.BaseClient()
		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_simple(self):
		"""
		Check that an awaited message is returned when is arrives
		"""
		async with self.client as client, anyio.fail_after(2):
			client.sock.recv.side_effect = [
				b"OK",  # Respond to ATTACH
				b"<2>EXAMPLE-EVENT",
				b"OK",  # Respond to DETACH
			]
			prio, evt, args = await client.event("EXAMPLE-EVENT")
			assert prio == 2
			assert evt == "EXAMPLE-EVENT"
			assert args is None

	@anyio_mock.with_anyio()
	async def test_multiple(self):
		"""
		Check that an awaited messages is returned when it arrives between others
		"""
		async with self.client as client, anyio.fail_after(2):
			client.sock.recv.side_effect = [
				b"OK",  # Respond to ATTACH
				b"<1>OTHER-MESSAGE",
				b"<2>OTHER-MESSAGE",
				b"<4>EXAMPLE-EVENT",
				b"OK",  # Respond to DETACH
				b"<3>OTHER-MESSAGE",
			]
			prio, evt, args = await client.event("EXAMPLE-EVENT")
			assert prio == 4
			assert evt == "EXAMPLE-EVENT"
			assert args is None

	@anyio_mock.with_anyio()
	async def test_wait_multiple(self):
		"""
		Check that the first of several awaited events is returned
		"""
		async with self.client as client, anyio.fail_after(2):
			client.sock.recv.side_effect = [
				b"OK",  # Respond to ATTACH
				b"<1>OTHER-MESSAGE",
				b"<2>OTHER-MESSAGE",
				b"<4>EXAMPLE-EVENT3",
				b"<4>EXAMPLE-EVENT1",
				b"OK",  # Respond to DETACH
				b"<3>OTHER-MESSAGE",
			]
			prio, evt, args = await client.event("EXAMPLE-EVENT1", "EXAMPLE-EVENT2", "EXAMPLE-EVENT3")
			assert prio == 4
			assert evt == "EXAMPLE-EVENT3"
			assert args is None

	@anyio_mock.with_anyio()
	async def test_interleaved(self):
		"""
		Check that messages are processed as well as replies
		"""
		async with self.client as client, anyio.fail_after(2):
			client.sock.recv.side_effect = [
				b"<1>OTHER-MESSAGE",
				b"OK",  # Respond to SOME_COMMAND
				b"OK",  # Respond to ATTACH
				b"<2>OTHER-MESSAGE",
				b"<4>EXAMPLE-EVENT",
				b"<3>OTHER-MESSAGE",
				b"OK",  # Respond to DETACH
				b"FOO",
			]

			assert await client.send_command("SOME_COMMAND") is None

			prio, evt, args = await client.event("EXAMPLE-EVENT")
			assert prio == 4
			assert evt == "EXAMPLE-EVENT"
			assert args is None

			assert await client.send_command("SOME_COMMAND", expect="FOO") is None