52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
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
|