1 files changed, 135 insertions, 0 deletions
diff --git a/tests/test_server_connection.py b/tests/test_server_connection.py
new file mode 100644
index 0000000..1fda258
--- /dev/null
+++ b/tests/test_server_connection.py
@@ -0,0 +1,135 @@
+import asyncio
+import json
+import os
+from threading import Thread
+from unittest.mock import Mock
+
+import pytest
+
+from pygls import IS_PYODIDE
+from pygls.server import LanguageServer
+
+try:
+    import websockets
+
+    WEBSOCKETS_AVAILABLE = True
+except ImportError:
+    WEBSOCKETS_AVAILABLE = False
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(IS_PYODIDE, reason="threads are not available in pyodide.")
+async def test_tcp_connection_lost():
+    loop = asyncio.new_event_loop()
+
+    server = LanguageServer("pygls-test", "v1", loop=loop)
+
+    server.lsp.connection_made = Mock()
+    server.lsp.connection_lost = Mock()
+
+    # Run the server over TCP in a separate thread
+    server_thread = Thread(
+        target=server.start_tcp,
+        args=(
+            "127.0.0.1",
+            0,
+        ),
+    )
+    server_thread.daemon = True
+    server_thread.start()
+
+    # Wait for server to be ready
+    while server._server is None:
+        await asyncio.sleep(0.5)
+
+    # Simulate client's connection
+    port = server._server.sockets[0].getsockname()[1]
+    reader, writer = await asyncio.open_connection("127.0.0.1", port)
+    await asyncio.sleep(1)
+
+    assert server.lsp.connection_made.called
+
+    # Socket is closed (client's process is terminated)
+    writer.close()
+    await asyncio.sleep(1)
+
+    assert server.lsp.connection_lost.called
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(IS_PYODIDE, reason="threads are not available in pyodide.")
+async def test_io_connection_lost():
+    # Client to Server pipe.
+    csr, csw = os.pipe()
+    # Server to client pipe.
+    scr, scw = os.pipe()
+
+    server = LanguageServer("pygls-test", "v1", loop=asyncio.new_event_loop())
+    server.lsp.connection_made = Mock()
+    server_thread = Thread(
+        target=server.start_io, args=(os.fdopen(csr, "rb"), os.fdopen(scw, "wb"))
+    )
+    server_thread.daemon = True
+    server_thread.start()
+
+    # Wait for server to be ready
+    while not server.lsp.connection_made.called:
+        await asyncio.sleep(0.5)
+
+    # Pipe is closed (client's process is terminated)
+    os.close(csw)
+    server_thread.join()
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(
+    IS_PYODIDE or not WEBSOCKETS_AVAILABLE,
+    reason="threads are not available in pyodide",
+)
+async def test_ws_server():
+    """Smoke test to ensure we can send/receive messages over websockets"""
+
+    loop = asyncio.new_event_loop()
+    server = LanguageServer("pygls-test", "v1", loop=loop)
+
+    # Run the server over Websockets in a separate thread
+    server_thread = Thread(
+        target=server.start_ws,
+        args=(
+            "127.0.0.1",
+            0,
+        ),
+    )
+    server_thread.daemon = True
+    server_thread.start()
+
+    # Wait for server to be ready
+    while server._server is None:
+        await asyncio.sleep(0.5)
+
+    port = server._server.sockets[0].getsockname()[1]
+    # Simulate client's connection
+    async with websockets.connect(f"ws://127.0.0.1:{port}") as connection:
+        # Send an 'initialize' request
+        msg = dict(
+            jsonrpc="2.0", id=1, method="initialize", params=dict(capabilities=dict())
+        )
+        await connection.send(json.dumps(msg))
+
+        response = await connection.recv()
+        assert "result" in response
+
+        # Shut the server down
+        msg = dict(
+            jsonrpc="2.0", id=2, method="shutdown", params=dict(capabilities=dict())
+        )
+        await connection.send(json.dumps(msg))
+
+        response = await connection.recv()
+        assert "result" in response
+
+        # Finally, tell it to exit
+        msg = dict(jsonrpc="2.0", id=2, method="exit", params=None)
+        await connection.send(json.dumps(msg))
+
+    server_thread.join()