removed with_retry

src/aare/daq/operations/common/runtime.py

@@ -41,39 +41,6 @@ class FaceDetectionProgressReporter(Protocol):
     def emit_progress(self, payload: dict) -> None: ...
 
 
-T = TypeVar("T")
-
-
-def with_retry(
-    func: Callable[[], T],
-    *,
-    max_attempts: int,
-    on: tuple[type[Exception], ...],
-    escalate: type[AareException] | None = None,
-    escalate_message: str | None = None,
-    escalate_critical: bool | None = None,
-) -> T:
-    attempts = max(1, int(max_attempts))
-    last_error: Exception | None = None
-    for _ in range(attempts):
-        try:
-            return func()
-        except on as error:
-            last_error = error
-
-    if escalate is not None:
-        kwargs: dict[str, bool] = {}
-        if escalate_critical is not None:
-            kwargs["critical"] = escalate_critical
-        message = escalate_message or str(last_error)
-        raise escalate(message, **kwargs) from last_error
-
-    if last_error is not None:
-        raise last_error
-
-    raise RuntimeError("with_retry exhausted without a captured exception")
-
-
 @dataclass
 class DAQRuntimeState:
     sample_provider: SampleProvider

tests/unit/daq/test_runtime_retry.py

@@ -1,51 +0,0 @@
-import sys
-import types
-
-import pytest
-
-
-if "jfjoch_client.models.scan_result" not in sys.modules:
-    jfjoch_client_mod = types.ModuleType("jfjoch_client")
-    jfjoch_client_models_mod = types.ModuleType("jfjoch_client.models")
-    jfjoch_client_scan_result_mod = types.ModuleType("jfjoch_client.models.scan_result")
-
-    jfjoch_client_scan_result_mod.ScanResult = dict
-    jfjoch_client_models_mod.scan_result = jfjoch_client_scan_result_mod
-    jfjoch_client_mod.models = jfjoch_client_models_mod
-    sys.modules["jfjoch_client"] = jfjoch_client_mod
-    sys.modules["jfjoch_client.models"] = jfjoch_client_models_mod
-    sys.modules["jfjoch_client.models.scan_result"] = jfjoch_client_scan_result_mod
-
-from aare.common.exception_handler import MountingFailed
-from aare.daq.operations.common.runtime import with_retry
-
-
-def test_with_retry_succeeds_on_later_attempt():
-    calls = {"count": 0}
-
-    def flaky_call() -> str:
-        calls["count"] += 1
-        if calls["count"] < 3:
-            raise MountingFailed("temporary")
-        return "ok"
-
-    assert with_retry(flaky_call, max_attempts=5, on=(MountingFailed,)) == "ok"
-    assert calls["count"] == 3
-
-
-def test_with_retry_escalates_on_exhaustion():
-    def always_fail() -> None:
-        raise MountingFailed("still failing")
-
-    with pytest.raises(MountingFailed) as exc:
-        with_retry(
-            always_fail,
-            max_attempts=2,
-            on=(MountingFailed,),
-            escalate=MountingFailed,
-            escalate_message="Mount failed 2 times in a row, stopping automation.",
-            escalate_critical=True,
-        )
-
-    assert str(exc.value) == "Mount failed 2 times in a row, stopping automation."
-    assert exc.value.critical is True