193 lines
8.7 KiB
Python
193 lines
8.7 KiB
Python
"""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()
|