fix: correctly retry device read when device is unstable

library/network_connections.py

@@ -235,7 +235,7 @@
                             "cannot read stable link-infos. They keep changing"
                         )
                 except Exception:
-                    if try_count < 50:
+                    if try_count >= 50:
                         raise
                     continue
                 break
@@ -985,7 +985,7 @@
             s_con.set_property(
                 # wokeignore:rule=dummy
                 NM.SETTING_CONNECTION_TYPE,
                 NM.SETTING_DUMMY_SETTING_NAME,
             )
         elif connection["type"] == "vlan":
             s_con.set_property(NM.SETTING_CONNECTION_TYPE, NM.SETTING_VLAN_SETTING_NAME)

tests/unit/test_network_connections.py

@@ -5591,12 +5591,71 @@ def test_invalid_pkey_values(self):
         )
 
 
-class TestSysUtils(unittest.TestCase):
+class TestSysUtils(Python26CompatTestCase):
+    def setUp(self):
+        if hasattr(SysUtil, "_link_infos"):
+            del SysUtil._link_infos
+
     def test_link_read_permaddress(self):
         self.assertEqual(SysUtil._link_read_permaddress("lo"), "00:00:00:00:00:00")
         self.assertEqual(SysUtil._link_read_permaddress("fakeiface"), None)
         self.assertEqual(SysUtil._link_read_permaddress("morethansixteenchars"), None)
 
+    def test_link_infos_retries_when_fetch_results_differ(self):
+        """Simulate unstable /sys reads; link_infos must retry, not fail once."""
+        stable = {
+            "eth0": {
+                "ifindex": 2,
+                "ifname": "eth0",
+                "address": "52:54:00:00:00:01",
+                "perm-address": None,
+                "bond-port-perm-hwaddr": None,
+            }
+        }
+        unstable = {
+            "eth1": {
+                "ifindex": 3,
+                "ifname": "eth1",
+                "address": "52:54:00:00:00:02",
+                "perm-address": None,
+                "bond-port-perm-hwaddr": None,
+            }
+        }
+        fetch_mock = mock.Mock(side_effect=[unstable, stable, stable])
+        with mock.patch.object(SysUtil, "_link_infos_fetch", fetch_mock):
+            result = SysUtil.link_infos(refresh=True)
+        self.assertEqual(result, stable)
+        self.assertEqual(fetch_mock.call_count, 3)
+
+    def test_link_infos_raises_after_max_retries(self):
+        """Unstable reads must eventually fail after 50 attempts."""
+        counter = itertools.count()
+
+        def unstable_fetch():
+            n = next(counter)
+            return {
+                "eth{n}".format(n=n): {
+                    "ifindex": n,
+                    "ifname": "eth{n}".format(n=n),
+                    "address": "52:54:00:00:00:{0:02x}".format(n % 256),
+                    "perm-address": None,
+                    "bond-port-perm-hwaddr": None,
+                }
+            }
+
+        # Use Mock(side_effect=...) so Py2.7 does not treat the replacement as an
+        # unbound method (plain functions patched onto @staticmethod fail there).
+        fetch_mock = mock.Mock(side_effect=unstable_fetch)
+        with mock.patch.object(SysUtil, "_link_infos_fetch", fetch_mock):
+            self.assertRaisesRegex(
+                Exception,
+                "stable link-infos",
+                SysUtil.link_infos,
+                refresh=True,
+            )
+        # 50 attempts: 2 fetches on first try, then 1 per retry (2 + 49 = 51).
+        self.assertEqual(fetch_mock.call_count, 51)
+
 
 if __name__ == "__main__":
     unittest.main()