Merge pull request #22277 from yuwata/test-network-activation-policy

test-network: fixes for test_activation_policy

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')