implement retry ourselves, don't depend on tenacity

This commit is contained in:
Jean-Paul Calderone 2023-03-17 15:30:14 -04:00
parent 8431bbff67
commit a3ebd21b25
2 changed files with 44 additions and 12 deletions

View File

@ -413,7 +413,6 @@ setup(name="tahoe-lafs", # also set in __init__.py
"beautifulsoup4", "beautifulsoup4",
"html5lib", "html5lib",
"junitxml", "junitxml",
"tenacity",
# Pin old version until # Pin old version until
# https://github.com/paramiko/paramiko/issues/1961 is fixed. # https://github.com/paramiko/paramiko/issues/1961 is fixed.
"paramiko < 2.9", "paramiko < 2.9",

View File

@ -4,18 +4,14 @@ Tests for allmydata.util.iputil.
Ported to Python 3. Ported to Python 3.
""" """
from __future__ import absolute_import from __future__ import annotations
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
from future.utils import PY2, native_str
if PY2:
from builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, list, object, range, str, max, min # noqa: F401
import os, socket import os, socket
import gc import gc
from functools import wraps
from typing import TypeVar, Callable
from typing_extensions import TypeAlias
from testtools.matchers import ( from testtools.matchers import (
MatchesAll, MatchesAll,
IsInstance, IsInstance,
@ -25,8 +21,6 @@ from testtools.matchers import (
from twisted.trial import unittest from twisted.trial import unittest
from tenacity import retry, stop_after_attempt
from foolscap.api import Tub from foolscap.api import Tub
from allmydata.util import iputil, gcutil from allmydata.util import iputil, gcutil
@ -39,6 +33,45 @@ from .common import (
SyncTestCase, SyncTestCase,
) )
T = TypeVar("T")
TestFunction: TypeAlias = Callable[[], T]
Decorator: TypeAlias = Callable[[TestFunction[T]], TestFunction[T]]
def retry(stop: Callable[[], bool]) -> Decorator[T]:
"""
Call a function until the predicate says to stop or the function stops
raising an exception.
:param stop: A callable to call after the decorated function raises an
exception. The decorated function will be called again if ``stop``
returns ``False``.
:return: A decorator function.
"""
def decorate(f: TestFunction[T]) -> TestFunction[T]:
@wraps(f)
def decorator(self) -> T:
while True:
try:
return f(self)
except Exception:
if stop():
raise
return decorator
return decorate
def stop_after_attempt(limit: int) -> Callable[[], bool]:
"""
Stop after ``limit`` calls.
"""
counter = 0
def check():
nonlocal counter
counter += 1
return counter < limit
return check
class ListenOnUsed(unittest.TestCase): class ListenOnUsed(unittest.TestCase):
"""Tests for listenOnUnused.""" """Tests for listenOnUnused."""
@ -127,7 +160,7 @@ class GetLocalAddressesSyncTests(SyncTestCase):
IsInstance(list), IsInstance(list),
AllMatch( AllMatch(
MatchesAll( MatchesAll(
IsInstance(native_str), IsInstance(str),
MatchesPredicate( MatchesPredicate(
lambda addr: socket.inet_pton(socket.AF_INET, addr), lambda addr: socket.inet_pton(socket.AF_INET, addr),
"%r is not an IPv4 address.", "%r is not an IPv4 address.",