fix: Erkennung von bereits laufendem Server und dessen beendung. incl. Tests

0_application/tests/test_server_lifecycle.py

@@ -0,0 +1,162 @@
+"""Exercise the actual PowerShell logic with simulated Windows processes."""
+
+import os
+from pathlib import Path
+import subprocess
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import server_lifecycle
+
+
+@unittest.skipUnless(os.name == "nt", "Windows process management")
+class ServerLifecycleTests(unittest.TestCase):
+    def simulate(self, process, listening=True, stop_error=False, confirm=True):
+        captured = {}
+
+        def capture(command, **kwargs):
+            captured.update(script=command[-1], env=kwargs["env"])
+            return subprocess.CompletedProcess(command, 0, "[]", "")
+
+        with patch.object(server_lifecycle.subprocess, "run", side_effect=capture):
+            server_lifecycle.stop_existing_server()
+
+        setup = r'''
+$script:occupied = LISTENING
+function Get-NetTCPConnection {
+    param($State, $ErrorAction)
+    if ($script:occupied) { [pscustomobject]@{ LocalPort=8501; OwningProcess=12345 } }
+}
+function Get-CimInstance {
+    param($ClassName, $Filter)
+    PROCESS
+}
+function Stop-Process {
+    param($Id, [switch]$Force, $ErrorAction)
+    [Console]::Error.WriteLine('STOP_CALLED')
+    STOP
+    $script:occupied = $false
+}
+'''.replace("LISTENING", "$true" if listening else "$false")
+        setup = setup.replace("PROCESS", process).replace(
+            "\n    STOP\n", "\n    throw 'simulated access denied'\n" if stop_error else "\n")
+        confirm_mock = """
+function Confirm-ServerRestart {
+    [Console]::Error.WriteLine('CONFIRM_CALLED')
+    return ANSWER
+}
+""".replace("ANSWER", "$true" if confirm else "$false")
+        script = captured["script"].replace(server_lifecycle._CONFIRM_SCRIPT, confirm_mock)
+        command = [
+            "powershell.exe", "-NoProfile", "-NonInteractive", "-Command",
+            setup + script,
+        ]
+        # Explicit pipes keep both outputs available independently of pytest's
+        # own console capture. communicate() drains both pipes concurrently.
+        with subprocess.Popen(
+            command, env=captured["env"], stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE, text=True, errors="replace",
+        ) as child:
+            try:
+                stdout, stderr = child.communicate(timeout=30)
+            except subprocess.TimeoutExpired:
+                child.kill()
+                child.communicate()
+                raise
+            self.assertIsInstance(stdout, str, "stdout wurde nicht erfasst")
+            self.assertIsInstance(stderr, str, "stderr wurde nicht erfasst")
+            return subprocess.CompletedProcess(command, child.returncode, stdout, stderr)
+
+    def test_free_port(self):
+        result = self.simulate("throw 'should not inspect a process'", listening=False)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.strip(), "[]")
+        self.assertNotIn("CONFIRM_CALLED", result.stderr)
+
+    def test_previous_exe_is_stopped(self):
+        result = self.simulate(
+            "[pscustomobject]@{ Name='Tätigkeitsnachweis-Generator.exe'; CommandLine=''; ExecutablePath='' }")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stdout.strip(), "[12345]")
+        self.assertIn("CONFIRM_CALLED", result.stderr)
+        self.assertIn("STOP_CALLED", result.stderr)
+
+    def test_declining_preserves_previous_instance(self):
+        result = self.simulate(
+            "[pscustomobject]@{ Name='Tätigkeitsnachweis-Generator.exe'; CommandLine=''; ExecutablePath='' }",
+            confirm=False)
+        self.assertEqual(result.returncode, 2, result.stderr)
+        self.assertIn("CONFIRM_CALLED", result.stderr)
+        self.assertNotIn("STOP_CALLED", result.stderr)
+
+    def test_declining_returns_false_to_launcher(self):
+        with patch.object(server_lifecycle.subprocess, "run", return_value=
+                          subprocess.CompletedProcess([], 2, "", "")):
+            self.assertFalse(server_lifecycle.stop_existing_server())
+
+    def test_development_absolute_path_is_stopped(self):
+        result = self.simulate(r'''[pscustomobject]@{
+            Name='python.exe'; ExecutablePath='C:\Python\python.exe';
+            CommandLine=('python -m streamlit run "' + $env:APP_RESTART_PATH + '"')
+        }''')
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_legacy_development_start_is_stopped(self):
+        result = self.simulate(r'''[pscustomobject]@{
+            Name='python.exe'; CommandLine='streamlit.exe run app.py';
+            ExecutablePath=($env:APP_RESTART_VENV + '\python.exe')
+        }''')
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_versioned_python_prompts_before_stopping(self):
+        # Windows Store Python reports the real interpreter name even when
+        # launched through the venv's python.exe.
+        process = r'''[pscustomobject]@{
+            Name='python3.13.exe';
+            ExecutablePath='C:\Program Files\WindowsApps\Python\python3.13.exe';
+            CommandLine=('".venv\Scripts\python.exe" -m streamlit run "' +
+                $env:APP_RESTART_PATH + '" --server.port=8501 --server.address=localhost')
+        }'''
+        with self.subTest(answer="yes"):
+            result = self.simulate(process, confirm=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertIn("CONFIRM_CALLED", result.stderr)
+            self.assertIn("STOP_CALLED", result.stderr)
+        with self.subTest(answer="no"):
+            result = self.simulate(process, confirm=False)
+            self.assertEqual(result.returncode, 2, result.stderr)
+            self.assertIn("CONFIRM_CALLED", result.stderr)
+            self.assertNotIn("STOP_CALLED", result.stderr)
+
+    def test_unrelated_versioned_python_is_preserved(self):
+        result = self.simulate(r'''[pscustomobject]@{
+            Name='python3.13.exe'; ExecutablePath='C:\Python\python3.13.exe';
+            CommandLine='python -m streamlit run "C:\other-project\app.py"'
+        }''')
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("PID 12345", result.stderr)
+        self.assertNotIn("CONFIRM_CALLED", result.stderr)
+        self.assertNotIn("STOP_CALLED", result.stderr)
+
+    def test_unrelated_server_is_preserved(self):
+        result = self.simulate(r'''[pscustomobject]@{
+            Name='python.exe'; CommandLine='streamlit run app.py';
+            ExecutablePath='C:\other-project\.venv\Scripts\python.exe'
+        }''')
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("PID 12345", result.stderr)
+        self.assertNotIn("STOP_CALLED", result.stderr)
+        self.assertNotIn("CONFIRM_CALLED", result.stderr)
+
+    def test_stop_failure_aborts_start(self):
+        result = self.simulate(
+            "[pscustomobject]@{ Name='Tätigkeitsnachweis-Generator.exe'; CommandLine=''; ExecutablePath='' }",
+            stop_error=True)
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("simulated access denied", result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()