Add a test for interleaved commands & events (3cc45de7) · Commits · Dom Sekotill / wpa-supplicant-client

tests/unit/test_base_client.py

+33 −0

Original line number	Diff line number	Diff line
		@@ -200,6 +200,39 @@ class SendMessageTests(unittest.TestCase):
		with self.assertRaises(RuntimeError):
		await client.send_command("SOME_COMMAND")

		@anyio_mock.with_anyio()
		async def test_multi_task(self):
		"""
		Check that calling send_command() from multiple tasks works as expected
		"""
		recv_responses = iter([
		b"OK", # Response to ATTACH
		b"OK", # Response to SOME_COMMAND1
		b"<2>CTRL-FOO", # Event
		b"REPLY2", # Response to SOME_COMMAND2
		b"OK", # Response to DETACH
		])

		async def recv():
		await anyio.sleep(0.1)
		return next(recv_responses)

		async with self.client as client, anyio.create_task_group() as task_group:
		client.sock.receive.side_effect = recv

		@task_group.start_soon
		async def wait_for_event():
		self.assertTupleEqual(
		await client.event("CTRL-FOO"),
		(base.EventPriority.INFO, "CTRL-FOO", None),
		)

		await anyio.sleep(0.0)
		task_group.start_soon(client.send_command, "SOME_COMMAND1")

		await anyio.sleep(0.0)
		await client.send_command("SOME_COMMAND2", expect="REPLY2")


		class EventTests(unittest.TestCase):
		"""