mirror of
https://github.com/kernelkit/infix.git
synced 2026-07-22 01:13:00 +02:00
Joachim Wiberg
a5c807abe0
Any exception raised by fn() propagated immediately out of the retry loop, effectively making until() a single-shot call the moment any transient error occurred. Fix by wrapping fn() in a try/except inside the loop and treating any exception as a "not yet" result. The last exception is preserved and re-raised if all attempts are exhausted, so failure output is still meaningful. Also fix a missing newline at end of file. Fixes #1403 Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
130 lines
3.7 KiB
Python
130 lines
3.7 KiB
Python
"""Helper functions for tests"""
|
|
import base64
|
|
import time
|
|
import threading
|
|
import infamy.neigh
|
|
from infamy import netconf
|
|
from infamy import restconf
|
|
import urllib.request
|
|
|
|
|
|
class ParallelFn(threading.Thread):
|
|
def __init__(self, group=None, target=None, name=None,
|
|
args=(), kwargs={}, Verbose=None):
|
|
threading.Thread.__init__(self, group, target, name, args, kwargs)
|
|
self._exc, self._return = None, None
|
|
|
|
def run(self):
|
|
if self._target is not None:
|
|
try:
|
|
self._return = self._target(*self._args,
|
|
**self._kwargs)
|
|
except Exception as e:
|
|
self._exc = e
|
|
|
|
def join(self, *args):
|
|
threading.Thread.join(self, *args)
|
|
|
|
if self._exc:
|
|
raise self._exc
|
|
|
|
return self._return
|
|
|
|
|
|
def parallel(*fns):
|
|
ths = [ParallelFn(target=fn) for fn in fns]
|
|
[th.start() for th in ths]
|
|
return [th.join() for th in ths]
|
|
|
|
|
|
def until(fn, attempts=10, interval=1):
|
|
last_exc = None
|
|
for attempt in range(attempts):
|
|
try:
|
|
result = fn()
|
|
except Exception as e:
|
|
last_exc = e
|
|
result = False
|
|
if result:
|
|
return result
|
|
|
|
time.sleep(interval)
|
|
|
|
if last_exc:
|
|
raise last_exc
|
|
raise Exception("Expected condition did not materialize")
|
|
|
|
|
|
def is_reachable(neigh, env, pwd):
|
|
if env.args.transport is not None:
|
|
if env.args.transport == "netconf":
|
|
return netconf.netconf_syn(neigh)
|
|
elif env.args.transport == "restconf":
|
|
return restconf.restconf_reachable(neigh, pwd)
|
|
else:
|
|
raise Exception(f"Unsupported transport {env.args.transport}!")
|
|
else:
|
|
# No transport specified, so check both netconf and restconf
|
|
netconf_reachable = netconf.netconf_syn(neigh)
|
|
restconf_reachable = restconf.restconf_reachable(neigh, pwd)
|
|
return netconf_reachable and restconf_reachable
|
|
|
|
|
|
def to_binary(text):
|
|
"""Base64 encode the text, removing newlines"""
|
|
enc = base64.b64encode(text.encode('utf-8'))
|
|
|
|
# Convert the encoded bytes to a string and remove any newlines
|
|
return enc.decode('utf-8').replace('\n', '')
|
|
|
|
|
|
def wait_boot(target, env):
|
|
print(f"{target} is shutting down ...")
|
|
until(lambda: not target.reachable(), attempts=100)
|
|
|
|
print(f"{target} is booting up ...")
|
|
until(lambda: target.reachable(), attempts=300)
|
|
|
|
iface = target.get_mgmt_iface()
|
|
if not iface:
|
|
return False
|
|
|
|
neigh = infamy.neigh.ll6ping(iface)
|
|
if not neigh:
|
|
return False
|
|
|
|
print(f"{target} is responding to IPv6 ping ...")
|
|
|
|
pwd = target.location.password
|
|
until(lambda: is_reachable(neigh, env, pwd), attempts=300)
|
|
|
|
return True
|
|
|
|
|
|
def warn(msg):
|
|
"""Print a warning message in yellow to stderr."""
|
|
YELLOW = "\033[93m"
|
|
RST = "\033[0m"
|
|
print(f"{YELLOW}warn - {msg}{RST}")
|
|
|
|
def curl(url, timeout=10, silent=False):
|
|
"""Fetch a URL and return its response body as a UTF-8 string.
|
|
|
|
Args:
|
|
url (str): The full URL to fetch.
|
|
timeout (int): Request timeout in seconds.
|
|
silent (bool): If True, suppress warning on failure (for retry scenarios).
|
|
|
|
Returns:
|
|
str | None: Response body as text, or None if the request failed.
|
|
"""
|
|
url = urllib.parse.quote(url, safe='/:')
|
|
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as response:
|
|
return response.read().decode('utf-8', errors='replace')
|
|
except (urllib.error.URLError, ConnectionResetError, UnicodeEncodeError, TimeoutError) as e:
|
|
if not silent:
|
|
print(f"[WARN] curl: failed to fetch {url}: {e}")
|
|
return ""
|