From d2a473f79ec16e5b518df9f4078e721d9359aa4d Mon Sep 17 00:00:00 2001 From: Lukas Geiger Date: Thu, 1 Oct 2026 23:00:35 +0200 Subject: [PATCH 1/3] fix: reserve clipboard outputs exclusively before publication --- CHANGELOG.md | 6 + absorber/core.py | 114 +++++++++++-- docs/OUTPUT_SAFETY.md | 39 +++++ tests/test_output_reservation.py | 280 +++++++++++++++++++++++++++++++ 4 files changed, 426 insertions(+), 13 deletions(-) create mode 100644 docs/OUTPUT_SAFETY.md create mode 100644 tests/test_output_reservation.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 26c1b17..1a8c28b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,12 @@ die Versionierung [Semantic Versioning](https://semver.org/lang/de/). ### Behoben +- Gleichzeitige Speicherungen konnten denselben vermeintlich freien Dateinamen + wählen und sich überschreiben. Text, Markdown, HTML, Bilder und Dateikopien + werden jetzt zunächst privat geschrieben und anschließend unter einem exklusiv + reservierten Namen veröffentlicht. Fehler bereinigen nur Dateien mit eigener + nachgewiesener Identität; vorhandene Dateien erhalten weiterhin `_2`, `_3` usw. + - Der `QShortcut` für Strg+V wurde mit einer `StandardKey` statt einer `QKeySequence` und einem `activated=`-Schlüsselwort gebaut. Zur Laufzeit funktionierte das, es traf aber keine dokumentierte Überladung — eine Form, diff --git a/absorber/core.py b/absorber/core.py index e4c4db1..6b971ad 100644 --- a/absorber/core.py +++ b/absorber/core.py @@ -7,11 +7,16 @@ from __future__ import annotations +import os import shutil +import stat +import sys +import tempfile +from collections.abc import Callable from pathlib import Path from . import config as config_module -from .naming import build_stem, unique_path +from .naming import build_stem #: Content kinds, in the order they are probed. KIND_FILES = "files" @@ -56,20 +61,18 @@ def materialize(mime, cfg: dict, namer=None) -> list[Path]: stem = build_stem(kind, sample, namer) if kind == KIND_IMAGE: - path = unique_path(target, stem, "png") image = mime.imageData() - if not image.save(str(path), "PNG"): - raise OSError(f"could not write image to {path}") - return [path] + def write_image(stage: Path) -> None: + if not image.save(str(stage), "PNG"): + raise OSError("Bild konnte nicht vollständig gespeichert werden.") + return [_write_unique(target, stem, "png", write_image)] if kind == KIND_HTML: - path = unique_path(target, stem, "html") - path.write_text(mime.html(), encoding="utf-8") - return [path] + return [_write_unique(target, stem, "html", + lambda stage: stage.write_text(mime.html(), encoding="utf-8"))] - path = unique_path(target, stem, config_module.text_extension(cfg)) - path.write_text(mime.text(), encoding="utf-8") - return [path] + return [_write_unique(target, stem, config_module.text_extension(cfg), + lambda stage: stage.write_text(mime.text(), encoding="utf-8"))] def _copy_files(mime, target: Path) -> list[Path]: @@ -81,12 +84,97 @@ def _copy_files(mime, target: Path) -> list[Path]: source_path = Path(source) if not source_path.is_file(): continue # directories are out of scope for the MVP - destination = unique_path(target, source_path.stem, source_path.suffix) - shutil.copy2(source_path, destination) + destination = _write_unique(target, source_path.stem, source_path.suffix, + lambda stage, source=source_path: shutil.copy2(source, stage)) written.append(destination) return written +def _owned(path: Path, identity: os.stat_result) -> bool: + """Check the directory entry itself, without following a replacement symlink.""" + current = path.lstat() + return stat.S_ISREG(current.st_mode) and os.path.samestat(current, identity) + + +def _remove_owned(path: Path, identity: os.stat_result) -> None: + try: + if not _owned(path, identity): + return + try: + path.unlink() + except PermissionError: + # copy2 may copy a readonly attribute to our stage, never to its source. + if _owned(path, identity): + path.chmod(stat.S_IREAD | stat.S_IWRITE) + if _owned(path, identity): + path.unlink() + except FileNotFoundError: + pass + + +def _write_unique(directory: Path, stem: str, extension: str, + write: Callable[[Path], object]) -> Path: + """Write privately, then replace only an exclusively reserved output name.""" + extension = extension.lstrip(".") + name = f"{stem}.{extension}" + if not stem or stem in (".", "..") or any(char in name for char in ("/", "\\")): + raise ValueError("Ausgabename muss ein einzelner Dateiname sein.") + if os.name == "nt" and ":" in name: + raise ValueError("Ausgabename darf keinen alternativen Datenstrom enthalten.") + stage = reservation = None + stage_identity = reservation_identity = None + try: + fd, filename = tempfile.mkstemp(prefix=".absorber-", suffix=".tmp", dir=directory) + stage = Path(filename) + try: + stage_identity = os.fstat(fd) + finally: + os.close(fd) + write(stage) + counter = 1 + while True: + suffix = "" if counter == 1 else f"_{counter}" + candidate = directory / f"{stem}{suffix}.{extension}" + try: + fd = os.open(candidate, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + counter += 1 + continue + except PermissionError as error: + # Windows reports EACCES for an existing directory under O_EXCL. + # Advance only if a real directory entry can be confirmed. + try: + candidate.lstat() + except OSError: + raise error + counter += 1 + continue + reservation = candidate + try: + reservation_identity = os.fstat(fd) + finally: + os.close(fd) + break + if not _owned(reservation, reservation_identity) or not _owned(stage, stage_identity): + raise OSError("Reservierte Ausgabedatei wurde während des Schreibens ersetzt.") + os.replace(stage, reservation) + result = reservation + reservation = None + return result + finally: + for path, identity in ((stage, stage_identity), (reservation, reservation_identity)): + if path is None or identity is None: + continue + try: + _remove_owned(path, identity) + except OSError: + # Keep the original error; never fall back to deleting unknown files. + try: + print("Temporäre Absorber-Datei konnte nicht bereinigt werden.", file=sys.stderr) + except Exception: # noqa: BLE001, S110 -- Preserve the original write error. + pass + + def make_namer(cfg: dict): """An OllamaNamer when model naming is switched on, else None.""" if not cfg.get("model_naming"): diff --git a/docs/OUTPUT_SAFETY.md b/docs/OUTPUT_SAFETY.md new file mode 100644 index 0000000..089b519 --- /dev/null +++ b/docs/OUTPUT_SAFETY.md @@ -0,0 +1,39 @@ +# Sichere Ausgabe bei gleichzeitigen Speicherungen + +Absorber schreibt Text, Markdown, HTML, Bilder und Dateikopien zunächst in eine +exklusiv angelegte temporäre Datei im Zielordner. Erst nach erfolgreichem Schreiben +reserviert es den endgültigen Namen mit `O_CREAT | O_EXCL`. Ist der Name schon +belegt, folgen wie bisher `_2`, `_3` usw. Auch ein defekter symbolischer Link zählt +als belegter Name. Die bisherige Funktion `naming.unique_path` bleibt eine reine +Namensvorhersage; sie reserviert keine Datei. + +Vor der Veröffentlichung prüft Absorber, ob temporäre Datei und Reservierung noch +die beim Anlegen über den eigenen Dateideskriptor erfasste Identität besitzen. +Die Prüfung verwendet `lstat` und folgt keinem nachträglich eingesetzten Link. +Anschließend ersetzt `os.replace` ausschließlich die eigene Reservierung. Erst +danach wird der Pfad als Ergebnis zurückgegeben. + +Bei einem Fehler versucht Absorber, die eigene temporäre Datei und eigene +Reservierung zu entfernen. Eine fremd ersetzte Datei bleibt erhalten. Wenn eine +Dateikopie ein Schreibschutzattribut übernimmt, wird dieses ausschließlich zur +Bereinigung der eigenen temporären Datei angepasst. Die Quelle wird nicht +verändert. Scheitert die Bereinigung oder lässt sich die Identität nicht sicher +feststellen, können eigene temporäre Dateien zurückbleiben; unbekannte Dateien +werden nicht als Ersatz gelöscht. Ein Bereinigungsfehler verdeckt keinen bereits +aufgetretenen Schreibfehler. + +Eine Dateiliste wird wie bisher einzeln verarbeitet. Scheitert eine spätere Kopie, +bleiben bereits erfolgreich veröffentlichte frühere Kopien erhalten. Kopierte +Metadaten und Änderungszeiten werden weiterhin von `copy2` übernommen. + +Während des kurzen Abschnitts zwischen Reservierung und Ersetzen kann ein leerer +Platzhalter sichtbar sein. Ein gewaltsamer Prozessabbruch kann Platzhalter oder +temporäre Dateien hinterlassen. Die Identitätsprüfung ist keine Sperre gegen eine +externe Änderung zwischen letzter Prüfung und Ersetzen. Es gibt keine zusätzliche +Garantie bei Stromausfall. + +Die Regressionen nutzen synthetische Inhalte, tatsächliche parallele Threads und +eine native Qt-Bildkodierung. Die GUI-Suite läuft mit der Qt-Plattform `offscreen`; +sie ist keine Abnahme der echten Windows-Zwischenablage oder des System-Trays. +Konfigurationsdatei, Recent-Liste und virtuelle Aufbewahrung haben getrennte +Lebenszyklen und sind nicht Bestandteil dieses Ausgabefixes. diff --git a/tests/test_output_reservation.py b/tests/test_output_reservation.py new file mode 100644 index 0000000..47fa9ef --- /dev/null +++ b/tests/test_output_reservation.py @@ -0,0 +1,280 @@ +"""Synthetic output-ownership regressions; no real clipboard or model access.""" +import os +import shutil +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest +from conftest import FakeImage, FakeMime + +from absorber import core + + +def make_mime(kind, text, source): + if kind == "image": + return FakeMime(image=FakeImage(text.encode())) + if kind == "files": + return FakeMime(urls=[str(source)]) + if kind == "html": + return FakeMime(text=text, html=text) + return FakeMime(text=text) + + +@pytest.mark.parametrize("kind", ["text", "markdown", "html", "image", "files"]) +def test_parallel_materializations_keep_both_payloads(cfg, tmp_path, monkeypatch, kind): + target = Path(cfg["target_dir"]) + cfg.update(copy_file_lists=True, text_formats=["html"] if kind == "html" else ["md"] if kind == "markdown" else ["txt"]) + sources = [] + for label in ("A", "B"): + source = tmp_path / label / "same.dat" + source.parent.mkdir() + source.write_bytes(("synthetic-" + label).encode()) + sources.append(source) + monkeypatch.setattr(core, "build_stem", lambda *a: "same") + barrier = threading.Barrier(2) + original_text, original_bytes, original_copy = Path.write_text, Path.write_bytes, shutil.copy2 + + def text_write(path, value, *a, **k): + if path.parent == target and value.startswith("synthetic-"): + barrier.wait(timeout=10) + return original_text(path, value, *a, **k) + + def bytes_write(path, value, *a, **k): + if path.parent == target and value.startswith(b"synthetic-"): + barrier.wait(timeout=10) + return original_bytes(path, value, *a, **k) + + def copy(source, destination, *a, **k): + barrier.wait(timeout=10) + return original_copy(source, destination, *a, **k) + + monkeypatch.setattr(Path, "write_text", text_write) + monkeypatch.setattr(Path, "write_bytes", bytes_write) + monkeypatch.setattr(core.shutil, "copy2", copy) + with ThreadPoolExecutor(max_workers=2) as pool: + returned = list(pool.map(lambda pair: core.materialize(make_mime(kind, "synthetic-" + pair[0], pair[1]), cfg), zip(("A", "B"), sources))) + paths = [result[0] for result in returned] + assert paths[0] != paths[1] + assert {path.read_bytes() for path in paths} == {b"synthetic-A", b"synthetic-B"} + assert set(target.iterdir()) == set(paths) + assert [source.read_bytes() for source in sources] == [b"synthetic-A", b"synthetic-B"] + + +@pytest.mark.parametrize("kind", ["text", "html", "image", "files"]) +def test_failed_write_leaves_no_partial_output(cfg, tmp_path, monkeypatch, kind): + target = Path(cfg["target_dir"]) + cfg.update(copy_file_lists=True, text_formats=["html", "txt"]) + source = tmp_path / "source.dat" + source.write_bytes(b"original-source") + original_write = Path.write_text + error = OSError("synthetic partial write") + + def partial(path, value, *a, **k): + original_write(path, "partial", encoding="utf-8") + raise error + + def partial_copy(source, destination, *a, **k): + Path(destination).write_bytes(b"partial") + raise error + + class BadImage: + def save(self, path, fmt): + Path(path).write_bytes(b"partial") + return False + + monkeypatch.setattr(Path, "write_text", partial) + monkeypatch.setattr(core.shutil, "copy2", partial_copy) + mime = FakeMime(image=BadImage()) if kind == "image" else make_mime(kind, "synthetic-text", source) + with pytest.raises(OSError) as caught: + core.materialize(mime, cfg) + if kind != "image": + assert caught.value is error + assert list(target.iterdir()) == [] + assert source.read_bytes() == b"original-source" + + +def test_real_qimage_output_is_readable_and_private_stage_removed(cfg): + from PySide6.QtGui import QColor, QImage + + image = QImage(2, 2, QImage.Format.Format_ARGB32) + image.fill(QColor("red")) + result = core.materialize(FakeMime(image=image), cfg)[0] + loaded = QImage(str(result)) + assert loaded.size() == image.size() + assert loaded.pixelColor(0, 0) == QColor("red") + assert list(result.parent.iterdir()) == [result] + + +def test_dangling_symlink_counts_as_collision(cfg, tmp_path, monkeypatch): + target = Path(cfg["target_dir"]) + target.mkdir() + original = target / "same.txt" + missing = tmp_path / "unrelated-missing.txt" + try: + original.symlink_to(missing) + except OSError as error: + if getattr(error, "winerror", None) == 1314: + pytest.skip("Windows symlink privilege unavailable (1314)") + raise + monkeypatch.setattr(core, "build_stem", lambda *a: "same") + result = core.materialize(FakeMime(text="synthetic"), cfg)[0] + assert result.name == "same_2.txt" + assert original.is_symlink() + assert not missing.exists() + + +def test_same_folder_file_copy_preserves_original(cfg, monkeypatch): + target = Path(cfg["target_dir"]) + target.mkdir() + source = target / "source.txt" + source.write_bytes(b"source") + cfg["copy_file_lists"] = True + result = core.materialize(FakeMime(urls=[str(source)]), cfg)[0] + assert result.name == "source_2.txt" + assert source.read_bytes() == result.read_bytes() == b"source" + + +@pytest.mark.parametrize("readonly", [False, True]) +def test_publication_error_preserves_source_and_cleans_owned_files(cfg, tmp_path, monkeypatch, readonly): + source = tmp_path / "source.txt" + source.write_bytes(b"original-source") + cfg["copy_file_lists"] = True + error = PermissionError("synthetic publication failure") + if readonly: + source.chmod(0o444) + original_mode = source.stat().st_mode + def fail(*args): + raise error + monkeypatch.setattr(core.os, "replace", fail) + try: + with pytest.raises(PermissionError) as caught: + core.materialize(FakeMime(urls=[str(source)]), cfg) + assert caught.value is error + assert source.read_bytes() == b"original-source" + assert source.stat().st_mode == original_mode + assert list(Path(cfg["target_dir"]).iterdir()) == [] + finally: + source.chmod(0o600) + + +def test_foreign_replacement_of_reservation_is_preserved(cfg, tmp_path, monkeypatch): + target = Path(cfg["target_dir"]) + original = tmp_path / "foreign-original.bin" + original.write_bytes(b"foreign-original") + expected = target / "reserved.txt" + original_open, original_close = os.open, os.close + descriptor = set() + def opened(path, flags, *args, **kwargs): + fd = original_open(path, flags, *args, **kwargs) + if Path(path) == expected and flags & os.O_EXCL: + descriptor.add(fd) + return fd + def closed(fd): + original_close(fd) + if fd in descriptor: + descriptor.remove(fd) + expected.unlink() + os.link(original, expected) + monkeypatch.setattr(core, "build_stem", lambda *a: "reserved") + monkeypatch.setattr(core.os, "open", opened) + monkeypatch.setattr(core.os, "close", closed) + with pytest.raises(OSError): + core.materialize(FakeMime(text="synthetic"), cfg) + assert list(target.iterdir()) == [expected] + assert os.path.samefile(expected, original) + assert original.read_bytes() == expected.read_bytes() == b"foreign-original" + + +def test_partial_file_batch_keeps_completed_files_and_sources(cfg, tmp_path, monkeypatch): + first, second = tmp_path / "first.txt", tmp_path / "second.txt" + first.write_bytes(b"first-original") + second.write_bytes(b"second-original") + cfg["copy_file_lists"] = True + original_copy = shutil.copy2 + error = OSError("second copy failed") + def copy(source, stage, *args, **kwargs): + if Path(source) == second: + Path(stage).write_bytes(b"partial") + raise error + return original_copy(source, stage, *args, **kwargs) + monkeypatch.setattr(core.shutil, "copy2", copy) + with pytest.raises(OSError) as caught: + core.materialize(FakeMime(urls=[str(first), str(second)]), cfg) + assert caught.value is error + target = Path(cfg["target_dir"]) + assert list(target.iterdir()) == [target / "first.txt"] + assert (target / "first.txt").read_bytes() == first.read_bytes() == b"first-original" + assert second.read_bytes() == b"second-original" + + +def test_preexisting_files_and_foreign_temporary_file_untouched(cfg, monkeypatch): + target = Path(cfg["target_dir"]) + target.mkdir() + existing = {"same.txt": b"old", "same_2.txt": b"old-two", ".absorber-foreign.tmp": b"foreign"} + for name, value in existing.items(): + (target / name).write_bytes(value) + monkeypatch.setattr(core, "build_stem", lambda *a: "same") + result = core.materialize(FakeMime(text="synthetic"), cfg)[0] + assert result.name == "same_3.txt" + assert {name: (target / name).read_bytes() for name in existing} == existing + assert {p.name for p in target.iterdir()} == {*existing, result.name} + + +@pytest.mark.parametrize("kind", ["directory", "readonly-file"]) +def test_preexisting_directory_or_readonly_file_is_a_collision(cfg, monkeypatch, kind): + target = Path(cfg["target_dir"]) + target.mkdir() + original = target / "same.txt" + if kind == "directory": + original.mkdir() + else: + original.write_bytes(b"readonly-original") + original.chmod(0o444) + monkeypatch.setattr(core, "build_stem", lambda *a: "same") + try: + result = core.materialize(FakeMime(text="synthetic"), cfg)[0] + assert result.name == "same_2.txt" + if kind == "directory": + assert original.is_dir() + else: + assert original.read_bytes() == b"readonly-original" + finally: + if kind == "readonly-file": + original.chmod(0o600) + + +def test_cleanup_and_error_reporting_failure_do_not_mask_write_error(cfg, monkeypatch): + import sys + + primary = PermissionError("primary synthetic error") + def publication_failure(*args): + raise primary + def cleanup_failure(*args): + raise OSError("secondary synthetic cleanup error") + class FailedStderr: + def write(self, text): + raise RuntimeError("secondary reporting error") + with monkeypatch.context() as patch: + patch.setattr(core.os, "replace", publication_failure) + patch.setattr(core, "_remove_owned", cleanup_failure) + patch.setattr(sys, "stderr", FailedStderr()) + with pytest.raises(PermissionError) as caught: + core.materialize(FakeMime(text="synthetic"), cfg) + assert caught.value is primary + + +def test_reservation_permission_denial_is_not_assumed_collision(cfg, monkeypatch): + target = Path(cfg["target_dir"]) + original_open = os.open + error = PermissionError("reservation forbidden") + monkeypatch.setattr(core, "build_stem", lambda *a: "same") + def opened(path, *args, **kwargs): + if Path(path) == target / "same.txt": + raise error + return original_open(path, *args, **kwargs) + monkeypatch.setattr(core.os, "open", opened) + with pytest.raises(PermissionError) as caught: + core.materialize(FakeMime(text="synthetic"), cfg) + assert caught.value is error + assert list(target.iterdir()) == [] From 3a45f6b8ae87eb937ecc6ed640401275fb68537b Mon Sep 17 00:00:00 2001 From: Lukas Geiger Date: Thu, 1 Oct 2026 23:09:07 +0200 Subject: [PATCH 2/3] fix: reserve Windows outputs without following reparse points --- absorber/core.py | 31 +++++++++++++++++++++- docs/OUTPUT_SAFETY.md | 9 ++++++- tests/test_output_reservation.py | 44 +++++++++++++++++++++++++++----- 3 files changed, 75 insertions(+), 9 deletions(-) diff --git a/absorber/core.py b/absorber/core.py index 6b971ad..05198fc 100644 --- a/absorber/core.py +++ b/absorber/core.py @@ -112,6 +112,35 @@ def _remove_owned(path: Path, identity: os.stat_result) -> None: pass +def _exclusive_file(path: Path) -> int: + if os.name != "nt": + return os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + # The Windows CRT follows dangling symlinks even with O_EXCL. CREATE_NEW + # together with OPEN_REPARSE_POINT reserves the directory entry itself. + import ctypes + import msvcrt + from ctypes import wintypes + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + create = kernel.CreateFileW + create.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, + wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE] + create.restype = wintypes.HANDLE + handle = create(os.path.abspath(path), 0x40000000, 0x7, None, 1, + 0x80 | 0x00200000, None) + if handle == ctypes.c_void_p(-1).value: + raise ctypes.WinError(ctypes.get_last_error()) + try: + # A successful conversion transfers handle ownership to the descriptor. + return msvcrt.open_osfhandle(handle, os.O_WRONLY | os.O_BINARY) + except BaseException: + close = kernel.CloseHandle + close.argtypes = [wintypes.HANDLE] + close.restype = wintypes.BOOL + close(handle) + raise + + def _write_unique(directory: Path, stem: str, extension: str, write: Callable[[Path], object]) -> Path: """Write privately, then replace only an exclusively reserved output name.""" @@ -136,7 +165,7 @@ def _write_unique(directory: Path, stem: str, extension: str, suffix = "" if counter == 1 else f"_{counter}" candidate = directory / f"{stem}{suffix}.{extension}" try: - fd = os.open(candidate, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + fd = _exclusive_file(candidate) except FileExistsError: counter += 1 continue diff --git a/docs/OUTPUT_SAFETY.md b/docs/OUTPUT_SAFETY.md index 089b519..df38122 100644 --- a/docs/OUTPUT_SAFETY.md +++ b/docs/OUTPUT_SAFETY.md @@ -2,11 +2,18 @@ Absorber schreibt Text, Markdown, HTML, Bilder und Dateikopien zunächst in eine exklusiv angelegte temporäre Datei im Zielordner. Erst nach erfolgreichem Schreiben -reserviert es den endgültigen Namen mit `O_CREAT | O_EXCL`. Ist der Name schon +reserviert es den endgültigen Namen exklusiv. Unter Unix geschieht dies mit +`O_CREAT | O_EXCL`. Ist der Name schon belegt, folgen wie bisher `_2`, `_3` usw. Auch ein defekter symbolischer Link zählt als belegter Name. Die bisherige Funktion `naming.unique_path` bleibt eine reine Namensvorhersage; sie reserviert keine Datei. +Unter Windows verwendet die Reservierung `CreateFileW` mit `CREATE_NEW` und +`FILE_FLAG_OPEN_REPARSE_POINT`, da die Windows-CRT bei `O_EXCL` defekten Links +folgen kann. Das eigene Windows-Handle wird nach erfolgreichem Anlegen in einen +Python-Dateideskriptor überführt; schlägt diese Überführung fehl, wird das Handle +geschlossen. Die bisherigen Identitäts- und Bereinigungsprüfungen gelten auch hier. + Vor der Veröffentlichung prüft Absorber, ob temporäre Datei und Reservierung noch die beim Anlegen über den eigenen Dateideskriptor erfasste Identität besitzen. Die Prüfung verwendet `lstat` und folgt keinem nachträglich eingesetzten Link. diff --git a/tests/test_output_reservation.py b/tests/test_output_reservation.py index 47fa9ef..1c7d4bf 100644 --- a/tests/test_output_reservation.py +++ b/tests/test_output_reservation.py @@ -163,11 +163,11 @@ def test_foreign_replacement_of_reservation_is_preserved(cfg, tmp_path, monkeypa original = tmp_path / "foreign-original.bin" original.write_bytes(b"foreign-original") expected = target / "reserved.txt" - original_open, original_close = os.open, os.close + original_open, original_close = core._exclusive_file, os.close descriptor = set() - def opened(path, flags, *args, **kwargs): - fd = original_open(path, flags, *args, **kwargs) - if Path(path) == expected and flags & os.O_EXCL: + def opened(path): + fd = original_open(path) + if Path(path) == expected: descriptor.add(fd) return fd def closed(fd): @@ -177,7 +177,7 @@ def closed(fd): expected.unlink() os.link(original, expected) monkeypatch.setattr(core, "build_stem", lambda *a: "reserved") - monkeypatch.setattr(core.os, "open", opened) + monkeypatch.setattr(core, "_exclusive_file", opened) monkeypatch.setattr(core.os, "close", closed) with pytest.raises(OSError): core.materialize(FakeMime(text="synthetic"), cfg) @@ -266,15 +266,45 @@ def write(self, text): def test_reservation_permission_denial_is_not_assumed_collision(cfg, monkeypatch): target = Path(cfg["target_dir"]) - original_open = os.open + original_open = core._exclusive_file error = PermissionError("reservation forbidden") monkeypatch.setattr(core, "build_stem", lambda *a: "same") def opened(path, *args, **kwargs): if Path(path) == target / "same.txt": raise error return original_open(path, *args, **kwargs) - monkeypatch.setattr(core.os, "open", opened) + monkeypatch.setattr(core, "_exclusive_file", opened) with pytest.raises(PermissionError) as caught: core.materialize(FakeMime(text="synthetic"), cfg) assert caught.value is error assert list(target.iterdir()) == [] + + +@pytest.mark.skipif(os.name != "nt", reason="Native Windows handle transfer") +def test_failed_native_descriptor_transfer_closes_own_handle(tmp_path, monkeypatch): + import ctypes + import msvcrt + from ctypes import wintypes + from types import SimpleNamespace + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.CloseHandle.argtypes = [wintypes.HANDLE] + kernel.CloseHandle.restype = wintypes.BOOL + closed, transferred = [], [] + error = OSError("synthetic descriptor transfer error") + def close(handle): + result = kernel.CloseHandle(handle) + closed.append((handle, bool(result))) + return result + def fail_transfer(handle, flags): + transferred.append(handle) + raise error + wrapper = SimpleNamespace(CreateFileW=kernel.CreateFileW, CloseHandle=close) + with monkeypatch.context() as patch: + patch.setattr(ctypes, "WinDLL", lambda *a, **k: wrapper) + patch.setattr(msvcrt, "open_osfhandle", fail_transfer) + with pytest.raises(OSError) as caught: + core._exclusive_file(tmp_path / "own-reservation.txt") + assert caught.value is error + assert len(transferred) == 1 + assert closed == [(transferred[0], True)] From d3136d5f510666b925bb1461bf1dbe64fd9c4ad9 Mon Sep 17 00:00:00 2001 From: Lukas Geiger Date: Thu, 1 Oct 2026 23:11:31 +0200 Subject: [PATCH 3/3] test: narrow Windows handle probe for Linux type checking --- tests/test_output_reservation.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/test_output_reservation.py b/tests/test_output_reservation.py index 1c7d4bf..75944fe 100644 --- a/tests/test_output_reservation.py +++ b/tests/test_output_reservation.py @@ -282,6 +282,8 @@ def opened(path, *args, **kwargs): @pytest.mark.skipif(os.name != "nt", reason="Native Windows handle transfer") def test_failed_native_descriptor_transfer_closes_own_handle(tmp_path, monkeypatch): + if os.name != "nt": + pytest.skip("Native Windows handle transfer") import ctypes import msvcrt from ctypes import wintypes