test: run end-to-end tests over WebSockets

pyproject.toml

@@ -68,6 +68,7 @@ poetry_lock_check = "poetry check"
 sequence = [
      { cmd = "pytest --cov" },
      { cmd = "pytest tests/e2e --lsp-transport tcp" },
+     { cmd = "pytest tests/e2e --lsp-transport websockets" },
 ]
 ignore_fail = "return_non_zero"
 

tests/conftest.py

@@ -133,7 +133,7 @@ def pytest_addoption(parser):
         dest="lsp_transport",
         action="store",
         default="stdio",
-        choices=("stdio", "tcp"),
+        choices=("stdio", "tcp", "websockets"),
         help="Choose the transport to use with servers under test.",
     )
 
@@ -212,6 +212,15 @@ async def fn(
             await asyncio.sleep(1)
             await client.start_tcp(host, port)
 
+        elif transport == "websockets":
+            # TODO: Make host/port configurable?
+            host, port = "localhost", 8888
+            server_cmd.extend(["--ws", "--host", host, "--port", f"{port}"])
+
+            server = await asyncio.create_subprocess_exec(*server_cmd)
+            await asyncio.sleep(1)
+            await client.start_ws(host, port)
+
         else:
             raise NotImplementedError(f"Unsupported transport: {transport!r}")
 

tests/e2e/test_threaded_handlers.py

@@ -130,7 +130,7 @@ async def test_countdown_threaded(
 ):
     """Ensure that the countdown threaded command is working as expected."""
 
-    if IS_WIN and transport == "tcp":
+    if (IS_WIN and transport == "tcp") or transport == "websockets":
         pytest.skip("see https://github.com/openlawlibrary/pygls/issues/502")
 
     client, initialize_result = threaded_handlers