Add some server tests

tests/test_apython.py

@@ -1,5 +1,6 @@
 import io
 import sys
+import asyncio
 from contextlib import contextmanager
 
 from unittest.mock import Mock, patch, call
@@ -8,6 +9,7 @@
 
 from aioconsole import compat
 from aioconsole import apython, rlwrap
+from aioconsole import InteractiveEventLoop
 
 
 @contextmanager
@@ -122,3 +124,16 @@ def test_apython_with_stdout_logs(capsys, use_readline):
     out, err = capsys.readouterr()
     assert out == 'logging'
     assert err == 'test\n>>> 7\n>>> \n'
+
+
+def test_apython_server(capsys, event_loop, monkeypatch):
+    def run_forever(self, orig=InteractiveEventLoop.run_forever):
+        if self.console_server is not None:
+            self.call_later(0, self.stop)
+        return orig(self)
+    with patch('aioconsole.InteractiveEventLoop.run_forever', run_forever):
+        with pytest.raises(SystemExit):
+            apython.run_apython(['--serve=:0'])
+    out, err = capsys.readouterr()
+    assert out.startswith('The console is being served on')
+    assert err == ''

tests/test_command.py

@@ -97,10 +97,10 @@ def say_hello(reader, writer, name=None):
     list(testdata.values()),
     ids=list(testdata.keys()))
 @pytest.mark.asyncio
-def test_async_cli(event_loop, input_string, expected):
-    sys.ps1 = "[Hello!] "
-    sys.stdin = io.StringIO(input_string)
-    sys.stderr = io.StringIO()
+def test_async_cli(event_loop, monkeypatch, input_string, expected):
+    monkeypatch.setattr('sys.ps1', "[Hello!] ", raising=False)
+    monkeypatch.setattr('sys.stdin', io.StringIO(input_string))
+    monkeypatch.setattr('sys.stderr', io.StringIO())
     yield from make_cli().interact(stop=False)
     print(sys.stderr.getvalue())
     assert sys.stderr.getvalue() == expected

tests/test_server.py

@@ -0,0 +1,18 @@
+import asyncio
+
+import pytest
+
+from aioconsole.server import start_console_server
+
+
+@pytest.mark.asyncio
+@asyncio.coroutine
+def test_server(event_loop):
+    server = yield from start_console_server(port=0, banner='test')
+    address = server.sockets[0].getsockname()
+    reader, writer = yield from asyncio.open_connection(*address)
+    assert (yield from reader.readline()) == b'test\n'
+    writer.write(b'1+1\n')
+    assert (yield from reader.readline()) == b'>>> 2\n'
+    writer.close()
+    assert (yield from reader.readline()) == b'>>> '