Source code for univention.testing.decorators
# SPDX-FileCopyrightText: 2024-2025 Univention GmbH
# SPDX-License-Identifier: AGPL-3.0-only
import time
from collections.abc import Callable
from functools import wraps
from typing import Any, TypeVar, cast
DEFAULT_TIMEOUT = 90 # seconds
F = TypeVar('F', bound=Callable[..., None])
[docs]
class WaitForNonzeroResultOrTimeout:
def __init__(self, func: Callable[..., Any], timeout: int = DEFAULT_TIMEOUT) -> None:
self.func = func
self.timeout = timeout
def __call__(self, *args: Any, **kwargs: Any) -> Any:
for _i in range(self.timeout):
result = self.func(*args, **kwargs)
if result:
break
else:
time.sleep(1)
return result
[docs]
class SetTimeout:
def __init__(self, func: Callable[..., None], timeout: int = DEFAULT_TIMEOUT) -> None:
self.func = func
self.timeout = timeout
def __call__(self, *args: Any, **kwargs: Any) -> Any:
for i in range(self.timeout):
try:
print("** Entering", self.func.__name__)
self.func(*args, **kwargs)
print("** Exiting", self.func.__name__)
break
except Exception as ex:
print("(%d)-- Exception cought: %s %s" % (i, type(ex), ex))
time.sleep(1)
else:
self.func(*args, **kwargs)
[docs]
def setTimeout(timeout: int = DEFAULT_TIMEOUT) -> Callable[[F], F]:
def decorator(func: F) -> F:
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> None:
for i in range(timeout):
try:
print("** Entering", func.__name__)
func(*args, **kwargs)
print("** Exiting", func.__name__)
break
except Exception as ex:
print("(%d)-- Exception cought: %s %s" % (i, type(ex), ex))
time.sleep(1)
else:
func(*args, **kwargs)
return cast(F, wrapper)
return decorator