3 Commits
11 changed files with 426 additions and 169 deletions

No files matched your search

+26 -7
View File
@@ -87,8 +87,26 @@ Mapping von Input-Feldern auf Template-Zellen:
## Tests ## Tests
End-to-End-Tests parsen die reale Input-Datei (`input/2025 JM.xlsx`) und das reale Beim Start über `start.bat` oder die neu gebaute EXE erscheint bei einer bereits
Template (`template/Tätigkeitsnachweise Stellenförderung.xlsx`) und validieren laufenden Programminstanz ein Ja/Nein-Dialog. „Ja“ beendet die bisherige Instanz
und startet die neue; „Nein“ lässt die bisherige Instanz geöffnet und bricht den
neuen Start ab. Aktuelle Auswahlen gehen beim Beenden verloren. Ein fremdes
Programm auf Port 8501 wird nicht beendet.
Unter Windows führt `test.bat` alle Tests im Ordner `tests/` aus. Das Skript
verwendet die Projektumgebung `.venv` und installiert bei Bedarf `pytest`.
Per Doppelklick bleibt das Ergebnisfenster geöffnet. Für automatisierte Aufrufe:
```powershell
.\0_application\test.bat --no-pause
```
Dieser Aufruf erfolgt aus dem Projekt-Hauptordner. Bei fehlgeschlagenen Tests
liefert das Skript einen Fehlercode und führt die übrigen Tests trotzdem aus.
Die Tests laufen nicht automatisch beim Programmstart oder Build.
End-to-End-Tests parsen die feste Testdatei (`tests/fixture/2026 JM.xlsx`) und das
PDF-Template (`template/Tätigkeitsnachweise Stellenförderung.pdf`) und validieren
den gesamten Generierungs-Workflow. den gesamten Generierungs-Workflow.
```bash ```bash
@@ -96,12 +114,13 @@ den gesamten Generierungs-Workflow.
.venv/bin/pip install pytest .venv/bin/pip install pytest
# Tests ausführen # Tests ausführen
.venv/bin/python -m pytest tests/test_e2e.py -v .venv/bin/python -m pytest tests -v
``` ```
Die Tests prüfen: Die Tests prüfen:
- Vorhandensein der Input- und Template-Dateien - Vorhandensein der Input- und Template-Dateien
- Korrekte Parsierung (168 Datensätze, 19 Personen) - Korrekte Parsierung (100 Datensätze, 17 Personen)
- Vollständige Pipeline für die erste Person („Draßdo, Cornelia"): 12 Monats-Sheets - Vollständige PDF-Pipeline für die erste Person („Bardaus, Madalina“): fünf Monate
mit allen Zellinhalten (Antragsnummer, Name, Stellenanteil, Qualifikation, Monat, von Januar bis Mai 2026 mit zehn Seiten, Antragsnummer, Name, Stellenanteil,
Funktion, Tätigkeit, Ort, Datum) Monat, Funktion, Tätigkeit, Ort und korrekten Monatsenddaten
- Neustartlogik unter Windows mit simulierten Prozessen
+2
View File
@@ -1,4 +1,6 @@
@echo off @echo off
chcp 65001 >nul
set "PYTHONUTF8=1"
title Tätigkeitsnachweis-Generator – Cold Run title Tätigkeitsnachweis-Generator – Cold Run
echo ======================================== echo ========================================
echo Cold Run – Headless-Generierung echo Cold Run – Headless-Generierung
+103
View File
@@ -0,0 +1,103 @@
"""Stop an existing instance of this application's Windows server."""
import json
import os
import subprocess
from pathlib import Path
_CONFIRM_SCRIPT = r'''
function Confirm-ServerRestart {
Add-Type -AssemblyName System.Windows.Forms
$answer = [System.Windows.Forms.MessageBox]::Show(
"Es läuft bereits eine Version des Tätigkeitsnachweis-Generators.`n`nSoll das alte Programm beendet und die neue Version gestartet werden?`nDie aktuellen Auswahlen gehen dabei verloren.",
'Tätigkeitsnachweis-Generator',
[System.Windows.Forms.MessageBoxButtons]::YesNo,
[System.Windows.Forms.MessageBoxIcon]::Question,
[System.Windows.Forms.MessageBoxDefaultButton]::Button2
)
return $answer -eq [System.Windows.Forms.DialogResult]::Yes
}
'''
def stop_existing_server(port: int = 8501) -> bool:
"""Ask before stopping this app. Return False if the user declines."""
if os.name != "nt":
raise RuntimeError("Der kontrollierte Server-Neustart unterstützt nur Windows.")
app_path = str(Path(__file__).resolve().with_name("app.py"))
# Pass paths through the environment, never interpolate them into shell code.
env = dict(os.environ, APP_RESTART_PATH=app_path, APP_RESTART_PID=str(os.getpid()),
APP_RESTART_PORT=str(port),
APP_RESTART_VENV=str(Path(app_path).parent / ".venv" / "Scripts"))
script = _CONFIRM_SCRIPT + r'''
[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new()
$ErrorActionPreference = 'Stop'
try {
$port = [int]$env:APP_RESTART_PORT
$listeners = @(Get-NetTCPConnection -State Listen -ErrorAction Stop |
Where-Object { $_.LocalPort -eq $port })
$processIds = @($listeners | Select-Object -ExpandProperty OwningProcess -Unique)
$targets = @()
foreach ($processId in $processIds) {
if ($processId -eq [int]$env:APP_RESTART_PID) {
throw "Der aktuelle Prozess belegt bereits Port $port."
}
$process = Get-CimInstance Win32_Process -Filter "ProcessId = $processId"
if (-not $process) { continue }
$isExe = $process.Name -eq 'Tätigkeitsnachweis-Generator.exe'
$command = $process.CommandLine
$appArgument = '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+"?' +
[regex]::Escape($env:APP_RESTART_PATH) + '"?(?:\s|$)'
$isDev = $process.Name -match '^(python(?:w)?(?:\d+(?:\.\d+)*)?(?:w)?|streamlit)\.exe$' -and
$command -match '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+' -and
($command -match $appArgument -or
($command -match '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+"?app\.py"?(?:\s|$)' -and
[IO.Path]::GetDirectoryName($process.ExecutablePath) -eq $env:APP_RESTART_VENV))
if (-not ($isExe -or $isDev)) {
throw "Port $port wird von einem anderen Programm verwendet (PID $processId, $($process.Name)). Bitte manuell prüfen."
}
$targets += $processId
}
if ($targets.Count -gt 0 -and -not (Confirm-ServerRestart)) {
exit 2
}
foreach ($processId in $targets) {
Stop-Process -Id $processId -Force -ErrorAction Stop
}
$deadline = (Get-Date).AddSeconds(10)
do {
$remaining = @(Get-NetTCPConnection -State Listen -ErrorAction Stop |
Where-Object { $_.LocalPort -eq $port })
if ($remaining.Count -eq 0) { break }
Start-Sleep -Milliseconds 200
} while ((Get-Date) -lt $deadline)
if ($remaining.Count -gt 0) { throw "Port $port wurde nicht rechtzeitig freigegeben." }
ConvertTo-Json -Compress -InputObject @($targets)
} catch {
[Console]::Error.WriteLine($_.Exception.Message)
exit 1
}
'''
result = subprocess.run(
["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", script],
env=env, capture_output=True, text=True, encoding="utf-8", errors="replace",
)
if result.returncode == 2:
return False
if result.returncode:
raise RuntimeError((result.stderr or "").strip() or "Der bestehende Server konnte nicht beendet werden.")
stopped = json.loads(result.stdout.strip() or "[]")
if stopped:
print(f"[INFO] Vorherige Instanz beendet (PID: {stopped}).", flush=True)
return True
if __name__ == "__main__":
try:
if not stop_existing_server():
print("[INFO] Die bisherige Version bleibt geöffnet. Keine neue Instanz gestartet.")
raise SystemExit(2)
except (RuntimeError, OSError, subprocess.SubprocessError) as exc:
print(f"[FEHLER] {exc}")
raise SystemExit(1)
+10 -1
View File
@@ -1,4 +1,7 @@
@echo off @echo off
chcp 65001 >nul
set "PYTHONUTF8=1"
cd /d "%~dp0"
title Tätigkeitsnachweis-Generator title Tätigkeitsnachweis-Generator
echo ======================================== echo ========================================
echo Tätigkeitsnachweis-Generator echo Tätigkeitsnachweis-Generator
@@ -43,6 +46,12 @@ echo.
echo [INFO] Starte Tätigkeitsnachweis-Generator... echo [INFO] Starte Tätigkeitsnachweis-Generator...
echo [INFO] Die App öffnet sich im Browser unter http://localhost:8501 echo [INFO] Die App öffnet sich im Browser unter http://localhost:8501
echo. echo.
streamlit run app.py ".venv\Scripts\python.exe" server_lifecycle.py
if %ERRORLEVEL% equ 2 exit /b 0
if %ERRORLEVEL% neq 0 (
pause
exit /b 1
)
".venv\Scripts\python.exe" -m streamlit run "%~dp0app.py" --server.port=8501 --server.address=localhost
pause pause
+45
View File
@@ -0,0 +1,45 @@
@echo off
chcp 65001 >nul
setlocal
set "PYTHONUTF8=1"
cd /d "%~dp0"
title Taetigkeitsnachweis-Generator - Tests
echo ========================================
echo Alle Tests ausfuehren
echo ========================================
echo.
if not exist ".venv\Scripts\python.exe" (
echo [FEHLER] Die virtuelle Umgebung fehlt. Bitte zuerst start.bat ausfuehren.
set "TEST_EXIT_CODE=1"
goto finish
)
".venv\Scripts\python.exe" -c "import pytest" >nul 2>nul
if errorlevel 1 (
echo [INFO] Installiere pytest in der Projektumgebung...
".venv\Scripts\python.exe" -m pip install pytest
if errorlevel 1 (
echo [FEHLER] pytest konnte nicht installiert werden.
set "TEST_EXIT_CODE=1"
goto finish
)
)
REM Keine Abbruchoption: Auch nach fehlgeschlagenen Tests laeuft die Suite weiter.
REM Ohne Cache funktioniert der Testlauf auch bei gesperrtem pytest-Cache.
".venv\Scripts\python.exe" -m pytest tests -v -p no:cacheprovider
set "TEST_EXIT_CODE=%ERRORLEVEL%"
echo.
if "%TEST_EXIT_CODE%"=="0" (
echo [OK] Alle Tests bestanden.
) else (
echo [FEHLER] Testlauf nicht erfolgreich. Details stehen oben.
)
:finish
REM Fuer automatisierte Aufrufe: test.bat --no-pause
if /I not "%~1"=="--no-pause" pause
exit /b %TEST_EXIT_CODE%
Binary file not shown.
+62 -161
View File
@@ -1,12 +1,11 @@
"""End-to-End-Test: Reale Input-Datei + reales PDF-Template → PDF-Output für erste Person.""" """End-to-end checks using the fixed 2026 workbook and the PDF template."""
import sys import sys
from pathlib import Path from pathlib import Path
import fitz # PyMuPDF import fitz
import pytest import pytest
# Projekt-Root zum Python-Path hinzufügen
PROJECT_ROOT = Path(__file__).resolve().parent.parent PROJECT_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT)) sys.path.insert(0, str(PROJECT_ROOT))
@@ -14,37 +13,22 @@ from src.config import load_settings, load_input_schema, load_output_mapping
from src.parser import parse_input_file, group_by_employee, get_unique_employees from src.parser import parse_input_file, group_by_employee, get_unique_employees
from src.writer import generate_activity_reports from src.writer import generate_activity_reports
# ──────────────────────────────────────────────
# Pfade zu den realen Fixture-Dateien
# ──────────────────────────────────────────────
CONFIG_DIR = PROJECT_ROOT / "config" CONFIG_DIR = PROJECT_ROOT / "config"
INPUT_FILE = PROJECT_ROOT / "input" / "2025 JM.xlsx" INPUT_FILE = Path(__file__).resolve().parent / "fixture" / "2026 JM.xlsx"
TEMPLATE_FILE = PROJECT_ROOT / "template" / "Tätigkeitsnachweise Stellenförderung.pdf" TEMPLATE_FILE = PROJECT_ROOT / "template" / "Tätigkeitsnachweise Stellenförderung.pdf"
def _load_app_config(): @pytest.fixture(scope="module")
"""Lädt die echte App-Konfiguration (Schema, Mapping, Settings).""" def app_config():
config = load_settings(CONFIG_DIR, PROJECT_ROOT) config = load_settings(CONFIG_DIR, PROJECT_ROOT)
config.input_schema = load_input_schema(CONFIG_DIR) config.input_schema = load_input_schema(CONFIG_DIR)
config.output_mapping = load_output_mapping(CONFIG_DIR) config.output_mapping = load_output_mapping(CONFIG_DIR)
return config return config
# ──────────────────────────────────────────────
# Fixtures
# ──────────────────────────────────────────────
@pytest.fixture(scope="module")
def app_config():
"""Reale App-Konfiguration aus config/."""
return _load_app_config()
@pytest.fixture(scope="module") @pytest.fixture(scope="module")
def parsed_records(app_config): def parsed_records(app_config):
"""Geparste Records aus der realen Input-Datei.""" assert INPUT_FILE.is_file(), f"Testdatei nicht gefunden: {INPUT_FILE}"
return parse_input_file( return parse_input_file(
filepath=INPUT_FILE, filepath=INPUT_FILE,
schema=app_config.input_schema, schema=app_config.input_schema,
@@ -56,165 +40,82 @@ def parsed_records(app_config):
@pytest.fixture @pytest.fixture
def output_dir(tmp_path): def output_dir(tmp_path):
"""Temporäres Output-Verzeichnis."""
out = tmp_path / "output" out = tmp_path / "output"
out.mkdir() out.mkdir()
return out return out
# ──────────────────────────────────────────────
# Hilfsfunktion: Text aus PDF extrahieren
# ──────────────────────────────────────────────
def _get_pdf_text(pdf_path: Path) -> str:
"""Extrahiert den gesamten Text aus einem PDF."""
doc = fitz.open(pdf_path)
text = ""
for page in doc:
text += page.get_text()
doc.close()
return text
def _get_page_text(pdf_path: Path, page_num: int = 0) -> str:
"""Extrahiert Text von einer bestimmten PDF-Seite."""
doc = fitz.open(pdf_path)
text = doc[page_num].get_text() if page_num < doc.page_count else ""
doc.close()
return text
# ──────────────────────────────────────────────
# End-to-End Tests
# ──────────────────────────────────────────────
class TestEndToEndFirstPerson: class TestEndToEndFirstPerson:
"""End-to-End: Reale Input-Datei + reales PDF-Template → nur erste Person."""
def test_input_file_exists(self): def test_input_file_exists(self):
"""Stellt sicher, dass die reale Input-Datei vorhanden ist.""" assert INPUT_FILE.is_file(), f"Testdatei nicht gefunden: {INPUT_FILE}"
assert INPUT_FILE.exists(), f"Input-Datei nicht gefunden: {INPUT_FILE}"
def test_template_file_exists(self): def test_template_file_exists(self):
"""Stellt sicher, dass das reale PDF-Template vorhanden ist.""" assert TEMPLATE_FILE.is_file(), f"PDF-Template nicht gefunden: {TEMPLATE_FILE}"
assert TEMPLATE_FILE.exists(), f"PDF-Template nicht gefunden: {TEMPLATE_FILE}"
def test_template_has_field_markers(self, app_config): def test_template_has_field_markers(self, app_config):
"""Stellt sicher, dass alle _Field-Platzhalter im PDF-Template vorhanden sind.""" with fitz.open(TEMPLATE_FILE) as doc:
doc = fitz.open(TEMPLATE_FILE) missing = [
missing = [] mapping.pdf_field
for m in app_config.output_mapping: for mapping in app_config.output_mapping
if m.pdf_field is None: if mapping.pdf_field is not None
continue and not any(page.search_for(mapping.pdf_field) for page in doc)
# Suche auf allen Seiten ]
found_any = False assert not missing, f"Folgende _Field-Platzhalter fehlen im PDF: {missing}"
for page_num in range(doc.page_count):
if doc[page_num].search_for(m.pdf_field):
found_any = True
break
if not found_any:
missing.append(m.pdf_field)
doc.close()
assert len(missing) == 0, (
f"Folgende _Field-Platzhalter fehlen im PDF: {missing}"
)
def test_parse_real_input(self, parsed_records): def test_parse_real_input(self, parsed_records):
"""Validiert das Parsen der realen Input-Datei.""" # Fixed expectations for tests/fixture/2026 JM.xlsx.
# Sollte 168 Einträge über 12 Monate für 19 Personen ergeben assert len(parsed_records) == 100
assert len(parsed_records) == 168, (
f"Erwartet 168 Records, aber {len(parsed_records)} gefunden"
)
employees = get_unique_employees(parsed_records) employees = get_unique_employees(parsed_records)
assert len(employees) == 19, ( assert len(employees) == 17
f"Erwartet 19 Mitarbeiter, aber {len(employees)} gefunden" assert employees[0] == "Bardaus, Madalina"
) first_records = group_by_employee(parsed_records)[employees[0]]
assert [record.monat for record in first_records] == [
"01.2026", "02.2026", "03.2026", "04.2026", "05.2026"
]
for record in first_records:
assert record.antragsnummer == "100781516"
assert record.funktion == "Psychologin"
assert record.stellenanteil == "50"
assert record.qualifikation == ""
assert record.taetigkeit.startswith("Psychologische Betreuung")
# Erste Person alphabetisch: Draßdo, Cornelia def test_full_pipeline_first_person(self, app_config, parsed_records, output_dir):
assert employees[0] == "Draßdo, Cornelia", (
f"Erwartet 'Draßdo, Cornelia' als erste Person, aber '{employees[0]}'"
)
def test_full_pipeline_first_person(
self, app_config, parsed_records, output_dir
):
"""Kompletter Workflow für die erste gefundene Person (Draßdo, Cornelia)."""
# 1. Nach Mitarbeiter gruppieren
grouped = group_by_employee(parsed_records) grouped = group_by_employee(parsed_records)
assert len(grouped) == 19 assert len(grouped) == 17
first_person = sorted(grouped)[0]
# 2. Erste Person alphabetisch assert first_person == "Bardaus, Madalina"
first_person = sorted(grouped.keys())[0] assert len(grouped[first_person]) == 5
assert first_person == "Draßdo, Cornelia"
first_records = grouped[first_person]
# Draßdo hat 12 Einträge (einen pro Monat)
assert len(first_records) == 12, (
f"Erwartet 12 Records für {first_person}, aber {len(first_records)}"
)
# 3. Reports NUR für die erste Person generieren
single_person_group = {first_person: first_records}
generated = generate_activity_reports( generated = generate_activity_reports(
employee_records=single_person_group, employee_records={first_person: grouped[first_person]},
template_path=TEMPLATE_FILE, template_path=TEMPLATE_FILE,
mapping=app_config.output_mapping, mapping=app_config.output_mapping,
output_dir=output_dir, output_dir=output_dir,
filename_pattern=app_config.filename_pattern, filename_pattern=app_config.filename_pattern,
fontsize=app_config.fontsize,
) )
assert len(generated) == 1
# 4. Ergebnisse prüfen
assert len(generated) == 1, (
f"Erwartet 1 generierte Datei, aber {len(generated)}"
)
output_file = generated[0] output_file = generated[0]
assert output_file.exists(), f"Output-Datei nicht gefunden: {output_file}" assert output_file.is_file()
assert output_file.suffix == ".pdf" assert output_file.name == "Bardaus_Madalina_Tätigkeitsnachweis.pdf"
with fitz.open(output_file) as doc:
erwarteter_name = "Draßdo_Cornelia_Tätigkeitsnachweis.pdf" assert doc.page_count == 10 # Five months, two pages per month.
assert output_file.name == erwarteter_name, ( assert "_Field" not in "".join(page.get_text() for page in doc)
f"Erwartet '{erwarteter_name}', aber '{output_file.name}'" # Independently specified month ends include February's 28 days.
) expected_months = [
("01.2026", "31.01.2026"),
# 5. PDF öffnen und Inhalt validieren ("02.2026", "28.02.2026"),
doc = fitz.open(output_file) ("03.2026", "31.03.2026"),
assert doc.page_count == 24, ( ("04.2026", "30.04.2026"),
f"Erwartet 24 Seiten (12 Monate × 2 Seiten), aber {doc.page_count}" ("05.2026", "31.05.2026"),
) ]
for index, (month, month_end) in enumerate(expected_months):
# 6. Prüfe, dass KEINE _Field-Platzhalter mehr im Output sind data_text = doc[index * 2].get_text()
full_text = "" signature_text = doc[index * 2 + 1].get_text()
for page in doc: for value in [
full_text += page.get_text() "100781516", "Bardaus, Madalina", "50,00", month,
assert "_Field" not in full_text, ( "Psychologin", "Psychologische Betreuung",
"Es sind noch _Field-Platzhalter im Output-PDF vorhanden!" ]:
) assert value in data_text, f"{value!r} fehlt für {month}"
assert "100781516" not in signature_text
# 7. Prüfe Seite 0 (Januar, Seite 1 – Stammdaten) assert "Dresden" in signature_text
jan0_text = doc[0].get_text() assert month_end in signature_text
assert "100740439" in jan0_text, "Antragsnummer fehlt auf Januar-Seite 1"
assert "Draßdo, Cornelia" in jan0_text, "Name fehlt auf Januar-Seite 1"
assert "50" in jan0_text, "Stellenanteil fehlt auf Januar-Seite 1"
assert "Meisterabschluß Küche" in jan0_text, "Qualifikation fehlt auf Januar-Seite 1"
assert "01.2025" in jan0_text, "Monat fehlt auf Januar-Seite 1"
assert "Anleiterin / Ausbilderin" in jan0_text, "Funktion fehlt auf Januar-Seite 1"
assert "Tätigkeiten im Praxisbereich Garten" in jan0_text, "Tätigkeit fehlt auf Januar-Seite 1"
# 8. Prüfe Seite 1 (Januar, Seite 2 – Unterschriftsblock mit Ort/Datum)
jan1_text = doc[1].get_text()
assert "100740439" not in jan1_text, "Antragsnummer sollte nicht auf Seite 2 sein"
assert "Dresden" in jan1_text, "Ort 'Dresden' fehlt auf Januar-Seite 2"
assert "31.01.2025" in jan1_text, "Datum '31.01.2025' fehlt auf Januar-Seite 2"
# Tätigkeit wiederholt auf Seite 2
assert "Tätigkeiten im Praxisbereich Garten" in jan1_text, "Tätigkeit fehlt auf Januar-Seite 2"
# 9. Prüfe Juli (Seite 12+13) – Stellenanteil wechselt auf 40%
jul0_text = doc[12].get_text()
assert "40" in jul0_text, "Stellenanteil 40% fehlt auf Juli-Seite 1"
assert "07.2025" in jul0_text, "Monat 07.2025 fehlt auf Juli-Seite 1"
jul1_text = doc[13].get_text()
assert "31.07.2025" in jul1_text, "Datum 31.07.2025 fehlt auf Juli-Seite 2"
doc.close()
@@ -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()
+1
View File
@@ -1,6 +1,7 @@
@echo off @echo off
chcp 65001 >nul chcp 65001 >nul
setlocal enabledelayedexpansion setlocal enabledelayedexpansion
set "PYTHONUTF8=1"
cd /d "%~dp0" cd /d "%~dp0"
cd .. cd ..
+2
View File
@@ -76,6 +76,8 @@ except ImportError:
# ── Data Files ────────────────────────────────────────────────── # ── Data Files ──────────────────────────────────────────────────
datas = [ datas = [
(str(_APP_DIR / "app.py"), "."), (str(_APP_DIR / "app.py"), "."),
(str(_APP_DIR / "version_check.py"), "."),
(str(_APP_DIR / "CHANGELOG.md"), "."),
(str(_APP_DIR / "src"), "src"), (str(_APP_DIR / "src"), "src"),
(str(_APP_DIR / "config"), "config"), (str(_APP_DIR / "config"), "config"),
] ]
+13
View File
@@ -12,12 +12,17 @@ Dieses Skript:
""" """
import os import os
import subprocess
import sys import sys
import threading import threading
import time import time
import webbrowser import webbrowser
from pathlib import Path from pathlib import Path
if not getattr(sys, "frozen", False):
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "0_application"))
from server_lifecycle import stop_existing_server
def _get_base_dir() -> Path: def _get_base_dir() -> Path:
"""Verzeichnis der .exe (bzw. des Skripts im Entwicklungsmodus).""" """Verzeichnis der .exe (bzw. des Skripts im Entwicklungsmodus)."""
@@ -44,6 +49,14 @@ def _ensure_dirs(base_dir: Path) -> None:
def main() -> None: def main() -> None:
# Only open the browser/start Streamlit once the previous server has stopped.
try:
if not stop_existing_server():
print("[INFO] Die bisherige Version bleibt geöffnet. Keine neue Instanz gestartet.")
return
except (RuntimeError, OSError, subprocess.SubprocessError) as exc:
print(f"[FEHLER] Server-Neustart fehlgeschlagen: {exc}", file=sys.stderr)
raise SystemExit(1)
base_dir = _get_base_dir() base_dir = _get_base_dir()
# Im gebundelten Modus: sys._MEIPASS enthält src/ und app.py # Im gebundelten Modus: sys._MEIPASS enthält src/ und app.py