Build: ein Dateileser mit Wiederholung und contextlib

from collections.abc import Iterator
from contextlib import contextmanager


@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
    log.append(f"attempt {number}: start")
    yield
    log.append(f"attempt {number}: end")

from main import attempt


def test_success():
    """Ein funktionierender Block protokolliert start und end"""
    log = []
    with attempt(log, 1):
        pass
    assert log == ["attempt 1: start", "attempt 1: end"], f"das Protokoll ist {log!r}"


def test_os_error_logged_and_raised():
    """Ein OSError wird mit seinem Typ protokolliert und erneut ausgelöst"""
    log = []
    try:
        with attempt(log, 2):
            open("no_such_file.txt", encoding="utf-8")
    except FileNotFoundError:
        pass
    else:
        assert False, "der FileNotFoundError kam nicht beim Aufrufer an; lösen Sie ihn erneut aus"
    assert log == ["attempt 2: start", "attempt 2: FileNotFoundError", "attempt 2: end"], f"das Protokoll ist {log!r}"


def test_other_error_passes():
    """Andere Fehler gelten nicht als Fehlschlag, beenden den Versuch aber trotzdem"""
    log = []
    try:
        with attempt(log, 3):
            int("x")
    except ValueError:
        pass
    else:
        assert False, "der ValueError kam nicht beim Aufrufer an"
    assert log == ["attempt 3: start", "attempt 3: end"], f"das Protokoll ist {log!r}"

from collections.abc import Callable, Iterator
from contextlib import contextmanager


@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
    log.append(f"attempt {number}: start")
    try:
        yield
    except OSError as err:
        log.append(f"attempt {number}: {type(err).__name__}")
        raise
    finally:
        log.append(f"attempt {number}: end")


class ReadFailedError(Exception):
    """The file could not be read in any attempt."""


def read_with_retry(
    path: str,
    attempts: int = 3,
    log: list[str] | None = None,
    wait: Callable[[int], None] | None = None,
) -> str:
    with open(path, encoding="utf-8") as f:
        return f.read()

import os

from main import ReadFailedError, read_with_retry


def test_reads_file():
    """Eine vorhandene Datei wird beim ersten Versuch gelesen"""
    with open("ready.txt", "w", encoding="utf-8") as f:
        f.write("hello")
    log = []
    got = read_with_retry("ready.txt", log=log)
    assert got == "hello", f"read_with_retry lieferte {got!r}"
    assert log == ["attempt 1: start", "attempt 1: end"], f"das Protokoll ist {log!r}"


def test_retries_until_file_appears():
    """Die Datei entsteht während der zweiten Wartezeit, Versuch 3 liest sie"""
    if os.path.exists("late.txt"):
        os.remove("late.txt")
    waits = []

    def wait(n):
        waits.append(n)
        if n == 2:
            with open("late.txt", "w", encoding="utf-8") as f:
                f.write("ready")

    log = []
    got = read_with_retry("late.txt", log=log, wait=wait)
    assert got == "ready", f"read_with_retry lieferte {got!r}"
    assert waits == [1, 2], f"wait wurde aufgerufen mit {waits!r}"
    assert log[-2:] == ["attempt 3: start", "attempt 3: end"], f"das Protokoll ist {log!r}"
    assert log.count("attempt 1: FileNotFoundError") == 1, f"das Protokoll ist {log!r}"


def test_gives_up():
    """Nach dem letzten Versuch: ReadFailedError aus einer Gruppe aller Fehler"""
    waits = []
    try:
        read_with_retry("never.txt", attempts=2, wait=waits.append)
    except ReadFailedError as err:
        assert str(err) == "could not read never.txt after 2 attempts", f"die Meldung ist {str(err)!r}"
        group = err.__cause__
        assert isinstance(group, ExceptionGroup), f"die Ursache ist {group!r}, erwartet: eine ExceptionGroup"
        notes = [getattr(e, "__notes__", []) for e in group.exceptions]
        assert notes == [["attempt 1"], ["attempt 2"]], f"die Notizen sind {notes!r}"
    else:
        assert False, "read_with_retry kehrte bei fehlender Datei normal zurück"
    assert waits == [1], f"wait wurde aufgerufen mit {waits!r}; nach dem letzten Versuch wird nicht gewartet"

import os
from collections.abc import Callable, Iterator
from contextlib import contextmanager, suppress


@contextmanager
def attempt(log: list[str], number: int) -> Iterator[None]:
    log.append(f"attempt {number}: start")
    try:
        yield
    except OSError as err:
        log.append(f"attempt {number}: {type(err).__name__}")
        raise
    finally:
        log.append(f"attempt {number}: end")


class ReadFailedError(Exception):
    """The file could not be read in any attempt."""


def read_with_retry(
    path: str,
    attempts: int = 3,
    log: list[str] | None = None,
    wait: Callable[[int], None] | None = None,
) -> str:
    if log is None:
        log = []
    errors: list[OSError] = []
    for n in range(1, attempts + 1):
        try:
            with attempt(log, n):
                with open(path, encoding="utf-8") as f:
                    return f.read()
        except OSError as err:
            err.add_note(f"attempt {n}")
            errors.append(err)
            if wait is not None and n < attempts:
                wait(n)
    raise ReadFailedError(f"could not read {path} after {attempts} attempts") from ExceptionGroup(
        "read errors", errors
    )


def read_locked(path: str, attempts: int = 3, wait: Callable[[int], None] | None = None) -> str:
    lock_path = path + ".lock"
    with open(lock_path, "x", encoding="utf-8"):
        pass
    text = read_with_retry(path, attempts, wait=wait)
    os.remove(lock_path)
    return text

import os

from main import ReadFailedError, read_locked


def clean(*names):
    for name in names:
        if os.path.exists(name):
            os.remove(name)


def test_success_removes_lock():
    """Nach erfolgreichem Lesen ist die Sperrdatei weg"""
    clean("a.txt.lock")
    with open("a.txt", "w", encoding="utf-8") as f:
        f.write("A")
    got = read_locked("a.txt")
    assert got == "A", f"read_locked lieferte {got!r}"
    assert not os.path.exists("a.txt.lock"), "a.txt.lock existiert nach erfolgreichem Lesen noch"


def test_failure_removes_lock():
    """Auch nach gescheitertem Lesen ist die Sperrdatei weg"""
    clean("b.txt", "b.txt.lock")
    try:
        read_locked("b.txt", attempts=2)
    except ReadFailedError:
        pass
    else:
        assert False, "read_locked kehrte bei fehlender Datei normal zurück"
    assert not os.path.exists("b.txt.lock"), "b.txt.lock existiert nach gescheitertem Lesen noch; entfernen Sie sie in finally"


def test_existing_lock_is_kept():
    """Existiert die Sperre, scheitert read_locked und lässt sie stehen"""
    with open("c.txt.lock", "w", encoding="utf-8") as f:
        f.write("held")
    try:
        read_locked("c.txt")
    except FileExistsError:
        pass
    else:
        assert False, "read_locked löste bei gehaltener Sperre keinen FileExistsError aus"
    assert os.path.exists("c.txt.lock"), "read_locked hat eine Sperre entfernt, die es nicht angelegt hat"
    clean("c.txt.lock")


def test_lock_already_gone():
    """Eine Sperre, die jemand anderes während des Lesens entfernt, ist kein Fehler"""
    clean("d.txt", "d.txt.lock")

    def wait(n):
        clean("d.txt.lock")
        with open("d.txt", "w", encoding="utf-8") as f:
            f.write("D")

    got = read_locked("d.txt", wait=wait)
    assert got == "D", f"read_locked lieferte {got!r}"