Improve windows compatibility

aioconsole/server.py

@@ -4,6 +4,7 @@
 import socket
 from functools import partial
 
+from . import compat
 from . import console
 
 
@@ -23,6 +24,8 @@ async def start_interactive_server(
     *,
     loop=None,
 ):
+    if compat.platform == "win32" and port is None:
+        raise ValueError("A TCP port should be provided")
     if (port is None) == (path is None):
         raise ValueError("Either a TCP port or a UDS path should be provided")
     if port is not None:
@@ -65,7 +68,8 @@ def factory(streams):
 
 def print_server(server, name="console", file=None):
     interface = server.sockets[0].getsockname()
-    if server.sockets[0].family != socket.AF_UNIX:
+    AF_UNIX = None if compat.platform == "win32" else socket.AF_UNIX
+    if server.sockets[0].family != AF_UNIX:
         interface = "{}:{}".format(*interface)
     print(f"The {name} is being served on {interface}", file=file)
 

tests/test_apython.py

@@ -1,14 +1,12 @@
 import io
 import sys
-import tempfile
 from contextlib import contextmanager
 
 from unittest.mock import Mock, patch, call
 
 import pytest
 
-from aioconsole import apython, rlwrap
-from aioconsole import InteractiveEventLoop
+from aioconsole import InteractiveEventLoop, apython, rlwrap, compat
 
 
 startupfile = """
@@ -28,12 +26,6 @@ def hehe():
 """
 
 
-@pytest.fixture
-def tempfd():
-    with tempfile.NamedTemporaryFile() as tf:
-        yield tf
-
-
 @contextmanager
 def mock_module(name):
     try:
@@ -177,6 +169,9 @@ def run_forever(self, orig=InteractiveEventLoop.run_forever):
     assert err == ""
 
 
+@pytest.mark.skipif(
+    compat.platform == "win32", reason="apython does not run in a subprocess on windows"
+)
 def test_apython_non_existing_file(capfd):
     with pytest.raises(SystemExit):
         apython.run_apython(["idontexist.py"])
@@ -186,6 +181,9 @@ def test_apython_non_existing_file(capfd):
     assert "idontexist.py" in err
 
 
+@pytest.mark.skipif(
+    compat.platform == "win32", reason="apython does not run in a subprocess on windows"
+)
 def test_apython_non_existing_module(capfd):
     with pytest.raises(SystemExit):
         apython.run_apython(["-m", "idontexist"])
@@ -194,11 +192,10 @@ def test_apython_non_existing_module(capfd):
     assert "No module named idontexist" in err
 
 
-def test_apython_pythonstartup(capfd, use_readline, monkeypatch, tempfd):
-
-    monkeypatch.setenv("PYTHONSTARTUP", tempfd.name)
-    tempfd.write(startupfile.encode())
-    tempfd.flush()
+def test_apython_pythonstartup(capfd, use_readline, monkeypatch, tmpdir):
+    python_startup = tmpdir / "python_startup.py"
+    monkeypatch.setenv("PYTHONSTARTUP", str(python_startup))
+    python_startup.write(startupfile)
 
     test_vectors = (
         ("print(foo)\n", "", ">>> 1\n"),

tests/test_server.py

@@ -33,6 +33,13 @@ async def test_server(event_loop):
 @pytest.mark.asyncio
 async def test_uds_server(event_loop, tmpdir):
     path = str(tmpdir / "test.uds")
+
+    # Not availble on windows
+    if compat.platform == "win32":
+        with pytest.raises(ValueError):
+            await start_console_server(path=path, banner="test")
+        return
+
     server = await start_console_server(path=path, banner="test")
 
     stream = io.StringIO()

tests/test_stream.py

@@ -116,7 +116,7 @@ async def test_read_from_closed_pipe():
     stdin.close()
 
     reader, writer1, writer2 = await create_standard_streams(
-        open(stdin_r, "rb"), open(stdout_w, "wb"), open(stderr_w, "rb")
+        open(stdin_r, "r"), open(stdout_w, "w"), open(stderr_w, "r")
     )
 
     result = await ainput(">>> ", streams=(reader, writer1))