Compare commits
3
Commits
v1.3
...
971459c9bf
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
971459c9bf | ||
|
|
1519b96c7b | ||
|
|
eaffa78e9c |
No files matched your search
@@ -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
|
||||||
@@ -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
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
@@ -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
@@ -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,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 ..
|
||||||
|
|||||||
@@ -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"),
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in new issue
Block a user