Add integration tests

#  Copyright 2021  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 connecting and communicating with a server

"""

import os

import sys

from unittest import TestCase

from tests._anyio import with_anyio

from tests.integration.util import start_server

from wpa_supplicant.client import master

class Tests(TestCase):

	"""

	Tests against live wpa_suppplicant servers

	The 'wpa_supplicant' executable is required in a PATH directory for these tests to work.

	"""

	@with_anyio('asyncio', 'trio')

	async def test_connect(self):

		"""

		Test connecting to the global wpa_supplicant control socket

		"""

		async with start_server() as sock, master.MasterClient() as client:

			await client.connect(sock)

			ifaces = await client.list_interfaces()

			assert len(ifaces) == 0

	@with_anyio('asyncio', 'trio')

	async def test_new_interface(self):

		"""

		Test adding a wireless interface and scanning for stations

		This test requires root privileges

		"""

		if os.getuid() != 0:

			self.skipTest("test_new_interface requires root privileges")

		async with start_server() as sock, master.MasterClient() as mclient:

			await mclient.connect(sock)

			async with await mclient.connect_interface('wlan0') as iclient:

				async for iface in iclient.scan():

					sys.stdout.write(

						"{id:>2}: {bssid} {freq} {noise}/{level} {ssid:32} {flags}\n"

						.format(**iface),

					)

#  Copyright 2021  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.

"""

Utilities for integration tests

"""

import tempfile

from contextlib import asynccontextmanager

from pathlib import Path

from typing import AsyncGenerator

import anyio

SOCKET_NAME = 'server.sock'

@asynccontextmanager

async def start_server() -> AsyncGenerator[Path, None]:

	"""

	Return an async context manager that runs a wpa_supplicant server for it's duration

	"""

	with tempfile.TemporaryDirectory() as tmpdir_:

		tmpdir = Path(tmpdir_)

		cmd = ['wpa_supplicant', '-g', str(tmpdir / SOCKET_NAME)]

		async with await anyio.open_process(cmd) as proc:

			# wpa_supplicant is pretty fast; one second to wait for startup should be long

			# enough on most systems…?

			await anyio.sleep(1.0)

			try:

				yield tmpdir / SOCKET_NAME

			finally:

				proc.terminate()