Skip to content

Commit

Permalink
tests: update tests to remove our weird synchronous lifespan.
Browse files Browse the repository at this point in the history
Signed-off-by: Romain Bezut <[email protected]>
  • Loading branch information
morian committed Sep 30, 2024
1 parent 7828bb0 commit dae4e37
Showing 1 changed file with 31 additions and 43 deletions.
74 changes: 31 additions & 43 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,6 @@
import asyncio
import os
from asyncio import create_task, start_unix_server
from contextlib import asynccontextmanager
from queue import Queue
from tempfile import TemporaryDirectory
from typing import TYPE_CHECKING

Expand All @@ -15,7 +13,7 @@

if TYPE_CHECKING:
from asyncio import StreamReader, StreamWriter
from collections.abc import AsyncIterator, Iterator
from collections.abc import Iterator


@pytest.fixture(scope='session')
Expand All @@ -26,11 +24,12 @@ def anyio_backend() -> str:
class FakeServer:
"""Fake server used to spawn fake LD2410 devices."""

def __init__(self, socket_path: str, lifespan):
def __init__(self, socket_path: str) -> None:
"""Create a fake unix socket server used to emulate a real device."""
self.lifespan = lifespan
self.should_exit = False
self.shutdown = asyncio.Event()
self.socket_path = socket_path
self.started = asyncio.Event()
self.stopped = asyncio.Event()
self._tasks = []

async def handle_connection(self, reader: StreamReader, writer: StreamWriter) -> None:
Expand All @@ -39,49 +38,38 @@ async def handle_connection(self, reader: StreamReader, writer: StreamWriter) ->
async with EmulatedDevice(reader, writer) as device:
await device.wait_for_closing()

async def wait_for_shutdown(self) -> None:
"""Wait for the shutdown signal."""
# An asyncio event does not work well here
# This may be because multiple loops are being used.
while not self.should_exit: # noqa: ASYNC110
await asyncio.sleep(0.1)

async def serve(self) -> None:
"""Serve requests until we are told to exit."""
server = await start_unix_server(self.handle_connection, self.socket_path)
async with server, self.lifespan():
task_wait = create_task(self.wait_for_shutdown())
task_serve = create_task(server.serve_forever())
done, pending = await asyncio.wait(
(task_wait, task_serve),
return_when=asyncio.FIRST_COMPLETED,
)
for task in pending:
task.cancel()
for task in self._tasks:
task.cancel()
try:
async with server:
self.started.set()
task_wait = create_task(self.shutdown.wait())
task_serve = create_task(server.serve_forever())
done, pending = await asyncio.wait(
(task_wait, task_serve),
return_when=asyncio.FIRST_COMPLETED,
)
for task in pending:
task.cancel()
for task in self._tasks:
task.cancel()
finally:
self.stopped.set()


@pytest.fixture(scope='class')
def fake_device_socket() -> Iterator[str]:
"""Run a real server in a separate thread."""

# Inspired by https://github.com/frankie567/httpx-ws/blob/main/tests/conftest.py.
q_startup: Queue[bool] = Queue()
q_shutdown: Queue[bool] = Queue()

@asynccontextmanager
async def lifespan() -> AsyncIterator[None]:
q_startup.put(True)
yield
q_shutdown.put(True)

with start_blocking_portal(backend='asyncio') as portal, TemporaryDirectory() as tmp:
server = FakeServer(os.path.join(tmp, 'server.sock'), lifespan)
portal.start_task_soon(server.serve)
try:
q_startup.get(True)
yield server.socket_path
finally:
server.should_exit = True
q_shutdown.get(True)
with TemporaryDirectory() as tmp:
tmp_socket = os.path.join(tmp, 'server.sock')
with start_blocking_portal(backend='asyncio') as portal:
server = FakeServer(tmp_socket)
portal.start_task_soon(server.serve)
try:
portal.call(server.started.wait)
yield server.socket_path
finally:
portal.call(server.shutdown.set)
portal.call(server.stopped.wait)

0 comments on commit dae4e37

Please sign in to comment.