Compare commits
5
Commits
v1.3
..
7d63063024
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7d63063024 | ||
|
|
2aad2c9d99 | ||
|
|
971459c9bf | ||
|
|
1519b96c7b | ||
|
|
eaffa78e9c |
No files matched your search
@@ -61,3 +61,7 @@ $RECYCLE.BIN/
|
||||
*.vsix
|
||||
3_Archiv
|
||||
2_build
|
||||
|
||||
# Python-Test- und Build-Artefakte
|
||||
__pycache__/
|
||||
.pytest_cache/
|
||||
@@ -7,6 +7,22 @@ und dieses Projekt folgt [Semantic Versioning](https://semver.org/lang/de/).
|
||||
|
||||
---
|
||||
|
||||
## [1.3.2] – 2026-10-07
|
||||
|
||||
### Behoben
|
||||
|
||||
- EXE erkennt laufende Entwicklungsinstanz und fragt vor dem Neustart
|
||||
|
||||
---
|
||||
|
||||
## [1.3.1] – 2026-10-07
|
||||
|
||||
### Behoben
|
||||
|
||||
- Python script erkennt jetzt laufende Entwicklungsinstanz und fragt vor dem Neustart
|
||||
|
||||
---
|
||||
|
||||
## [1.3] – 2026-08-11
|
||||
|
||||
### Hinzugefügt
|
||||
|
||||
@@ -87,8 +87,27 @@ Mapping von Input-Feldern auf Template-Zellen:
|
||||
|
||||
## Tests
|
||||
|
||||
End-to-End-Tests parsen die reale Input-Datei (`input/2025 JM.xlsx`) und das reale
|
||||
Template (`template/Tätigkeitsnachweise Stellenförderung.xlsx`) und validieren
|
||||
Beim Start über `start.bat` oder die neu gebaute EXE erscheint bei einer bereits
|
||||
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.
|
||||
Beim Build laufen die Tests automatisch vor der Versionierung und dem Bauen.
|
||||
Beim Programmstart werden sie nicht ausgeführt.
|
||||
|
||||
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.
|
||||
|
||||
```bash
|
||||
@@ -96,12 +115,67 @@ den gesamten Generierungs-Workflow.
|
||||
.venv/bin/pip install pytest
|
||||
|
||||
# Tests ausführen
|
||||
.venv/bin/python -m pytest tests/test_e2e.py -v
|
||||
.venv/bin/python -m pytest tests -v
|
||||
```
|
||||
|
||||
Die Tests prüfen:
|
||||
- Vorhandensein der Input- und Template-Dateien
|
||||
- Korrekte Parsierung (168 Datensätze, 19 Personen)
|
||||
- Vollständige Pipeline für die erste Person („Draßdo, Cornelia"): 12 Monats-Sheets
|
||||
mit allen Zellinhalten (Antragsnummer, Name, Stellenanteil, Qualifikation, Monat,
|
||||
Funktion, Tätigkeit, Ort, Datum)
|
||||
- Korrekte Parsierung (100 Datensätze, 17 Personen)
|
||||
- Vollständige PDF-Pipeline für die erste Person („Bardaus, Madalina“): fünf Monate
|
||||
von Januar bis Mai 2026 mit zehn Seiten, Antragsnummer, Name, Stellenanteil,
|
||||
Monat, Funktion, Tätigkeit, Ort und korrekten Monatsenddaten
|
||||
- Neustartlogik unter Windows mit simulierten Prozessen
|
||||
|
||||
## Build und Deployment unter Windows
|
||||
|
||||
Alle folgenden Befehle werden im Projekt-Hauptordner ausgeführt:
|
||||
|
||||
```powershell
|
||||
# Interaktiv: Versionsänderung auswählen
|
||||
.\1_build-pipeline\build.bat
|
||||
|
||||
# Bugfix-Version mit Changelog-Beschreibung
|
||||
.\1_build-pipeline\build.bat --bump patch --message "Fehler beim Programmstart behoben"
|
||||
|
||||
# Ablauf anzeigen, ohne Tests, Build oder Deployment auszuführen
|
||||
.\1_build-pipeline\build.bat --bump patch --dry-run
|
||||
|
||||
# Aktuelle Version mit aktualisiertem Changelog erneut als EXE bauen
|
||||
.\1_build-pipeline\build.bat --no-bump --no-deploy
|
||||
|
||||
# Nur den letzten erfolgreichen Build bereitstellen (kein neuer Build)
|
||||
.\1_build-pipeline\deploy.bat
|
||||
|
||||
# Deployment ohne abschließende Tasteneingabe
|
||||
.\1_build-pipeline\deploy.bat --no-pause
|
||||
```
|
||||
|
||||
Die Pipeline führt zuerst alle Tests aus. Bei einem Testfehler bleiben Version
|
||||
und Changelog unverändert; Build und Deployment werden nicht gestartet.
|
||||
Nach erfolgreichen Tests aktualisiert sie die Version, erstellt die EXE und
|
||||
legt sie mit Konfiguration, PDF-Template, README und Changelog in `2_build/` ab.
|
||||
`release.json` hält die Version und Prüfsummen des erfolgreichen Builds fest.
|
||||
Bei einem Buildfehler wird der Changelog zurückgesetzt.
|
||||
|
||||
Danach erzeugt das Deployment `Tätigkeitsnachweis-Generator vX.Y.Z.zip` und
|
||||
überträgt es an:
|
||||
|
||||
```text
|
||||
j:\Fachbereiche\BO\Produktionsschule\Maßnahmeunterlagen\20_Personal\05_Tätigkeitsnachweise ESF\Nachweise-automatisiert-generieren\
|
||||
```
|
||||
|
||||
Das Ziel muss bereits vorhanden und über VPN/SMB erreichbar sein. Ist es nicht
|
||||
erreichbar oder fehlen Schreibrechte, bricht das Deployment verständlich ab.
|
||||
Der lokale Build bleibt erhalten. Nach Wiederherstellung der VPN-Verbindung
|
||||
kann `deploy.bat` die Bereitstellung ohne erneutes Testen, Bauen oder Erhöhen
|
||||
der Versionsnummer wiederholen. Das Skript akzeptiert nur einen bestätigten,
|
||||
unveränderten Build mit `release.json`; ältere Builds müssen einmal neu gebaut
|
||||
werden.
|
||||
|
||||
Das ZIP enthält leere `input/`- und `output/`-Ordner; Arbeitsdateien und
|
||||
Testdaten werden nicht mit ausgeliefert. Es wird zunächst als temporäre Datei
|
||||
übertragen, per Prüfsumme geprüft und erst dann unter dem Versionsnamen
|
||||
sichtbar gemacht. Bestehende Versionen mit anderem Inhalt werden nicht
|
||||
überschrieben. Ein identisches Paket kann erneut bereitgestellt werden.
|
||||
Die Neustarttests simulieren Prozesse und Dialogantworten; ein manueller
|
||||
Starttest der fertigen EXE bleibt vor der Weitergabe sinnvoll.
|
||||
@@ -1,4 +1,6 @@
|
||||
@echo off
|
||||
chcp 65001 >nul
|
||||
set "PYTHONUTF8=1"
|
||||
title Tätigkeitsnachweis-Generator – Cold Run
|
||||
echo ========================================
|
||||
echo Cold Run – Headless-Generierung
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Stop an existing instance of this application's Windows server."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
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 = Path(__file__).resolve().with_name("app.py")
|
||||
app_paths = [str(app_path)]
|
||||
if getattr(sys, "frozen", False):
|
||||
# A locally built EXE in 2_build/ must also recognize the sibling
|
||||
# development application, whose path differs from _MEIPASS/app.py.
|
||||
exe_dir = Path(sys.executable).resolve().parent
|
||||
development_app = exe_dir.parent / "0_application" / "app.py"
|
||||
if exe_dir.name.lower() == "2_build" and development_app.is_file():
|
||||
app_paths.append(str(development_app))
|
||||
# Pass paths through the environment, never interpolate them into shell code.
|
||||
env = dict(os.environ, APP_RESTART_PATH=str(app_path),
|
||||
APP_RESTART_PATHS=json.dumps(app_paths), 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
|
||||
$matchesApp = $false
|
||||
foreach ($appPath in ($env:APP_RESTART_PATHS | ConvertFrom-Json)) {
|
||||
$appArgument = '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+"?' +
|
||||
[regex]::Escape($appPath) + '"?(?:\s|$)'
|
||||
if ($command -match $appArgument) { $matchesApp = $true }
|
||||
}
|
||||
$isDev = $process.Name -match '^(python(?:w)?(?:\d+(?:\.\d+)*)?(?:w)?|streamlit)\.exe$' -and
|
||||
$command -match '(?i)\bstreamlit(?:\.exe)?"?\s+run\s+' -and
|
||||
($matchesApp -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
|
||||
chcp 65001 >nul
|
||||
set "PYTHONUTF8=1"
|
||||
cd /d "%~dp0"
|
||||
title Tätigkeitsnachweis-Generator
|
||||
echo ========================================
|
||||
echo Tätigkeitsnachweis-Generator
|
||||
@@ -43,6 +46,12 @@ echo.
|
||||
echo [INFO] Starte Tätigkeitsnachweis-Generator...
|
||||
echo [INFO] Die App öffnet sich im Browser unter http://localhost:8501
|
||||
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
|
||||
@@ -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%
|
||||
@@ -27,7 +27,7 @@ echo ""
|
||||
echo "[INFO] Führe End-to-End-Tests aus…"
|
||||
echo ""
|
||||
|
||||
"$VENV_DIR/bin/python" -m pytest tests/test_e2e.py -v
|
||||
"$VENV_DIR/bin/python" -m pytest tests -v
|
||||
|
||||
echo ""
|
||||
read -p "Drücke Enter zum Beenden..."
|
||||
Binary file not shown.
@@ -0,0 +1,185 @@
|
||||
"""Verify pipeline gates and deployment against temporary local directories."""
|
||||
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from unittest.mock import patch, Mock
|
||||
import zipfile
|
||||
|
||||
import pytest
|
||||
|
||||
PIPELINE_DIR = Path(__file__).resolve().parents[2] / "1_build-pipeline"
|
||||
sys.path.insert(0, str(PIPELINE_DIR))
|
||||
import deploy
|
||||
|
||||
spec = importlib.util.spec_from_file_location("build_pipeline", PIPELINE_DIR / "build.py")
|
||||
build = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(build)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def release(tmp_path):
|
||||
directory = tmp_path / "build"
|
||||
directory.mkdir()
|
||||
for name in deploy.PACKAGE_FILES:
|
||||
path = directory / name
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(b"test release payload")
|
||||
deploy.write_release_manifest(directory, "1.3")
|
||||
target = tmp_path / "share"
|
||||
target.mkdir()
|
||||
return directory, target
|
||||
|
||||
|
||||
def test_deployment_packages_release_without_work_files(release):
|
||||
directory, target = release
|
||||
for folder in ["input", "output"]:
|
||||
(directory / folder).mkdir()
|
||||
(directory / folder / "private.xlsx").write_bytes(b"private")
|
||||
result = deploy.deploy(directory, target)
|
||||
assert result.name == "Tätigkeitsnachweis-Generator v1.3.0.zip"
|
||||
with zipfile.ZipFile(result) as archive:
|
||||
assert set(archive.namelist()) == set(deploy.PACKAGE_FILES + ["input/", "output/"])
|
||||
assert archive.testzip() is None
|
||||
assert not list(target.glob(".upload-*"))
|
||||
# An unchanged release can be retried without overwriting the published ZIP.
|
||||
original_time = result.stat().st_mtime_ns
|
||||
assert deploy.deploy(directory, target) == result
|
||||
assert result.stat().st_mtime_ns == original_time
|
||||
|
||||
|
||||
def test_missing_vpn_target_is_not_created(release):
|
||||
directory, target = release
|
||||
missing = target / "missing-share"
|
||||
with pytest.raises(OSError, match="VPN"):
|
||||
deploy.deploy(directory, missing)
|
||||
assert not missing.exists()
|
||||
assert not list(directory.glob("*.zip"))
|
||||
|
||||
|
||||
def test_changed_payload_is_not_deployed(release):
|
||||
directory, target = release
|
||||
(directory / deploy.EXE_NAME).write_bytes(b"different executable")
|
||||
with pytest.raises(ValueError, match="verändert"):
|
||||
deploy.deploy(directory, target)
|
||||
assert not list(target.iterdir())
|
||||
|
||||
|
||||
def test_unconfirmed_old_build_is_not_deployed(release):
|
||||
directory, target = release
|
||||
(directory / deploy.MANIFEST_NAME).unlink()
|
||||
with pytest.raises(ValueError, match="bestätigter Build"):
|
||||
deploy.deploy(directory, target)
|
||||
|
||||
|
||||
def test_existing_different_release_is_preserved(release):
|
||||
directory, target = release
|
||||
existing = target / "Tätigkeitsnachweis-Generator v1.3.0.zip"
|
||||
existing.write_bytes(b"already published")
|
||||
with pytest.raises(FileExistsError, match="anderem Inhalt"):
|
||||
deploy.deploy(directory, target)
|
||||
assert existing.read_bytes() == b"already published"
|
||||
|
||||
|
||||
def test_interrupted_upload_is_not_published(release):
|
||||
directory, target = release
|
||||
with patch.object(deploy.shutil, "copyfile", side_effect=OSError("VPN disconnected")):
|
||||
with pytest.raises(OSError, match="lokale Build bleibt erhalten"):
|
||||
deploy.deploy(directory, target)
|
||||
assert not list(target.iterdir())
|
||||
assert list(directory.glob("*.zip"))
|
||||
assert deploy.validate_release(directory) == "1.3.0"
|
||||
|
||||
|
||||
def test_corrupt_upload_is_not_published(release):
|
||||
directory, target = release
|
||||
def corrupt_copy(source, destination):
|
||||
Path(destination).write_bytes(b"corrupted")
|
||||
with patch.object(deploy.shutil, "copyfile", side_effect=corrupt_copy):
|
||||
with pytest.raises(OSError):
|
||||
deploy.deploy(directory, target)
|
||||
assert not list(target.iterdir())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def pipeline(tmp_path, monkeypatch):
|
||||
changelog = tmp_path / "CHANGELOG.md"
|
||||
changelog.write_text("## [1.3]\n\nOriginal release\n", encoding="utf-8")
|
||||
version_file = tmp_path / ".last_version"
|
||||
version_file.write_text("1.3", encoding="utf-8")
|
||||
monkeypatch.setattr(build, "CHANGELOG_PATH", changelog)
|
||||
monkeypatch.setattr(build, "VERSION_FILE", version_file)
|
||||
monkeypatch.setattr(build, "BUILD_OUTPUT_DIR", tmp_path)
|
||||
monkeypatch.setattr(sys, "argv", ["build.py", "--bump", "patch", "--message", "Fix"])
|
||||
events = []
|
||||
names = ["_run_tests", "_ensure_pyinstaller", "_prepare_build_dir", "_run_pyinstaller",
|
||||
"_copy_to_build", "write_release_manifest", "_cleanup", "_print_summary", "deploy"]
|
||||
mocks = {}
|
||||
for name in names:
|
||||
mocks[name] = Mock(side_effect=lambda *args, _name=name: events.append(_name))
|
||||
monkeypatch.setattr(build, name, mocks[name])
|
||||
return changelog, version_file, events, mocks
|
||||
|
||||
|
||||
def test_pipeline_tests_build_then_deploy(pipeline):
|
||||
changelog, version_file, events, mocks = pipeline
|
||||
build.main()
|
||||
assert events == ["_run_tests", "_ensure_pyinstaller", "_prepare_build_dir", "_run_pyinstaller",
|
||||
"_copy_to_build", "write_release_manifest", "_cleanup", "_print_summary", "deploy"]
|
||||
assert version_file.read_text() == "1.3.1"
|
||||
assert "## [1.3.1]" in changelog.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_failing_tests_prevent_version_change_build_and_deploy(pipeline):
|
||||
changelog, version_file, events, mocks = pipeline
|
||||
original = changelog.read_bytes()
|
||||
mocks["_run_tests"].side_effect = SystemExit(1)
|
||||
with pytest.raises(SystemExit):
|
||||
build.main()
|
||||
assert changelog.read_bytes() == original
|
||||
assert version_file.read_text() == "1.3"
|
||||
mocks["_run_pyinstaller"].assert_not_called()
|
||||
mocks["deploy"].assert_not_called()
|
||||
|
||||
|
||||
def test_build_failure_rolls_back_changelog_and_prevents_deployment(pipeline):
|
||||
changelog, version_file, events, mocks = pipeline
|
||||
original = changelog.read_text(encoding="utf-8")
|
||||
mocks["_run_pyinstaller"].side_effect = SystemExit(1)
|
||||
with pytest.raises(SystemExit):
|
||||
build.main()
|
||||
assert changelog.read_text(encoding="utf-8") == original
|
||||
assert version_file.read_text() == "1.3"
|
||||
mocks["deploy"].assert_not_called()
|
||||
mocks["_cleanup"].assert_called_once()
|
||||
|
||||
|
||||
def test_deploy_failure_keeps_successful_build_version(pipeline):
|
||||
changelog, version_file, events, mocks = pipeline
|
||||
mocks["deploy"].side_effect = OSError("VPN unavailable")
|
||||
with pytest.raises(SystemExit):
|
||||
build.main()
|
||||
assert version_file.read_text() == "1.3.1"
|
||||
assert "## [1.3.1]" in changelog.read_text(encoding="utf-8")
|
||||
mocks["write_release_manifest"].assert_called_once()
|
||||
|
||||
|
||||
def test_dry_run_does_not_test_build_or_deploy(pipeline, monkeypatch):
|
||||
changelog, version_file, events, mocks = pipeline
|
||||
monkeypatch.setattr(sys, "argv", ["build.py", "--bump", "patch", "--dry-run"])
|
||||
build.main()
|
||||
assert not events
|
||||
assert version_file.read_text() == "1.3"
|
||||
|
||||
|
||||
def test_rebuild_preserves_changelog_and_version(pipeline, monkeypatch):
|
||||
changelog, version_file, events, mocks = pipeline
|
||||
original = changelog.read_bytes()
|
||||
monkeypatch.setattr(sys, "argv", ["build.py", "--no-bump", "--no-deploy"])
|
||||
build.main()
|
||||
assert changelog.read_bytes() == original
|
||||
assert version_file.read_text() == "1.3"
|
||||
mocks["write_release_manifest"].assert_called_once_with(build.BUILD_OUTPUT_DIR, "1.3")
|
||||
mocks["_run_tests"].assert_called_once()
|
||||
mocks["_run_pyinstaller"].assert_called_once()
|
||||
mocks["deploy"].assert_not_called()
|
||||
+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
|
||||
from pathlib import Path
|
||||
|
||||
import fitz # PyMuPDF
|
||||
import fitz
|
||||
import pytest
|
||||
|
||||
# Projekt-Root zum Python-Path hinzufügen
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
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.writer import generate_activity_reports
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Pfade zu den realen Fixture-Dateien
|
||||
# ──────────────────────────────────────────────
|
||||
|
||||
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"
|
||||
|
||||
|
||||
def _load_app_config():
|
||||
"""Lädt die echte App-Konfiguration (Schema, Mapping, Settings)."""
|
||||
@pytest.fixture(scope="module")
|
||||
def app_config():
|
||||
config = load_settings(CONFIG_DIR, PROJECT_ROOT)
|
||||
config.input_schema = load_input_schema(CONFIG_DIR)
|
||||
config.output_mapping = load_output_mapping(CONFIG_DIR)
|
||||
return config
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────
|
||||
# Fixtures
|
||||
# ──────────────────────────────────────────────
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def app_config():
|
||||
"""Reale App-Konfiguration aus config/."""
|
||||
return _load_app_config()
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
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(
|
||||
filepath=INPUT_FILE,
|
||||
schema=app_config.input_schema,
|
||||
@@ -56,165 +40,82 @@ def parsed_records(app_config):
|
||||
|
||||
@pytest.fixture
|
||||
def output_dir(tmp_path):
|
||||
"""Temporäres Output-Verzeichnis."""
|
||||
out = tmp_path / "output"
|
||||
out.mkdir()
|
||||
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:
|
||||
"""End-to-End: Reale Input-Datei + reales PDF-Template → nur erste Person."""
|
||||
|
||||
def test_input_file_exists(self):
|
||||
"""Stellt sicher, dass die reale Input-Datei vorhanden ist."""
|
||||
assert INPUT_FILE.exists(), f"Input-Datei nicht gefunden: {INPUT_FILE}"
|
||||
assert INPUT_FILE.is_file(), f"Testdatei nicht gefunden: {INPUT_FILE}"
|
||||
|
||||
def test_template_file_exists(self):
|
||||
"""Stellt sicher, dass das reale PDF-Template vorhanden ist."""
|
||||
assert TEMPLATE_FILE.exists(), f"PDF-Template nicht gefunden: {TEMPLATE_FILE}"
|
||||
assert TEMPLATE_FILE.is_file(), f"PDF-Template nicht gefunden: {TEMPLATE_FILE}"
|
||||
|
||||
def test_template_has_field_markers(self, app_config):
|
||||
"""Stellt sicher, dass alle _Field-Platzhalter im PDF-Template vorhanden sind."""
|
||||
doc = fitz.open(TEMPLATE_FILE)
|
||||
missing = []
|
||||
for m in app_config.output_mapping:
|
||||
if m.pdf_field is None:
|
||||
continue
|
||||
# Suche auf allen Seiten
|
||||
found_any = False
|
||||
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}"
|
||||
)
|
||||
with fitz.open(TEMPLATE_FILE) as doc:
|
||||
missing = [
|
||||
mapping.pdf_field
|
||||
for mapping in app_config.output_mapping
|
||||
if mapping.pdf_field is not None
|
||||
and not any(page.search_for(mapping.pdf_field) for page in doc)
|
||||
]
|
||||
assert not missing, f"Folgende _Field-Platzhalter fehlen im PDF: {missing}"
|
||||
|
||||
def test_parse_real_input(self, parsed_records):
|
||||
"""Validiert das Parsen der realen Input-Datei."""
|
||||
# Sollte 168 Einträge über 12 Monate für 19 Personen ergeben
|
||||
assert len(parsed_records) == 168, (
|
||||
f"Erwartet 168 Records, aber {len(parsed_records)} gefunden"
|
||||
)
|
||||
|
||||
# Fixed expectations for tests/fixture/2026 JM.xlsx.
|
||||
assert len(parsed_records) == 100
|
||||
employees = get_unique_employees(parsed_records)
|
||||
assert len(employees) == 19, (
|
||||
f"Erwartet 19 Mitarbeiter, aber {len(employees)} gefunden"
|
||||
)
|
||||
assert len(employees) == 17
|
||||
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
|
||||
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
|
||||
def test_full_pipeline_first_person(self, app_config, parsed_records, output_dir):
|
||||
grouped = group_by_employee(parsed_records)
|
||||
assert len(grouped) == 19
|
||||
|
||||
# 2. Erste Person alphabetisch
|
||||
first_person = sorted(grouped.keys())[0]
|
||||
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}
|
||||
assert len(grouped) == 17
|
||||
first_person = sorted(grouped)[0]
|
||||
assert first_person == "Bardaus, Madalina"
|
||||
assert len(grouped[first_person]) == 5
|
||||
generated = generate_activity_reports(
|
||||
employee_records=single_person_group,
|
||||
employee_records={first_person: grouped[first_person]},
|
||||
template_path=TEMPLATE_FILE,
|
||||
mapping=app_config.output_mapping,
|
||||
output_dir=output_dir,
|
||||
filename_pattern=app_config.filename_pattern,
|
||||
fontsize=app_config.fontsize,
|
||||
)
|
||||
|
||||
# 4. Ergebnisse prüfen
|
||||
assert len(generated) == 1, (
|
||||
f"Erwartet 1 generierte Datei, aber {len(generated)}"
|
||||
)
|
||||
assert len(generated) == 1
|
||||
output_file = generated[0]
|
||||
assert output_file.exists(), f"Output-Datei nicht gefunden: {output_file}"
|
||||
assert output_file.suffix == ".pdf"
|
||||
|
||||
erwarteter_name = "Draßdo_Cornelia_Tätigkeitsnachweis.pdf"
|
||||
assert output_file.name == erwarteter_name, (
|
||||
f"Erwartet '{erwarteter_name}', aber '{output_file.name}'"
|
||||
)
|
||||
|
||||
# 5. PDF öffnen und Inhalt validieren
|
||||
doc = fitz.open(output_file)
|
||||
assert doc.page_count == 24, (
|
||||
f"Erwartet 24 Seiten (12 Monate × 2 Seiten), aber {doc.page_count}"
|
||||
)
|
||||
|
||||
# 6. Prüfe, dass KEINE _Field-Platzhalter mehr im Output sind
|
||||
full_text = ""
|
||||
for page in doc:
|
||||
full_text += page.get_text()
|
||||
assert "_Field" not in full_text, (
|
||||
"Es sind noch _Field-Platzhalter im Output-PDF vorhanden!"
|
||||
)
|
||||
|
||||
# 7. Prüfe Seite 0 (Januar, Seite 1 – Stammdaten)
|
||||
jan0_text = doc[0].get_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()
|
||||
assert output_file.is_file()
|
||||
assert output_file.name == "Bardaus_Madalina_Tätigkeitsnachweis.pdf"
|
||||
with fitz.open(output_file) as doc:
|
||||
assert doc.page_count == 10 # Five months, two pages per month.
|
||||
assert "_Field" not in "".join(page.get_text() for page in doc)
|
||||
# Independently specified month ends include February's 28 days.
|
||||
expected_months = [
|
||||
("01.2026", "31.01.2026"),
|
||||
("02.2026", "28.02.2026"),
|
||||
("03.2026", "31.03.2026"),
|
||||
("04.2026", "30.04.2026"),
|
||||
("05.2026", "31.05.2026"),
|
||||
]
|
||||
for index, (month, month_end) in enumerate(expected_months):
|
||||
data_text = doc[index * 2].get_text()
|
||||
signature_text = doc[index * 2 + 1].get_text()
|
||||
for value in [
|
||||
"100781516", "Bardaus, Madalina", "50,00", month,
|
||||
"Psychologin", "Psychologische Betreuung",
|
||||
]:
|
||||
assert value in data_text, f"{value!r} fehlt für {month}"
|
||||
assert "100781516" not in signature_text
|
||||
assert "Dresden" in signature_text
|
||||
assert month_end in signature_text
|
||||
@@ -0,0 +1,192 @@
|
||||
"""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()
|
||||
@@ -1 +1 @@
|
||||
1.3
|
||||
1.3.2
|
||||
@@ -1,6 +1,7 @@
|
||||
@echo off
|
||||
chcp 65001 >nul
|
||||
setlocal enabledelayedexpansion
|
||||
set "PYTHONUTF8=1"
|
||||
|
||||
cd /d "%~dp0"
|
||||
cd ..
|
||||
@@ -36,11 +37,11 @@ REM ── Build ausführen ─────────────────
|
||||
|
||||
if errorlevel 1 (
|
||||
echo.
|
||||
echo [FEHLER] Build fehlgeschlagen!
|
||||
echo [FEHLER] Pipeline fehlgeschlagen. Details stehen oben.
|
||||
pause
|
||||
exit /b 1
|
||||
)
|
||||
|
||||
echo.
|
||||
echo Build abgeschlossen.
|
||||
echo Tests, Build und Deployment abgeschlossen.
|
||||
pause
|
||||
+68
-15
@@ -2,7 +2,8 @@
|
||||
|
||||
Liest die aktuelle Version aus CHANGELOG.md, bump't sie gemäß
|
||||
Benutzereingabe (major/minor/patch), aktualisiert das Changelog
|
||||
und baut via PyInstaller eine standalone .exe nach 2_build/.
|
||||
führt zuerst alle Tests aus, baut via PyInstaller eine standalone .exe
|
||||
nach 2_build/ und stellt sie anschließend als ZIP auf dem SMB-Laufwerk bereit.
|
||||
|
||||
Verwendung:
|
||||
python build.py # interaktiver Modus
|
||||
@@ -20,6 +21,8 @@ from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from deploy import deploy, write_release_manifest
|
||||
|
||||
# ── Pfade ───────────────────────────────────────────────────────
|
||||
BUILD_PIPELINE_DIR = Path(__file__).resolve().parent
|
||||
PROJECT_ROOT = BUILD_PIPELINE_DIR.parent
|
||||
@@ -177,6 +180,21 @@ def _build_sections(bump_type: str, message: Optional[str]) -> str:
|
||||
|
||||
# ── Build ───────────────────────────────────────────────────────
|
||||
|
||||
def _run_tests() -> None:
|
||||
"""Test with the same interpreter used to build the application."""
|
||||
try:
|
||||
import pytest # noqa: F401
|
||||
except ImportError:
|
||||
subprocess.check_call([sys.executable, "-m", "pip", "install", "pytest"])
|
||||
print("\n[INFO] Führe alle Tests aus ...", flush=True)
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "pytest", str(APP_DIR / "tests"), "-v", "-p", "no:cacheprovider"],
|
||||
cwd=str(PROJECT_ROOT),
|
||||
)
|
||||
if result.returncode != 0:
|
||||
print("[FEHLER] Tests fehlgeschlagen. Kein Build und kein Deployment.")
|
||||
raise SystemExit(1)
|
||||
|
||||
def _ensure_pyinstaller() -> None:
|
||||
"""Stellt sicher, dass PyInstaller installiert ist."""
|
||||
try:
|
||||
@@ -196,14 +214,20 @@ def _prepare_build_dir() -> None:
|
||||
target = BUILD_OUTPUT_DIR / folder
|
||||
target.mkdir(exist_ok=True)
|
||||
|
||||
# Kopiere config-Dateien, falls noch nicht vorhanden
|
||||
# Liefere die zum aktuellen Quellstand gehörende Konfiguration aus.
|
||||
config_src = APP_DIR / "config"
|
||||
config_dst = BUILD_OUTPUT_DIR / "config"
|
||||
if config_src.exists():
|
||||
for f in config_src.iterdir():
|
||||
if f.is_file() and not (config_dst / f.name).exists():
|
||||
if f.is_file():
|
||||
shutil.copy2(f, config_dst / f.name)
|
||||
|
||||
shutil.copy2(CHANGELOG_PATH, BUILD_OUTPUT_DIR / "CHANGELOG.md")
|
||||
shutil.copy2(
|
||||
APP_DIR / "template" / "Tätigkeitsnachweise Stellenförderung.pdf",
|
||||
BUILD_OUTPUT_DIR / "template" / "Tätigkeitsnachweise Stellenförderung.pdf",
|
||||
)
|
||||
|
||||
# README für input-Ordner
|
||||
input_readme = BUILD_OUTPUT_DIR / "input" / "_DATEIEN_HIER_ABLEGEN_.txt"
|
||||
if not input_readme.exists():
|
||||
@@ -308,11 +332,16 @@ def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Build mit Semantic Versioning",
|
||||
)
|
||||
parser.add_argument(
|
||||
version_options = parser.add_mutually_exclusive_group()
|
||||
version_options.add_argument(
|
||||
"--bump",
|
||||
choices=["major", "minor", "patch"],
|
||||
help="Version bump-Typ (ohne Angabe: interaktiv)",
|
||||
)
|
||||
version_options.add_argument("--no-bump", action="store_true",
|
||||
help="Aktuelle Version erneut bauen; Changelog unverändert übernehmen")
|
||||
parser.add_argument("--no-deploy", action="store_true",
|
||||
help="Nur lokal bauen, ohne anschließendes Deployment")
|
||||
parser.add_argument(
|
||||
"--message", "-m",
|
||||
help="Changelog-Beschreibung für diese Version",
|
||||
@@ -327,6 +356,8 @@ def main() -> None:
|
||||
# ── Version ermitteln ────────────────────────────────────
|
||||
current = get_current_version()
|
||||
if not current:
|
||||
if args.no_bump:
|
||||
parser.error("Für --no-bump muss eine Version im CHANGELOG.md vorhanden sein.")
|
||||
print("❌ Keine Version in CHANGELOG.md gefunden. Beginne mit 0.1.")
|
||||
current = "0.1"
|
||||
|
||||
@@ -334,7 +365,7 @@ def main() -> None:
|
||||
|
||||
# Bump-Typ bestimmen
|
||||
bump_type = args.bump
|
||||
if not bump_type:
|
||||
if not bump_type and not args.no_bump:
|
||||
print("\nWelche Art von Änderung?")
|
||||
print(" [1] major – inkompatible API-Änderungen")
|
||||
print(" [2] minor – neue Funktionalität (abwärtskompatibel)")
|
||||
@@ -346,32 +377,54 @@ def main() -> None:
|
||||
print("\nAbgebrochen.")
|
||||
sys.exit(0)
|
||||
|
||||
new_version = bump_version(current, bump_type)
|
||||
new_version = current if args.no_bump else bump_version(current, bump_type)
|
||||
print(f"🚀 Neue Version: {new_version}")
|
||||
|
||||
if args.dry_run:
|
||||
print("\n[Trockenlauf – kein Build]")
|
||||
print(f" Würde CHANGELOG.md aktualisieren: [{new_version}]")
|
||||
print(" Würde zuerst alle Tests ausführen")
|
||||
if not args.no_bump:
|
||||
print(f" Würde CHANGELOG.md aktualisieren: [{new_version}]")
|
||||
print(f" Würde PyInstaller-Build starten")
|
||||
if not args.no_deploy:
|
||||
print(" Würde anschließend ein versioniertes ZIP auf dem SMB-Laufwerk bereitstellen")
|
||||
return
|
||||
|
||||
# ── CHANGELOG aktualisieren ──────────────────────────────
|
||||
print("\n📝 Changelog …")
|
||||
update_changelog(new_version, bump_type, args.message)
|
||||
|
||||
# ── Build ────────────────────────────────────────────────
|
||||
_ensure_pyinstaller()
|
||||
_prepare_build_dir()
|
||||
_run_tests()
|
||||
|
||||
previous_changelog = _read_changelog()
|
||||
try:
|
||||
_ensure_pyinstaller()
|
||||
(BUILD_OUTPUT_DIR / "release.json").unlink(missing_ok=True)
|
||||
if not args.no_bump:
|
||||
print("\n📝 Changelog …")
|
||||
update_changelog(new_version, bump_type, args.message)
|
||||
_prepare_build_dir()
|
||||
exe_path = _run_pyinstaller()
|
||||
final_exe = _copy_to_build(exe_path)
|
||||
write_release_manifest(BUILD_OUTPUT_DIR, new_version)
|
||||
except (Exception, SystemExit, KeyboardInterrupt):
|
||||
_write_changelog(previous_changelog)
|
||||
raise
|
||||
finally:
|
||||
_cleanup()
|
||||
|
||||
_save_version(new_version)
|
||||
_print_summary(new_version, final_exe)
|
||||
if args.no_deploy:
|
||||
print("[INFO] Lokaler Build abgeschlossen; Deployment nicht angefordert.")
|
||||
return
|
||||
try:
|
||||
deploy()
|
||||
except (OSError, ValueError, KeyError) as exc:
|
||||
print(f"[FEHLER] {exc}")
|
||||
print("[INFO] Der Build ist fertig. Deployment später mit 1_build-pipeline\\deploy.bat wiederholen.")
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
try:
|
||||
main()
|
||||
except (OSError, ValueError, subprocess.CalledProcessError) as exc:
|
||||
print(f"[FEHLER] Pipeline abgebrochen: {exc}")
|
||||
raise SystemExit(1)
|
||||
@@ -76,6 +76,8 @@ except ImportError:
|
||||
# ── Data Files ──────────────────────────────────────────────────
|
||||
datas = [
|
||||
(str(_APP_DIR / "app.py"), "."),
|
||||
(str(_APP_DIR / "version_check.py"), "."),
|
||||
(str(_APP_DIR / "CHANGELOG.md"), "."),
|
||||
(str(_APP_DIR / "src"), "src"),
|
||||
(str(_APP_DIR / "config"), "config"),
|
||||
]
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
@echo off
|
||||
chcp 65001 >nul
|
||||
setlocal
|
||||
set "PYTHONUTF8=1"
|
||||
cd /d "%~dp0.."
|
||||
|
||||
if exist "0_application\.venv\Scripts\python.exe" (
|
||||
set "DEPLOY_PYTHON=0_application\.venv\Scripts\python.exe"
|
||||
) else if exist ".venv\Scripts\python.exe" (
|
||||
set "DEPLOY_PYTHON=.venv\Scripts\python.exe"
|
||||
) else (
|
||||
set "DEPLOY_PYTHON=python"
|
||||
)
|
||||
|
||||
"%DEPLOY_PYTHON%" "1_build-pipeline\deploy.py"
|
||||
set "DEPLOY_EXIT_CODE=%ERRORLEVEL%"
|
||||
if /I not "%~1"=="--no-pause" pause
|
||||
exit /b %DEPLOY_EXIT_CODE%
|
||||
@@ -0,0 +1,143 @@
|
||||
"""Package and publish the last successful build to the SMB update directory."""
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import tempfile
|
||||
import zipfile
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
BUILD_DIR = PROJECT_ROOT / "2_build"
|
||||
DEPLOY_DIR = Path(r"j:\Fachbereiche\BO\Produktionsschule\Maßnahmeunterlagen\20_Personal\05_Tätigkeitsnachweise ESF\Nachweise-automatisiert-generieren")
|
||||
EXE_NAME = "Tätigkeitsnachweis-Generator.exe"
|
||||
MANIFEST_NAME = "release.json"
|
||||
PACKAGE_FILES = [
|
||||
EXE_NAME, "README.md", "CHANGELOG.md",
|
||||
"config/settings.ini", "config/input-fields.json", "config/output-fields.json",
|
||||
"template/Tätigkeitsnachweise Stellenförderung.pdf",
|
||||
]
|
||||
|
||||
|
||||
def normalize_version(version: str) -> str:
|
||||
if not re.fullmatch(r"\d+\.\d+(?:\.\d+)?", version):
|
||||
raise ValueError(f"Ungültige Versionsnummer: {version}")
|
||||
return ".".join(str(int(part)) for part in (version.split(".") + ["0"])[:3])
|
||||
|
||||
|
||||
def file_hash(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as source:
|
||||
for block in iter(lambda: source.read(1024 * 1024), b""):
|
||||
digest.update(block)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def write_release_manifest(build_dir: Path, version: str) -> None:
|
||||
"""Record the version and all payload hashes after a successful build."""
|
||||
manifest = {
|
||||
"version": normalize_version(version),
|
||||
"files": {name: file_hash(build_dir / name) for name in PACKAGE_FILES},
|
||||
}
|
||||
(build_dir / MANIFEST_NAME).write_text(
|
||||
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def validate_release(build_dir: Path) -> str:
|
||||
if not (build_dir / MANIFEST_NAME).is_file():
|
||||
raise ValueError("Kein bestätigter Build vorhanden. Bitte zuerst build.bat ausführen.")
|
||||
manifest = json.loads((build_dir / MANIFEST_NAME).read_text(encoding="utf-8"))
|
||||
version = normalize_version(manifest["version"])
|
||||
for name in PACKAGE_FILES:
|
||||
path = build_dir / name
|
||||
if not path.is_file() or file_hash(path) != manifest.get("files", {}).get(name):
|
||||
raise ValueError(f"Build-Datei fehlt oder wurde nach dem Build verändert: {name}. Bitte neu bauen.")
|
||||
return version
|
||||
|
||||
|
||||
def check_deploy_target(target_dir: Path) -> None:
|
||||
"""Never create the SMB destination: a missing mount must be an error."""
|
||||
try:
|
||||
if not target_dir.is_dir():
|
||||
raise OSError("Verzeichnis ist nicht verfügbar")
|
||||
# Actually access the directory; exists() alone does not check read access.
|
||||
with os.scandir(target_dir) as entries:
|
||||
next(entries, None)
|
||||
except OSError as exc:
|
||||
raise OSError(
|
||||
f"Deployment-Ziel nicht erreichbar: {target_dir}\n"
|
||||
"Bitte VPN-Verbindung und Netzlaufwerk J: prüfen und deploy.bat erneut starten."
|
||||
) from exc
|
||||
|
||||
|
||||
def deploy(build_dir: Path = BUILD_DIR, target_dir: Path = DEPLOY_DIR) -> Path:
|
||||
version = validate_release(build_dir)
|
||||
check_deploy_target(target_dir)
|
||||
archive_name = f"Tätigkeitsnachweis-Generator v{version}.zip"
|
||||
archive = build_dir / archive_name
|
||||
descriptor, temporary = tempfile.mkstemp(prefix="release-", suffix=".zip", dir=build_dir)
|
||||
os.close(descriptor)
|
||||
temporary = Path(temporary)
|
||||
try:
|
||||
with zipfile.ZipFile(temporary, "w", zipfile.ZIP_DEFLATED) as package:
|
||||
for name in PACKAGE_FILES:
|
||||
package.write(build_dir / name, name)
|
||||
for directory in ["input/", "output/"]:
|
||||
# Fixed timestamps keep repeated deployments byte-identical.
|
||||
package.writestr(zipfile.ZipInfo(directory, date_time=(2020, 1, 1, 0, 0, 0)), b"")
|
||||
with zipfile.ZipFile(temporary) as package:
|
||||
if package.testzip() is not None:
|
||||
raise ValueError("Das erzeugte ZIP-Paket ist beschädigt.")
|
||||
temporary.replace(archive)
|
||||
finally:
|
||||
temporary.unlink(missing_ok=True)
|
||||
|
||||
destination = target_dir / archive_name
|
||||
if destination.exists():
|
||||
if file_hash(destination) == file_hash(archive):
|
||||
print(f"[OK] Diese Version ist bereits unverändert bereitgestellt: {destination}")
|
||||
return destination
|
||||
raise FileExistsError(f"{destination} existiert bereits mit anderem Inhalt. Bitte eine neue Version bauen.")
|
||||
|
||||
# Write access is checked by creating the staging file on the actual share.
|
||||
remote_temporary = None
|
||||
try:
|
||||
descriptor, name = tempfile.mkstemp(prefix=".upload-", suffix=".tmp", dir=target_dir)
|
||||
os.close(descriptor)
|
||||
remote_temporary = Path(name)
|
||||
shutil.copyfile(archive, remote_temporary)
|
||||
if file_hash(remote_temporary) != file_hash(archive):
|
||||
raise OSError("Die Prüfsumme nach der SMB-Übertragung stimmt nicht überein.")
|
||||
# On Windows rename refuses to overwrite another published release.
|
||||
remote_temporary.rename(destination)
|
||||
except OSError as exc:
|
||||
raise OSError(
|
||||
"Deployment fehlgeschlagen. VPN-Verbindung und Schreibrechte prüfen. "
|
||||
"Der erfolgreiche lokale Build bleibt erhalten; erneut mit deploy.bat starten."
|
||||
) from exc
|
||||
finally:
|
||||
if remote_temporary is not None:
|
||||
try:
|
||||
remote_temporary.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
pass # The VPN may have disconnected during transfer.
|
||||
print(f"[OK] Deployment abgeschlossen: {destination}")
|
||||
return destination
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Letzten erfolgreichen Build auf dem SMB-Laufwerk bereitstellen")
|
||||
parser.parse_args()
|
||||
try:
|
||||
deploy()
|
||||
except (OSError, ValueError, KeyError) as exc:
|
||||
print(f"[FEHLER] {exc}")
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -12,12 +12,17 @@ Dieses Skript:
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import webbrowser
|
||||
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:
|
||||
"""Verzeichnis der .exe (bzw. des Skripts im Entwicklungsmodus)."""
|
||||
@@ -44,6 +49,14 @@ def _ensure_dirs(base_dir: Path) -> 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()
|
||||
|
||||
# Im gebundelten Modus: sys._MEIPASS enthält src/ und app.py
|
||||
|
||||
Reference in new issue
Block a user