test-network: simplify and unify waiting loop

[thirdparty/systemd.git] / test / test-network / systemd-networkd-tests.py

diff --git a/test/test-network/systemd-networkd-tests.py b/test/test-network/systemd-networkd-tests.py
index d371ff193ee3c7238a56e7f080ff88c5103752dd..eb9a2ebdfcbd1cf87718bf08240041eaded87b60 100755 (executable)
--- a/test/test-network/systemd-networkd-tests.py
+++ b/test/test-network/systemd-networkd-tests.py
@@ -1017,32 +1017,26 @@ class Utilities():
     def check_ipv6_neigh_sysctl_attr(self, link, attribute, expected):
         self.assertEqual(read_ipv6_neigh_sysctl_attr(link, attribute), expected)
 
-    def wait_links(self, *links, timeout=20, fail_assert=True):
-        for iteration in range(timeout + 1):
-            if iteration > 0:
-                time.sleep(1)
-
+    def wait_links(self, *links, trial=40):
+        for _ in range(trial):
             if link_exists(*links):
-                return True
-        if fail_assert:
+                break
+            time.sleep(0.5)
+        else:
             self.fail('Timed out waiting for all links to be created: ' + ', '.join(list(links)))
-        return False
 
-    def wait_activated(self, link, state='down', timeout=20, fail_assert=True):
+    def wait_activated(self, link, state='down', trial=40):
         # wait for the interface is activated.
         needle = f'{link}: Bringing link {state}'
         flag = state.upper()
-        for iteration in range(timeout + 1):
-            if iteration != 0:
-                time.sleep(1)
-            if not link_exists(link):
-                continue
-            output = read_networkd_log()
-            if needle in output and flag in check_output(f'ip link show {link}'):
-                return True
-        if fail_assert:
+        self.wait_links(link, trial=trial)
+        self.check_networkd_log(needle, trial=trial)
+        for _ in range(trial):
+            if flag in check_output(f'ip link show {link}'):
+                break
+            time.sleep(0.5)
+        else:
             self.fail(f'Timed out waiting for {link} activated.')
-        return False
 
     def wait_operstate(self, link, operstate='degraded', setup_state='configured', setup_timeout=5, fail_assert=True):
         """Wait for the link to reach the specified operstate and/or setup state.
@@ -1063,14 +1057,13 @@ class Utilities():
         if not setup_state:
             setup_state = r'\S+'
 
-        for secs in range(setup_timeout + 1):
-            if secs != 0:
-                time.sleep(1)
+        for _ in range(setup_timeout * 2):
             if not link_exists(link):
                 continue
             output = networkctl_status(link)
             if re.search(rf'(?m)^\s*State:\s+{operstate}\s+\({setup_state}\)\s*$', output):
                 return True
+            time.sleep(0.5)
 
         if fail_assert:
             self.fail(f'Timed out waiting for {link} to reach state {operstate}/{setup_state}')
@@ -1126,42 +1119,38 @@ class Utilities():
                 self.wait_operstate(link.split(':')[0], None, setup_state, setup_timeout)
 
     def wait_address(self, link, address_regex, scope='global', ipv='', timeout_sec=100):
-        for i in range(timeout_sec):
-            if i > 0:
-                time.sleep(1)
+        for _ in range(timeout_sec * 2):
             output = check_output(f'ip {ipv} address show dev {link} scope {scope}')
             if re.search(address_regex, output) and 'tentative' not in output:
                 break
+            time.sleep(0.5)
 
         self.assertRegex(output, address_regex)
 
     def wait_address_dropped(self, link, address_regex, scope='global', ipv='', timeout_sec=100):
-        for i in range(timeout_sec):
-            if i > 0:
-                time.sleep(1)
+        for _ in range(timeout_sec * 2):
             output = check_output(f'ip {ipv} address show dev {link} scope {scope}')
             if not re.search(address_regex, output):
                 break
+            time.sleep(0.5)
 
         self.assertNotRegex(output, address_regex)
 
     def wait_route(self, link, route_regex, table='main', ipv='', timeout_sec=100):
-        for i in range(timeout_sec):
-            if i > 0:
-                time.sleep(1)
+        for _ in range(timeout_sec * 2):
             output = check_output(f'ip {ipv} route show dev {link} table {table}')
             if re.search(route_regex, output):
                 break
+            time.sleep(0.5)
 
         self.assertRegex(output, route_regex)
 
     def wait_route_dropped(self, link, route_regex, table='main', ipv='', timeout_sec=100):
-        for i in range(timeout_sec):
-            if i > 0:
-                time.sleep(1)
+        for _ in range(timeout_sec * 2):
             output = check_output(f'ip {ipv} route show dev {link} table {table}')
             if not re.search(route_regex, output):
                 break
+            time.sleep(0.5)
 
         self.assertNotRegex(output, route_regex)
 
@@ -1855,25 +1844,23 @@ class NetworkdNetDevTests(unittest.TestCase, Utilities):
         self.assertRegex(output, r'(?m)testtap99: tap pi (multi_queue |)vnet_hdr persist filter *(0x100|)\n\tAttached to processes:$')
         self.assertRegex(output, r'(?m)testtun99: tun pi (multi_queue |)vnet_hdr persist filter *(0x100|)\n\tAttached to processes:$')
 
-        for i in range(10):
-            if i != 0:
-                time.sleep(1)
+        for _ in range(20):
             output = check_output('ip -d link show testtun99')
             print(output)
             self.assertRegex(output, 'tun (type tun pi on vnet_hdr on multi_queue|addrgenmode) ')
             if 'NO-CARRIER' in output:
                 break
+            time.sleep(0.5)
         else:
             self.fail()
 
-        for i in range(10):
-            if i != 0:
-                time.sleep(1)
+        for _ in range(20):
             output = check_output('ip -d link show testtap99')
             print(output)
             self.assertRegex(output, 'tun (type tap pi on vnet_hdr on multi_queue|addrgenmode) ')
             if 'NO-CARRIER' in output:
                 break
+            time.sleep(0.5)
         else:
             self.fail()
 
@@ -5406,14 +5393,12 @@ class NetworkdLLDPTests(unittest.TestCase, Utilities):
         start_networkd()
         self.wait_online('veth99:degraded', 'veth-peer:degraded')
 
-        for trial in range(10):
-            if trial > 0:
-                time.sleep(1)
-
+        for _ in range(20):
             output = networkctl('lldp')
             print(output)
             if re.search(r'veth99 .* veth-peer .* .......a...', output):
                 break
+            time.sleep(0.5)
         else:
             self.fail()
 
@@ -5460,14 +5445,12 @@ class NetworkdLLDPTests(unittest.TestCase, Utilities):
         networkctl_reload()
         self.wait_online('veth-peer:degraded')
 
-        for trial in range(10):
-            if trial > 0:
-                time.sleep(1)
-
+        for _ in range(20):
             output = networkctl('lldp')
             print(output)
             if re.search(r'veth99 .* veth-peer .* ....r......', output):
                 break
+            time.sleep(0.5)
         else:
             self.fail()