pytest-lsp: Cancel pendining notification futures on server exit

lib/pytest-lsp/changes/186.fix.md

@@ -0,0 +1 @@
+`pytest-lsp` is now able to detect the situation where the server process exits while the client is waiting on a notification message and fail the test accordingly

lib/pytest-lsp/pytest_lsp/client.py

@@ -99,6 +99,14 @@ async def server_exit(self, server: asyncio.subprocess.Process):
         if self._stop_event.is_set():
             return
 
+        reason = (
+            f"Server process {server.pid} exited with return code: {server.returncode}"
+        )
+        for id_, fut in self.protocol._notification_futures.items():
+            if not fut.done():
+                fut.set_exception(RuntimeError(reason))
+                logger.debug("Cancelled pending request '%s': %s", id_, reason)
+
     def report_server_error(
         self, error: Exception, source: PyglsError | JsonRpcException
     ):

lib/pytest-lsp/tests/servers/notify_exit.py

@@ -0,0 +1,30 @@
+# A server that exits mid request.
+import sys
+
+from lsprotocol import types
+from pygls.lsp.server import LanguageServer
+
+
+class CountingLanguageServer(LanguageServer):
+    count: int = 0
+
+
+server = CountingLanguageServer(name="completion-exit-server", version="v1.0")
+
+
+@server.feature("server/exit")
+def server_exit(*args):
+    sys.exit(0)
+
+
+@server.feature(types.TEXT_DOCUMENT_COMPLETION)
+def on_complete(server: CountingLanguageServer, params: types.CompletionParams):
+    server.count += 1
+    if server.count == 5:
+        sys.exit(0)
+
+    return [types.CompletionItem(label=f"{server.count}")]
+
+
+if __name__ == "__main__":
+    server.start_io()

lib/pytest-lsp/tests/test_plugin.py

@@ -167,6 +167,43 @@ async def test_capabilities(client):
     results.stdout.fnmatch_lines("E*RuntimeError: Client has been stopped.")
 
 
+def test_detect_server_exit_pending_notification(pytester: pytest.Pytester):
+    """Ensure that the plugin can detect when the server process exits while the client
+    is waiting for a notification to arrive."""
+
+    test_code = """\
+import pytest
+from lsprotocol.types import CompletionParams
+from lsprotocol.types import Position
+from lsprotocol.types import TextDocumentIdentifier
+
+
+@pytest.mark.asyncio
+async def test_capabilities(client):
+    expected = {str(i) for i in range(10)}
+
+    for i in range(10):
+        client.protocol.notify("server/exit")
+        await client.wait_for_notification("never/happening")
+
+        params = CompletionParams(
+            text_document=TextDocumentIdentifier(uri="file:///test.txt"),
+            position=Position(line=0, character=0)
+        )
+        items = await client.text_document_completion_async(params)
+        assert len({i.label for i in items} & expected) == len(items)
+"""
+
+    setup_test(pytester, "notify_exit.py", test_code)
+    results = pytester.runpytest("-vv")
+
+    results.assert_outcomes(failed=1, errors=1)
+
+    message = r"E\s+RuntimeError: Server process \d+ exited with return code: 0"
+    results.stdout.re_match_lines(message)
+    results.stdout.fnmatch_lines("E*RuntimeError: Client has been stopped.")
+
+
 def test_detect_server_crash(pytester: pytest.Pytester):
     """Ensure the plugin can detect when the server process crashes on boot."""