diff --git a/test/test-network/systemd-networkd-tests.py b/test/test-network/systemd-networkd-tests.py index 0233c4dd1f..ba16170393 100755 --- a/test/test-network/systemd-networkd-tests.py +++ b/test/test-network/systemd-networkd-tests.py @@ -594,6 +594,21 @@ class Utilities(): def check_link_attr(self, *args): self.assertEqual(read_link_attr(*args[:-1]), args[-1]); + def wait_activated(self, link, state='down', timeout=20, fail_assert=True): + # wait for the interface is activated. + invocation_id = check_output('systemctl show systemd-networkd -p InvocationID --value') + needle = f'{link}: Bringing link {state}' + flag = state.upper() + for iteration in range(timeout+1): + output = check_output('journalctl _SYSTEMD_INVOCATION_ID=' + invocation_id) + if needle in output and flag in check_output(f'ip link show {link}'): + return True + if iteration < timeout: + time.sleep(1) + if fail_assert: + 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. @@ -3053,7 +3068,6 @@ class NetworkdNetworkTests(unittest.TestCase, Utilities): self.wait_operstate('test1', 'routable') def _test_activation_policy(self, test): - self.setUp() conffile = '25-activation-policy.network' if test: conffile = f'{conffile}.d/{test}.conf' @@ -3061,13 +3075,13 @@ class NetworkdNetworkTests(unittest.TestCase, Utilities): start_networkd() always = test.startswith('always') - if test == 'manual': - initial_up = 'UP' in check_output('ip link show test1') - else: - initial_up = not test.endswith('down') # note: default is up + initial_up = test != 'manual' and not test.endswith('down') # note: default is up expect_up = initial_up next_up = not expect_up + if test.endswith('down'): + self.wait_activated('test1') + for iteration in range(4): with self.subTest(iteration=iteration, expect_up=expect_up): operstate = 'routable' if expect_up else 'off' @@ -3087,16 +3101,17 @@ class NetworkdNetworkTests(unittest.TestCase, Utilities): check_output('ip link set dev test1 down') expect_up = initial_up if always else next_up next_up = not next_up - - self.tearDown() + if always: + time.sleep(1) def test_activation_policy(self): for test in ['up', 'always-up', 'manual', 'always-down', 'down', '']: with self.subTest(test=test): + self.setUp() self._test_activation_policy(test) + self.tearDown() def _test_activation_policy_required_for_online(self, policy, required): - self.setUp() conffile = '25-activation-policy.network' units = ['11-dummy.netdev', '12-dummy.netdev', '12-dummy.network', conffile] if policy: @@ -3106,6 +3121,9 @@ class NetworkdNetworkTests(unittest.TestCase, Utilities): copy_unit_to_networkd_unit_path(*units, dropins=False) start_networkd() + if policy.endswith('down'): + self.wait_activated('test1') + if policy.endswith('down') or policy == 'manual': self.wait_operstate('test1', 'off', setup_state='configuring') else: @@ -3131,13 +3149,13 @@ class NetworkdNetworkTests(unittest.TestCase, Utilities): yesno = 'yes' if expected else 'no' self.assertRegex(output, f'Required For Online: {yesno}') - self.tearDown() - def test_activation_policy_required_for_online(self): for policy in ['up', 'always-up', 'manual', 'always-down', 'down', 'bound', '']: for required in ['yes', 'no', '']: with self.subTest(policy=policy, required=required): + self.setUp() self._test_activation_policy_required_for_online(policy, required) + self.tearDown() def test_domain(self): copy_unit_to_networkd_unit_path('12-dummy.netdev', '24-search-domain.network')