feat: LSP-Diagnostics & Completion + Tests

Lukas Geiger · Lukas Geiger · commit 114af04df001 · 2026-05-01T18:30:00.000+02:00
- features/lsp_client.py: Diagnostics thread-sicher ueber Qt-Signale,
  Completion-Anfragen beim Tippen
- tests/: test_lsp_client.py, test_lsp_runtime.py
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -7,6 +7,8 @@ Format basiert auf [Keep a Changelog](https://keepachangelog.com/de/1.1.0/).
 
 ### Hinzugefügt
 - Headless-Smoke-Test für MainWindow-Instanziierung
+- Optionale LSP-Runtime-Tests für `python-lsp-server[all]`:
+  Diagnostics bei Syntaxfehlern und Completion über `pylsp`.
 - `__all__`-Exports in allen Modul-`__init__.py`
 - LSP-Diagnostics und LSP-Completion sind jetzt im Editor verdrahtet:
   Diagnostics laufen thread-sicher über Qt-Signale, Completion-Anfragen werden
@@ -17,6 +19,10 @@ Format basiert auf [Keep a Changelog](https://keepachangelog.com/de/1.1.0/).
 - Diverse ungenutzte Imports entfernt (core, features, languages, ui)
 - Fenstertitel liest die Version jetzt aus `version.py` statt aus einem Hardcode
 - Theme-Wechsel setzt Palette und QSS gemeinsam; Light-Mode bleibt nicht mehr auf Dark-Basis hängen
+- Python-LSP-Erkennung startet `pylsp` jetzt auch über `python -m pylsp`,
+  wenn das Script nicht auf `PATH` liegt, das Modul aber installiert ist.
+- LSP-Subprocess-Pipes werden beim Stoppen geschlossen; der Runtime-Test läuft
+  dadurch ohne ResourceWarnings.
 
 ### Geändert
 - Deutschsprachige Doku sowie Python-Kommentare, Docstrings und naheliegende UI-Texte
diff --git a/README.md b/README.md
@@ -36,12 +36,17 @@ Alternativ per Doppelklick auf `start.bat`.
 
 ### Optionale LSP-Server
 
-- Python: `pip install python-lsp-server`
+- Python: `pip install "python-lsp-server[all]"` für Completion plus Diagnostics
+  (`pip install python-lsp-server` reicht nur für Completion)
 - TypeScript: `npm install -g typescript-language-server`
 - Rust: `rustup component add rust-analyzer`
 - Go: `go install golang.org/x/tools/gopls@latest`
 - C++: `clangd` bzw. LLVM
 
+Der Python-LSP wird bevorzugt über `pylsp` auf `PATH` gestartet. Falls das
+Script nach der Installation nicht auf `PATH` liegt, nutzt CodeBox den aktuellen
+Python-Interpreter als Fallback: `python -m pylsp`.
+
 ## Lokaler Windows-Build
 
 ```bat
diff --git a/features/lsp_client.py b/features/lsp_client.py
@@ -11,8 +11,10 @@
 - textDocument/publishDiagnostics (Fehler/Warnungen)
 """
 
+import importlib.util
 import json
 import subprocess
+import sys
 import threading
 import shutil
 from typing import Optional, Dict, List, Callable
@@ -21,7 +23,7 @@
 
 # Bekannte LSP-Server pro Sprache
 LSP_SERVERS = {
-    "Python": {"cmd": ["pylsp"], "check": "pylsp"},
+    "Python": {"cmd": ["pylsp"], "check": "pylsp", "module": "pylsp"},
     "JavaScript": {"cmd": ["typescript-language-server", "--stdio"], "check": "typescript-language-server"},
     "TypeScript": {"cmd": ["typescript-language-server", "--stdio"], "check": "typescript-language-server"},
     "Rust": {"cmd": ["rust-analyzer"], "check": "rust-analyzer"},
@@ -83,22 +85,37 @@ def __init__(self, language: str, root_path: str = "."):
     def server_config(self) -> Optional[dict]:
         return LSP_SERVERS.get(self.language)
 
-    def is_available(self) -> bool:
-        """Prüft, ob der LSP-Server installiert ist."""
+    def _resolve_command(self) -> Optional[List[str]]:
+        """Ermittelt das startbare Server-Kommando.
+
+        Bei Python ist `pylsp.exe` nach einer pip-Installation nicht immer auf
+        PATH. Wenn das Modul im aktuellen Interpreter vorhanden ist, startet
+        CodeBox den Server deshalb über `python -m pylsp`.
+        """
         config = self.server_config
         if not config:
-            return False
-        return shutil.which(config["check"]) is not None
+            return None
+        check = config.get("check")
+        if check and shutil.which(check):
+            return list(config["cmd"])
+        module = config.get("module")
+        if module and importlib.util.find_spec(module):
+            return [sys.executable, "-m", module]
+        return None
+
+    def is_available(self) -> bool:
+        """Prüft, ob der LSP-Server installiert ist."""
+        return self._resolve_command() is not None
 
     def start(self) -> bool:
         """Startet den LSP-Server-Prozess."""
-        if not self.is_available():
+        command = self._resolve_command()
+        if not command:
             return False
 
-        config = self.server_config
         try:
             self.process = subprocess.Popen(
-                config["cmd"],
+                command,
                 stdin=subprocess.PIPE,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
@@ -131,18 +148,34 @@ def start(self) -> bool:
     def stop(self):
         """Stoppt den LSP-Server."""
         self._running = False
-        if self.process:
+        process = self.process
+        if process:
             try:
                 self._send_request("shutdown", {})
                 self._send_notification("exit", {})
             except (BrokenPipeError, OSError):
                 pass
             try:
-                self.process.terminate()
-                self.process.wait(timeout=3)
+                process.terminate()
+                process.wait(timeout=3)
             except subprocess.TimeoutExpired:
-                self.process.kill()
-            self.process = None
+                process.kill()
+                process.wait(timeout=3)
+            finally:
+                for stream in (process.stdin, process.stdout, process.stderr):
+                    if stream:
+                        try:
+                            stream.close()
+                        except OSError:
+                            pass
+                self.process = None
+        if (
+            self._reader_thread
+            and self._reader_thread.is_alive()
+            and threading.current_thread() is not self._reader_thread
+        ):
+            self._reader_thread.join(timeout=1)
+        self._reader_thread = None
 
     def did_open(self, uri: str, language_id: str, text: str, version: int = 1):
         """Benachrichtigt den Server über eine geöffnete Datei."""
diff --git a/tests/test_lsp_client.py b/tests/test_lsp_client.py
@@ -0,0 +1,43 @@
+import unittest
+from unittest.mock import patch
+
+from features.lsp_client import LSPClient
+
+
+class LSPClientCommandResolutionTests(unittest.TestCase):
+    def test_prefers_server_executable_when_on_path(self):
+        client = LSPClient("Python")
+
+        with patch("features.lsp_client.shutil.which", return_value="C:/bin/pylsp.exe"):
+            self.assertTrue(client.is_available())
+            self.assertEqual(client._resolve_command(), ["pylsp"])
+
+    def test_python_server_falls_back_to_current_interpreter_module(self):
+        client = LSPClient("Python")
+
+        with patch("features.lsp_client.shutil.which", return_value=None), \
+                patch("features.lsp_client.importlib.util.find_spec", return_value=object()), \
+                patch("features.lsp_client.sys.executable", "C:/Python/python.exe"):
+            self.assertTrue(client.is_available())
+            self.assertEqual(
+                client._resolve_command(),
+                ["C:/Python/python.exe", "-m", "pylsp"],
+            )
+
+    def test_unavailable_when_no_executable_or_module_exists(self):
+        client = LSPClient("Python")
+
+        with patch("features.lsp_client.shutil.which", return_value=None), \
+                patch("features.lsp_client.importlib.util.find_spec", return_value=None):
+            self.assertFalse(client.is_available())
+            self.assertIsNone(client._resolve_command())
+
+    def test_unknown_language_is_unavailable(self):
+        client = LSPClient("Unknown")
+
+        self.assertFalse(client.is_available())
+        self.assertIsNone(client._resolve_command())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_lsp_runtime.py b/tests/test_lsp_runtime.py
@@ -0,0 +1,70 @@
+import os
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+
+from features.lsp_client import LSPClient
+
+
+@unittest.skipUnless(
+    os.environ.get("CODEBOX_LSP_RUNTIME") == "1",
+    "Set CODEBOX_LSP_RUNTIME=1 and install python-lsp-server[all] to run.",
+)
+class PythonLSPRuntimeTests(unittest.TestCase):
+    def test_pylsp_diagnostics_and_completion_roundtrip(self):
+        client = LSPClient("Python", tempfile.mkdtemp(prefix="codebox-lsp-"))
+        if not client.is_available():
+            self.skipTest("pylsp is not available on PATH or as python -m pylsp")
+
+        root = Path(client.root_path)
+        path = root / "sample.py"
+        uri = path.resolve().as_uri()
+        diagnostics = []
+        completions = []
+        diag_event = threading.Event()
+        completion_event = threading.Event()
+
+        def on_diagnostics(params):
+            if params.get("uri") != uri:
+                return
+            batch = params.get("diagnostics", [])
+            diagnostics[:] = batch
+            if batch:
+                diag_event.set()
+
+        def on_completion(result):
+            items = result.get("items", []) if isinstance(result, dict) else (result or [])
+            completions[:] = [
+                item.get("label") if isinstance(item, dict) else str(item)
+                for item in items
+            ]
+            completion_event.set()
+
+        try:
+            self.assertTrue(client.start())
+            client.on_diagnostics = on_diagnostics
+
+            broken_text = "def broken(:\n    pass\n"
+            path.write_text(broken_text, encoding="utf-8")
+            client.did_open(uri, "python", broken_text, version=1)
+            self.assertTrue(
+                diag_event.wait(15),
+                "No diagnostics received; install python-lsp-server[all] with lint plugins.",
+            )
+            self.assertTrue(
+                any("invalid syntax" in item.get("message", "") for item in diagnostics)
+            )
+
+            completion_text = "import os\nos.pa"
+            path.write_text(completion_text, encoding="utf-8")
+            client.did_change(uri, completion_text, version=2)
+            client.request_completion(uri, 1, 5, callback=on_completion)
+            self.assertTrue(completion_event.wait(10), "No completion response received.")
+            self.assertIn("path", {item for item in completions if item})
+        finally:
+            client.stop()
+
+
+if __name__ == "__main__":
+    unittest.main()
