FIX: Python mockliveserver on Windows

Author: nmacholl

pyproject.toml

@@ -76,4 +76,4 @@ warn_unused_ignores = true
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
-addopts = "--asyncio-mode auto"
+asyncio_mode = "auto"

tests/mockliveserver/__main__.py

@@ -11,7 +11,7 @@
 from databento.common.publishers import Dataset
 from databento_dbn import Schema
 
-from tests.mockliveserver.controller import CommandProtocol
+from tests.mockliveserver.controller import Controller
 from tests.mockliveserver.source import ReplayProtocol
 
 from .server import MockLiveServerProtocol
@@ -84,13 +84,11 @@ async def main() -> None:
     ip, port, *_ = server._sockets[-1].getsockname()  # type: ignore [attr-defined]
 
     # Create command interface for stdin
-    await loop.connect_read_pipe(
-        protocol_factory=lambda: CommandProtocol(
-            api_key_table=api_key_table,
-            file_replay_table=file_replay_table,
-            server=server,
-        ),
-        pipe=sys.stdin,
+    _ = Controller(
+        server=server,
+        api_key_table=api_key_table,
+        file_replay_table=file_replay_table,
+        loop=loop,
     )
 
     # Log Arguments

tests/mockliveserver/controller.py

@@ -3,6 +3,7 @@
 import argparse
 import asyncio
 import logging
+import sys
 from collections.abc import Mapping
 from collections.abc import MutableMapping
 from pathlib import Path
@@ -18,7 +19,7 @@
 logger = logging.getLogger(__name__)
 
 
-class CommandProtocol(asyncio.Protocol):
+class Controller:
     command_parser = argparse.ArgumentParser(prog="mockliveserver")
     subparsers = command_parser.add_subparsers(dest="command")
 
@@ -45,47 +46,43 @@ def __init__(
         server: asyncio.base_events.Server,
         api_key_table: Mapping[str, set[str]],
         file_replay_table: MutableMapping[tuple[Dataset, Schema], ReplayProtocol],
+        loop: asyncio.AbstractEventLoop,
     ) -> None:
         self._server = server
         self._api_key_table = api_key_table
         self._file_replay_table = file_replay_table
-
-    def eof_received(self) -> bool | None:
-        self._server.close()
-        return super().eof_received()
-
-    def data_received(self, data: bytes) -> None:
-        logger.debug("%d bytes from stdin", len(data))
-        try:
-            command_str = data.decode("utf-8")
-        except Exception:
-            logger.error("error parsing command")
-            raise
-
-        for command in command_str.splitlines():
-            params = self.command_parser.parse_args(command.split())
-            command_func = getattr(self, f"_command_{params.command}", None)
-            if command_func is None:
-                raise ValueError(f"{params.command} does not have a command handler")
+        self._loop = loop
+
+        self._read_task = loop.create_task(self._read_commands())
+
+    async def _read_commands(self) -> None:
+        while self._server.is_serving():
+            line = await self._loop.run_in_executor(None, sys.stdin.readline)
+            self.data_received(line.strip())
+
+    def data_received(self, command_str: str) -> None:
+        params = self.command_parser.parse_args(command_str.split())
+        command_func = getattr(self, f"_command_{params.command}", None)
+        if command_func is None:
+            raise ValueError(f"{params.command} does not have a command handler")
+        else:
+            logger.info("received command: %s", command_str)
+            command_params = dict(params._get_kwargs())
+            command_params.pop("command")
+            try:
+                command_func(**command_params)
+            except Exception:
+                logger.exception("error processing command: %s", params.command)
+                print(f"nack: {command_str}", flush=True)
             else:
-                logger.info("received command: %s", command)
-                command_params = dict(params._get_kwargs())
-                command_params.pop("command")
-                try:
-                    command_func(**command_params)
-                except Exception:
-                    logger.exception("error processing command: %s", params.command)
-                    print(f"nack: {command}", flush=True)
-                else:
-                    print(f"ack: {command}", flush=True)
-
-        return super().data_received(data)
+                print(f"ack: {command_str}", flush=True)
 
     def _command_close(self, *_: str) -> None:
         """
         Close the server.
         """
-        self._server.close()
+        self._read_task.cancel()
+        self._loop.call_soon(self._server.close)
 
     def _command_active_count(self, *_: str) -> None:
         """

tests/mockliveserver/fixture.py

@@ -4,7 +4,6 @@
 import contextlib
 import os
 import pathlib
-import signal
 import sys
 from asyncio.subprocess import Process
 from collections.abc import AsyncGenerator
@@ -70,13 +69,22 @@ def stdout(self) -> asyncio.StreamReader:
             return self._process.stdout
         raise RuntimeError("no stream reader for stdout")
 
-    async def _send_command(self, command: str) -> None:
+    async def _send_command(
+        self,
+        command: str,
+        timeout: float = 1.0,
+    ) -> None:
         if self._process.stdin is None:
             raise RuntimeError("cannot write command to mock live server")
         self._process.stdin.write(
             f"{command.strip()}\n".encode(),
         )
-        line = await self.stdout.readline()
+
+        try:
+            line = await asyncio.wait_for(self.stdout.readline(), timeout)
+        except asyncio.TimeoutError:
+            raise RuntimeError("timeout waiting for command acknowledgement")
+
         line_str = line.decode("utf-8")
 
         if line_str.startswith(f"ack: {command}"):
@@ -140,9 +148,9 @@ async def close(self) -> None:
 
     def kill(self) -> None:
         """
-        Kill the mock live server by sending SIGKILL.
+        Kill the mock live server.
         """
-        self._process.send_signal(signal.SIGKILL)
+        self._process.kill()
 
     @contextlib.contextmanager
     def test_context(self) -> Generator[None, None, None]:
@@ -214,6 +222,7 @@ async def fixture_mock_live_server(
         "--echo",
         echo_file.resolve(),
         "--verbose",
+        executable=sys.executable,
         stdin=asyncio.subprocess.PIPE,
         stdout=asyncio.subprocess.PIPE,
         stderr=sys.stderr,
@@ -236,5 +245,5 @@ async def fixture_mock_live_server(
 
     yield interface
 
-    interface.kill()
-    await asyncio.wait_for(process.wait(), timeout=1)
+    process.terminate()
+    await process.wait()

tests/test_live_client.py

@@ -5,7 +5,6 @@
 from __future__ import annotations
 
 import pathlib
-import platform
 import random
 import string
 from io import BytesIO
@@ -894,7 +893,6 @@ def test_live_add_stream_path_directory(
         live_client.add_stream(tmp_path)
 
 
-@pytest.mark.skipif(platform.system() == "Windows", reason="flaky on windows")
 async def test_live_async_iteration(
     live_client: client.Live,
 ) -> None:
@@ -998,7 +996,6 @@ async def test_live_async_iteration_dropped(
     assert live_client._dbn_queue.empty()
 
 
-@pytest.mark.skipif(platform.system() == "Windows", reason="flaky on windows")
 async def test_live_async_iteration_stop(
     live_client: client.Live,
 ) -> None:
@@ -1025,7 +1022,6 @@ async def test_live_async_iteration_stop(
     assert live_client._dbn_queue.empty()
 
 
-@pytest.mark.skipif(platform.system() == "Windows", reason="flaky on windows")
 def test_live_sync_iteration(
     live_client: client.Live,
 ) -> None: