"""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, frozen=False): captured = {} def capture(command, **kwargs): captured.update(script=command[-1], env=kwargs["env"]) return subprocess.CompletedProcess(command, 0, "[]", "") root = Path(__file__).resolve().parents[2] with patch.object(server_lifecycle.subprocess, "run", side_effect=capture), \ patch.object(server_lifecycle.sys, "frozen", frozen, create=True), \ patch.object(server_lifecycle.sys, "executable", str(root / "2_build" / "Tätigkeitsnachweis-Generator.exe")), \ patch.object(server_lifecycle, "__file__", str(root / "_MEI-test" / "server_lifecycle.py") if frozen else server_lifecycle.__file__): 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_exe_recognizes_sibling_development_server(self): process = r'''[pscustomobject]@{ Name='python3.13.exe'; ExecutablePath='C:\Python\python3.13.exe'; CommandLine=('".venv\Scripts\python.exe" -m streamlit run "' + (($env:APP_RESTART_PATHS | ConvertFrom-Json)[1]) + '" --server.port=8501 --server.address=localhost') }''' for answer, expected_code in [(True, 0), (False, 2)]: with self.subTest(confirm=answer): result = self.simulate(process, frozen=True, confirm=answer) self.assertEqual(result.returncode, expected_code, result.stderr) self.assertIn("CONFIRM_CALLED", result.stderr) if answer: self.assertIn("STOP_CALLED", result.stderr) else: self.assertNotIn("STOP_CALLED", result.stderr) def test_exe_preserves_unrelated_development_server(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"' }''', frozen=True) self.assertNotEqual(result.returncode, 0) 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()