From 3086b894c7ecbbf0a4a814cc9154cabb821ea096 Mon Sep 17 00:00:00 2001 From: Lukas Geiger Date: Thu, 1 Oct 2026 16:46:51 +0200 Subject: [PATCH 1/2] fix: preserve CSV exports and protect invoice originals --- .github/workflows/source-platform-smoke.yml | 7 +- CHANGELOG.md | 7 + UniversalInvoiceMail.py | 6 +- cli.py | 19 ++- csv_export.py | 51 +++++++ tests/test_csv_export_transaction.py | 139 ++++++++++++++++++++ 6 files changed, 218 insertions(+), 11 deletions(-) create mode 100644 csv_export.py create mode 100644 tests/test_csv_export_transaction.py diff --git a/.github/workflows/source-platform-smoke.yml b/.github/workflows/source-platform-smoke.yml index fe12b5d..04d277b 100644 --- a/.github/workflows/source-platform-smoke.yml +++ b/.github/workflows/source-platform-smoke.yml @@ -47,8 +47,11 @@ jobs: libxcb-keysyms1 libxcb-randr0 libxcb-render-util0 \ libxcb-xinerama0 libxcb-xfixes0 - - name: Install PySide6 - run: pip install PySide6>=6.5.0 + - name: Install PySide6 and pytest + run: python -m pip install "PySide6>=6.5.0" pytest + + - name: Run CSV publication regressions + run: python -m pytest -q tests/test_csv_export_transaction.py - name: Run source platform smoke env: diff --git a/CHANGELOG.md b/CHANGELOG.md index 150eab6..004260d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,13 @@ Alle wichtigen Änderungen an diesem Projekt werden in dieser Datei dokumentiert ## [Unreleased] +### Fixed +- CSV-Exporte aus GUI und CLI werden vollständig in einer eigenen temporären Datei + vorbereitet. Format-, Schreib- und Ersetzungsfehler erhalten vorhandene Exporte. + Rechnungsoriginale sowie Konfiguration und Datenbank sind einschließlich + Datei-Aliassen als Exportziel geschützt; auch ausgeschlossene Rechnungen zählen. + UTF-8-BOM, Semikolon, Spalten, Auswahl und Filter bleiben erhalten. + ### CLI Resilience, Amount Normalization & Error Handling Härtung (2026-09-30) - **Headless CLI Robustness & Fehlerbehebung (`cli.py`)**: - `_format_display_amount()`: Robuste Betragsformatierung für `--list-invoices` implementiert; verhindert unhandled `ValueError: Unknown format code 'f' for object of type 'str'` bei Rechnungsbeträgen im String- oder Währungsformat (`42.50`, `19,99 €`). diff --git a/UniversalInvoiceMail.py b/UniversalInvoiceMail.py index 130b95b..e8d4919 100644 --- a/UniversalInvoiceMail.py +++ b/UniversalInvoiceMail.py @@ -5406,9 +5406,11 @@ def export_invoices_csv(self, filepath: Optional[Union[str, Path]] = None) -> Op try: import csv + from csv_export import atomic_csv_output target_path = Path(filepath) - target_path.parent.mkdir(parents=True, exist_ok=True) - with open(target_path, 'w', newline='', encoding='utf-8-sig') as f: + protected_paths = [CONFIG_FILE, INVOICES_DB] + protected_paths.extend(inv.path for inv in self.invoices if inv.path) + with atomic_csv_output(target_path, protected_paths) as f: writer = csv.writer(f, delimiter=';') # Header mit erweiterten Feldern writer.writerow([ diff --git a/cli.py b/cli.py index dd09814..2c805c1 100644 --- a/cli.py +++ b/cli.py @@ -30,7 +30,9 @@ import sys from datetime import datetime from pathlib import Path -from typing import Any, List, Optional, Sequence, Tuple +from typing import Any, Iterable, List, Optional, Sequence, Tuple + +from csv_export import atomic_csv_output # Version and identity constants APP_NAME = "UniversalInvoiceMail" @@ -180,11 +182,12 @@ def export_invoices_to_csv( output_path: Path, profile_filter: Optional[str] = None, status_filter: Optional[str] = None, + protected_paths: Iterable[Path] = (), ) -> int: """Exports invoices to a semicolon-separated CSV file with UTF-8 BOM.""" - output_path.parent.mkdir(parents=True, exist_ok=True) - - filtered = invoices + originals = list(invoices) + protected = [*protected_paths, *(i.get("path") for i in originals if i.get("path"))] + filtered = originals if profile_filter: p_low = profile_filter.strip().lower() filtered = [i for i in filtered if p_low in _normalize_filter_profile(i.get("profile_name"))] @@ -192,7 +195,7 @@ def export_invoices_to_csv( s_low = status_filter.strip().lower() filtered = [i for i in filtered if s_low == _normalize_filter_status(i.get("review_status"))] - with open(output_path, "w", newline="", encoding="utf-8-sig") as f: + with atomic_csv_output(output_path, protected) as f: writer = csv.writer(f, delimiter=";") writer.writerow([ "Datum", "Profil/Shop", "Absender", "Betreff", @@ -447,15 +450,17 @@ def run_cli(argv: Optional[Sequence[str]] = None) -> int: target = args.export_csv if target == "DEFAULT": target = str(DEFAULT_BASE_DIR / f"rechnungen_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv") - out_path = Path(target).resolve() + out_path = Path(target) try: + out_path = out_path.resolve() count = export_invoices_to_csv( invoices=invoices, output_path=out_path, profile_filter=args.filter_profile, status_filter=args.filter_status, + protected_paths=get_default_paths(args.config_path, args.invoices_db_path), ) - except OSError as e: + except (OSError, ValueError, TypeError, RuntimeError) as e: if args.json_output: print(json.dumps({ "status": "error", diff --git a/csv_export.py b/csv_export.py new file mode 100644 index 0000000..7ea78c0 --- /dev/null +++ b/csv_export.py @@ -0,0 +1,51 @@ +"""Private staging for CSV exports, without GUI or mail dependencies.""" + +from contextlib import contextmanager +import logging +import os +from pathlib import Path +import tempfile + + +def _check_target(target, protected_paths): + resolved = target.resolve() + for source in protected_paths: + if resolved == source.resolve(): + raise ValueError("CSV-Ziel ist eine geschützte Quelldatei. Bitte einen anderen Pfad wählen.") + try: + same = target.samefile(source) + except FileNotFoundError: + same = False + if same: + raise ValueError("CSV-Ziel ist eine geschützte Quelldatei. Bitte einen anderen Pfad wählen.") + + +@contextmanager +def atomic_csv_output(output_path, protected_paths=()): + """Publish only a complete UTF-8-BOM CSV; retain old output on failure. + + Identity checks protect known originals and state files, including aliases. + They do not provide a transaction against concurrent external path changes. + """ + target = Path(output_path) + protected = tuple(Path(path) for path in protected_paths if path) + _check_target(target, protected) + target.parent.mkdir(parents=True, exist_ok=True) + temporary = None + try: + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8-sig", newline="", delete=False, + dir=target.parent, prefix=".uim-csv-", suffix=".tmp", + ) as stream: + temporary = Path(stream.name) + yield stream + stream.flush() + os.fsync(stream.fileno()) + _check_target(target, protected) + os.replace(temporary, target) + finally: + if temporary is not None: + try: + temporary.unlink(missing_ok=True) + except OSError: + logging.warning("Eigene temporäre CSV-Datei konnte nicht entfernt werden: %s", temporary) diff --git a/tests/test_csv_export_transaction.py b/tests/test_csv_export_transaction.py new file mode 100644 index 0000000..8687269 --- /dev/null +++ b/tests/test_csv_export_transaction.py @@ -0,0 +1,139 @@ +"""Exports retain old data and never replace source/state files.""" + +import ast +import csv +from datetime import datetime +import json +import os +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import cli +import csv_export + + +@pytest.fixture +def gui_export(tmp_path): + # Execute the exact GUI method without constructing a window or touching settings. + source = Path(__file__).parents[1] / "UniversalInvoiceMail.py" + tree = ast.parse(source.read_text(encoding="utf-8-sig")) + main = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == "MainWindow") + method = next(n for n in main.body if isinstance(n, ast.FunctionDef) and n.name == "export_invoices_csv") + module = ast.Module(body=[ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0), method], type_ignores=[]) + namespace = {"Path": Path, "datetime": datetime, "CONFIG_FILE": tmp_path / "config.json", "INVOICES_DB": tmp_path / "invoices.json"} + exec(compile(ast.fix_missing_locations(module), str(source), "exec"), namespace) + return namespace["export_invoices_csv"] + + +def invoice(tmp_path, **changes): + values = dict(date="2026-10-01", profile_name="Äpfel", sender="test@example.invalid", subject="Test", filename="original.pdf", path=str(tmp_path / "original.pdf"), amount=12.5, currency="EUR", review_status="unchecked", notes="Größe; geprüft") + values.update(changes) + return values + + +@pytest.mark.parametrize("failure", ["encoding", "body", "fsync", "replace"]) +def test_old_export_survives_failure(tmp_path, monkeypatch, failure): + target = tmp_path / "old.csv" + target.write_bytes(b"previous export") + foreign = tmp_path / ".uim-csv-foreign.tmp" + foreign.write_bytes(b"foreign") + if failure in ("fsync", "replace"): + def fail(*args): + raise OSError("injected failure") + monkeypatch.setattr(csv_export.os, failure, fail) + with pytest.raises((OSError, ValueError, UnicodeError)): + with csv_export.atomic_csv_output(target) as stream: + stream.write("\ud800" if failure == "encoding" else "complete output") + if failure == "body": + raise ValueError("invalid row") + assert target.read_bytes() == b"previous export" + assert foreign.read_bytes() == b"foreign" + assert list(tmp_path.glob(".uim-csv-*.tmp")) == [foreign] + + +def test_complete_csv_preserves_format(tmp_path): + target = tmp_path / "nested" / "output.csv" + count = cli.export_invoices_to_csv([invoice(tmp_path)], target) + raw = target.read_bytes() + assert count == 1 and raw.startswith(b"\xef\xbb\xbf") + assert raw.count(b"\r\n") == 2 + rows = list(csv.reader(raw.decode("utf-8-sig").splitlines(), delimiter=";")) + assert rows[1][1] == "Äpfel" and rows[1][6] == "12,50" and rows[1][9] == "Größe; geprüft" + assert not list(target.parent.glob(".uim-csv-*.tmp")) + + +@pytest.mark.parametrize("alias", ["direct", "hardlink"]) +def test_filtered_originals_are_protected(tmp_path, alias): + original = tmp_path / "excluded.pdf" + original.write_bytes(b"source") + target = original + if alias == "hardlink": + target = tmp_path / "alias.csv" + os.link(original, target) + invoices = [invoice(tmp_path), invoice(tmp_path, path=str(original), profile_name="excluded")] + with pytest.raises(ValueError, match="geschützte"): + cli.export_invoices_to_csv(invoices, target, profile_filter="Äpfel") + assert original.read_bytes() == target.read_bytes() == b"source" + assert not list(tmp_path.glob(".uim-csv-*.tmp")) + + +def test_new_alias_before_publication_is_protected(tmp_path): + source = tmp_path / "source.pdf" + source.write_bytes(b"source") + target = tmp_path / "out.csv" + with pytest.raises(ValueError, match="geschützte"): + with csv_export.atomic_csv_output(target, [source]) as stream: + stream.write("new output") + os.link(source, target) + assert source.read_bytes() == target.read_bytes() == b"source" + assert not list(tmp_path.glob(".uim-csv-*.tmp")) + + +@pytest.mark.parametrize("kind", ["config", "database"]) +@pytest.mark.parametrize("alias", [False, True]) +def test_cli_actual_state_paths_protected(tmp_path, capsys, kind, alias): + config = tmp_path / "custom-config.json" + database = tmp_path / "custom-db.json" + config.write_text("{}", encoding="utf-8") + database.write_text(json.dumps([invoice(tmp_path)]), encoding="utf-8") + source = config if kind == "config" else database + before = source.read_bytes() + target = source + if alias: + target = tmp_path / "alias.csv" + os.link(source, target) + rc = cli.run_cli(["--config", str(config), "--invoices-db", str(database), "--export-csv", str(target), "--json"]) + assert rc == 1 and json.loads(capsys.readouterr().out)["status"] == "error" + assert source.read_bytes() == target.read_bytes() == before + + +def test_cli_encoding_error_is_structured_and_preserves_export(tmp_path, capsys): + database = tmp_path / "db.json" + database.write_text(json.dumps([invoice(tmp_path, notes="\ud800")]), encoding="utf-8") + target = tmp_path / "old.csv" + target.write_bytes(b"old") + assert cli.run_cli(["--config", str(tmp_path / "absent.json"), "--invoices-db", str(database), "--export-csv", str(target), "--json"]) == 1 + assert json.loads(capsys.readouterr().out)["status"] == "error" + assert target.read_bytes() == b"old" + + +def test_gui_late_amount_error_preserves_old_export(tmp_path, gui_export): + target = tmp_path / "old.csv" + target.write_bytes(b"old") + logs = [] + fake = SimpleNamespace(invoices=[SimpleNamespace(**invoice(tmp_path, amount="invalid"))], _get_selected_invoice_paths=lambda: set(), _log=logs.append) + assert gui_export(fake, target) is None + assert target.read_bytes() == b"old" and "[ERROR]" in logs[-1] + + +@pytest.mark.parametrize("kind", ["excluded", "config", "database"]) +def test_gui_protects_unselected_and_state_files(tmp_path, gui_export, kind): + first = SimpleNamespace(**invoice(tmp_path)) + excluded = SimpleNamespace(**invoice(tmp_path, path=str(tmp_path / "excluded.pdf"))) + target = Path(excluded.path) if kind == "excluded" else tmp_path / ("config.json" if kind == "config" else "invoices.json") + target.write_bytes(b"protected") + fake = SimpleNamespace(invoices=[first, excluded], _get_selected_invoice_paths=lambda: {first.path}, _log=lambda text: None) + assert gui_export(fake, target) is None + assert target.read_bytes() == b"protected" From 40f0a70a0a2671dd742e624a736f5acc871373b8 Mon Sep 17 00:00:00 2001 From: Lukas Geiger Date: Thu, 1 Oct 2026 16:58:52 +0200 Subject: [PATCH 2/2] fix: retain originals across modal export and repair Qt button route --- UniversalInvoiceMail.py | 11 +++++++---- cli.py | 2 +- tests/test_csv_export.py | 26 ++++++++++++++++++++++++++ tests/test_csv_export_transaction.py | 18 ++++++++++++++++-- 4 files changed, 50 insertions(+), 7 deletions(-) diff --git a/UniversalInvoiceMail.py b/UniversalInvoiceMail.py index e8d4919..30887dc 100644 --- a/UniversalInvoiceMail.py +++ b/UniversalInvoiceMail.py @@ -4003,7 +4003,7 @@ def setup_ui(self): ) btn_refresh.setToolTip("Rechnungstabelle mit Ordnerinhalt synchronisieren") btn_export_csv = QPushButton("CSV Export") - btn_export_csv.clicked.connect(self.export_invoices_csv) + btn_export_csv.clicked.connect(lambda _checked=False: self.export_invoices_csv()) btn_export_csv.setObjectName("export_invoices_csv_button") btn_export_csv.setAccessibleName("Rechnungsliste als CSV exportieren") btn_export_csv.setAccessibleDescription( @@ -5381,11 +5381,14 @@ def export_invoices_csv(self, filepath: Optional[Union[str, Path]] = None) -> Op return None # Ausgewaehlte Rechnungen ermitteln (falls Checkboxen in Spalte 0 aktiv sind) + original_invoices = list(self.invoices) + protected_paths = [CONFIG_FILE, INVOICES_DB] + protected_paths.extend(inv.path for inv in original_invoices if inv.path) selected_paths = self._get_selected_invoice_paths() if selected_paths: - source_invoices = [inv for inv in self.invoices if inv.path in selected_paths] + source_invoices = [inv for inv in original_invoices if inv.path in selected_paths] else: - source_invoices = list(self.invoices) + source_invoices = original_invoices if not source_invoices: if interactive: @@ -5408,7 +5411,7 @@ def export_invoices_csv(self, filepath: Optional[Union[str, Path]] = None) -> Op import csv from csv_export import atomic_csv_output target_path = Path(filepath) - protected_paths = [CONFIG_FILE, INVOICES_DB] + protected_paths.extend([CONFIG_FILE, INVOICES_DB]) protected_paths.extend(inv.path for inv in self.invoices if inv.path) with atomic_csv_output(target_path, protected_paths) as f: writer = csv.writer(f, delimiter=';') diff --git a/cli.py b/cli.py index 2c805c1..595abb4 100644 --- a/cli.py +++ b/cli.py @@ -460,7 +460,7 @@ def run_cli(argv: Optional[Sequence[str]] = None) -> int: status_filter=args.filter_status, protected_paths=get_default_paths(args.config_path, args.invoices_db_path), ) - except (OSError, ValueError, TypeError, RuntimeError) as e: + except (OSError, ValueError, TypeError, RuntimeError, OverflowError, csv.Error) as e: if args.json_output: print(json.dumps({ "status": "error", diff --git a/tests/test_csv_export.py b/tests/test_csv_export.py index ad0d831..44eb1c3 100644 --- a/tests/test_csv_export.py +++ b/tests/test_csv_export.py @@ -38,6 +38,32 @@ def qapp(): return app +def test_export_button_opens_dialog_without_boolean_filepath(tmp_path, monkeypatch, qapp): + """A real Qt clicked(bool) must retain the interactive save dialog.""" + from PySide6.QtCore import QCoreApplication, QEvent + from PySide6.QtWidgets import QPushButton + + monkeypatch.setattr(uim, "CONFIG_FILE", tmp_path / "config.json") + monkeypatch.setattr(uim, "INVOICES_DB", tmp_path / "invoices.json") + window = uim.MainWindow() + target = tmp_path / "clicked.csv" + calls = [] + monkeypatch.setattr(uim.QFileDialog, "getSaveFileName", lambda *args, **kwargs: (calls.append("dialog") or (str(target), "CSV"))) + monkeypatch.setattr(uim.QMessageBox, "information", lambda *args: calls.append("success")) + monkeypatch.setattr(uim.QMessageBox, "warning", lambda *args: calls.append("error")) + monkeypatch.setattr(window, "_get_selected_invoice_paths", lambda: set()) + try: + window.invoices = [uim.Invoice(id="clicked", profile_name="Prüfung", filename="dummy.pdf", date="2026-10-01", path=str(tmp_path / "dummy.pdf"), amount=12.5)] + button = window.findChild(QPushButton, "export_invoices_csv_button") + assert button is not None and button.isEnabled() + button.click() + assert calls == ["dialog", "success"] + assert target.read_bytes().startswith(b"\xef\xbb\xbf") + finally: + window.deleteLater() + QCoreApplication.sendPostedEvents(window, QEvent.Type.DeferredDelete) + + def test_export_invoices_csv_headless_all(tmp_path: Path, monkeypatch, qapp): """Test exporting all invoices programmatically with comprehensive columns.""" invoices_db = tmp_path / "invoices.json" diff --git a/tests/test_csv_export_transaction.py b/tests/test_csv_export_transaction.py index 8687269..a32463b 100644 --- a/tests/test_csv_export_transaction.py +++ b/tests/test_csv_export_transaction.py @@ -109,9 +109,10 @@ def test_cli_actual_state_paths_protected(tmp_path, capsys, kind, alias): assert source.read_bytes() == target.read_bytes() == before -def test_cli_encoding_error_is_structured_and_preserves_export(tmp_path, capsys): +@pytest.mark.parametrize("changes", [{"notes": "\ud800"}, {"amount": 10 ** 400}]) +def test_cli_format_error_is_structured_and_preserves_export(tmp_path, capsys, changes): database = tmp_path / "db.json" - database.write_text(json.dumps([invoice(tmp_path, notes="\ud800")]), encoding="utf-8") + database.write_text(json.dumps([invoice(tmp_path, **changes)]), encoding="utf-8") target = tmp_path / "old.csv" target.write_bytes(b"old") assert cli.run_cli(["--config", str(tmp_path / "absent.json"), "--invoices-db", str(database), "--export-csv", str(target), "--json"]) == 1 @@ -137,3 +138,16 @@ def test_gui_protects_unselected_and_state_files(tmp_path, gui_export, kind): fake = SimpleNamespace(invoices=[first, excluded], _get_selected_invoice_paths=lambda: {first.path}, _log=lambda text: None) assert gui_export(fake, target) is None assert target.read_bytes() == b"protected" + + +def test_gui_original_removed_during_dialog_stays_protected(tmp_path, gui_export): + target = tmp_path / "source.pdf" + target.write_bytes(b"original") + fake = SimpleNamespace(invoices=[SimpleNamespace(**invoice(tmp_path, path=str(target)))], settings=SimpleNamespace(download_path=str(tmp_path)), _get_selected_invoice_paths=lambda: set(), _log=lambda text: None) + def choose(*args): + fake.invoices = [] + return str(target), "CSV" + gui_export.__globals__["QFileDialog"] = SimpleNamespace(getSaveFileName=choose) + gui_export.__globals__["QMessageBox"] = SimpleNamespace(information=lambda *args: None, warning=lambda *args: None) + assert gui_export(fake) is None + assert target.read_bytes() == b"original"