summaryrefslogtreecommitdiff
path: root/src/tests
diff options
context:
space:
mode:
authorChristian Breunig <christian@breunig.cc>2026-08-06 21:59:21 +0200
committerChristian Breunig <christian@breunig.cc>2026-08-08 18:33:38 +0200
commit22cfee116c9935f4a33da4923eff189f2f646371 (patch)
treed0a125a6178d4bbefd9515003e285479012103f2 /src/tests
parent3c7aa6d76c4de62f27a0493baefe8e10044c9f9d (diff)
downloadvyos-1x-22cfee116c9935f4a33da4923eff189f2f646371.tar.gz
vyos-1x-22cfee116c9935f4a33da4923eff189f2f646371.zip
T3871: make boot-time interface naming deterministic and hw-id aware
Multi-vendor PCIe NIC systems could lose or rename Ethernet/wireless interfaces on boot, because naming was decided from a single per-device udev event before all hardware had a chance to enumerate. Replace that with one authoritative pass, run once configuration is available during router startup: wait for configured hardware, apply every hw-id binding, and name whatever has none yet by PCIe distance from the root complex and MAC address - the same result every boot. A node with its hw-id deleted (NIC replacement) or its whole configuration removed is treated as an ordinary open slot, filled the same way - recovering its original hardware whenever the interfaces that vacated slots this boot also show up as candidates. Boot now reports any interface still unresolved, and the pass is covered by a new test suite in vyos-build named: make testifname Assisted-by: Claude:claude-sonnet-5
Diffstat (limited to 'src/tests')
-rw-r--r--src/tests/helper.py11
-rw-r--r--src/tests/test_net_name_resolve.py1534
2 files changed, 1545 insertions, 0 deletions
diff --git a/src/tests/helper.py b/src/tests/helper.py
index 2c4421d0a..5a74525b7 100644
--- a/src/tests/helper.py
+++ b/src/tests/helper.py
@@ -14,9 +14,20 @@
import sys
import importlib.util
+from importlib.machinery import SourceFileLoader
def prepare_module(file_path='', module_name=''):
+ """Load file_path as module_name, returning the loaded module.
+
+ file_path may be an extensionless script (e.g. udev helpers under
+ src/udev/) - spec_from_file_location() cannot infer a loader for those
+ from the suffix alone, so fall back to an explicit SourceFileLoader.
+ """
spec = importlib.util.spec_from_file_location(module_name, file_path)
+ if spec is None:
+ loader = SourceFileLoader(module_name, file_path)
+ spec = importlib.util.spec_from_loader(module_name, loader)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
sys.modules[module_name] = module
+ return module
diff --git a/src/tests/test_net_name_resolve.py b/src/tests/test_net_name_resolve.py
new file mode 100644
index 000000000..cc401ddb3
--- /dev/null
+++ b/src/tests/test_net_name_resolve.py
@@ -0,0 +1,1534 @@
+# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+# Coverage for T3871 (boot race on hw-id based interface naming): the
+# MAC-driven, collision-safe rename plan that replaces the old per-uevent
+# udev decision, the find_available() crash it used to hit, the
+# ethtool/sysfs permanent-MAC fallback, and the config.boot-availability
+# ordering corner case (vyos-net-name-resolve must run from inside
+# vyos-router, after the config directory is mounted/decrypted, not as an
+# independently-scheduled early-boot unit).
+
+import json
+import os
+import shutil
+import tempfile
+import unittest
+from unittest import mock
+
+from helper import prepare_module
+
+_here = os.path.dirname(__file__)
+
+resolver = prepare_module(
+ os.path.join(_here, '../system/vyos-net-name-resolve.py'),
+ 'vyos_net_name_resolve')
+vyos_net_name = prepare_module(
+ os.path.join(_here, '../udev/vyos_net_name'),
+ 'vyos_net_name')
+
+
+class TestGetPendingHwidNodes(unittest.TestCase):
+ """A node that exists under interfaces/{ethernet,wireless} but has no
+ hw-id (e.g. after `delete interfaces ethernet eth1 hw-id`, the
+ documented "NIC replaced" remediation) must be surfaced separately
+ from the MAC-keyed configured dict, since it has no MAC to key by yet
+ - this is what lets main() try to reclaim it for the exact NIC that
+ vacated it, instead of silently orphaning its other settings.
+ """
+
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+ self.config_path = os.path.join(self.tmp, 'config.boot')
+ self._orig_config_path = resolver.config_path
+ resolver.config_path = self.config_path
+ self.addCleanup(setattr, resolver, 'config_path', self._orig_config_path)
+
+ def _write_config(self, text):
+ with open(self.config_path, 'w') as f:
+ f.write(text)
+
+ def test_node_with_hwid_not_pending(self):
+ self._write_config(
+ 'interfaces {\n'
+ ' ethernet eth0 {\n'
+ ' hw-id 00:11:22:33:44:55\n'
+ ' }\n'
+ '}\n'
+ )
+ self.assertEqual(resolver.get_pending_hwid_nodes(),
+ {'ethernet': set(), 'wireless': set()})
+
+ def test_node_without_hwid_is_pending(self):
+ self._write_config(
+ 'interfaces {\n'
+ ' ethernet eth1 {\n'
+ ' address 10.99.1.1/24\n'
+ ' }\n'
+ '}\n'
+ )
+ self.assertEqual(resolver.get_pending_hwid_nodes(),
+ {'ethernet': {'eth1'}, 'wireless': set()})
+
+ def test_mixed_ethernet_and_wireless(self):
+ self._write_config(
+ 'interfaces {\n'
+ ' ethernet eth0 {\n'
+ ' hw-id 00:11:22:33:44:55\n'
+ ' }\n'
+ ' ethernet eth1 {\n'
+ ' address 10.99.1.1/24\n'
+ ' }\n'
+ ' wireless wlan0 {\n'
+ ' ssid example\n'
+ ' }\n'
+ '}\n'
+ )
+ self.assertEqual(resolver.get_pending_hwid_nodes(),
+ {'ethernet': {'eth1'}, 'wireless': {'wlan0'}})
+
+ def test_no_interfaces_section(self):
+ self._write_config('system {\n host-name vyos\n}\n')
+ self.assertEqual(resolver.get_pending_hwid_nodes(),
+ {'ethernet': set(), 'wireless': set()})
+
+ def test_no_config_boot_at_all(self):
+ # livecd/ISO case - config_path doesn't exist
+ self.assertEqual(resolver.get_pending_hwid_nodes(),
+ {'ethernet': set(), 'wireless': set()})
+
+
+class TestFindAvailableCrashFix(unittest.TestCase):
+ """find_available() used to raise IndexError on an empty candidate set
+ (index_list[0]) and ValueError on a name with no trailing digit
+ (int('')) - both are realistic inputs on a box's first boot, or when a
+ biosdevname/predefined name has no numeric suffix.
+ """
+
+ def test_resolver_empty_set(self):
+ self.assertEqual(resolver.find_available(set(), 'eth'), 'eth0')
+
+ def test_resolver_no_prefix_match(self):
+ self.assertEqual(resolver.find_available({'wlan3'}, 'eth'), 'eth0')
+
+ def test_resolver_contiguous(self):
+ names = {'eth0', 'eth1', 'eth2'}
+ self.assertEqual(resolver.find_available(names, 'eth'), 'eth3')
+
+ def test_resolver_fills_hole(self):
+ self.assertEqual(resolver.find_available({'eth2', 'eth5'}, 'eth'), 'eth3')
+
+ def test_resolver_no_hole_appends(self):
+ self.assertEqual(resolver.find_available({'eth2', 'eth3'}, 'eth'), 'eth4')
+
+ def test_vyos_net_name_empty_dict(self):
+ self.assertEqual(vyos_net_name.find_available({}, 'eth'), 'eth0')
+
+ def test_vyos_net_name_no_digit_suffix(self):
+ self.assertEqual(
+ vyos_net_name.find_available({'m': 'wlan3'}, 'eth'), 'eth0')
+
+ def test_vyos_net_name_contiguous(self):
+ intfs = {'a': 'eth0', 'b': 'eth1'}
+ self.assertEqual(vyos_net_name.find_available(intfs, 'eth'), 'eth2')
+
+ def test_vyos_net_name_is_available(self):
+ self.assertTrue(vyos_net_name.is_available({'a': 'eth0'}, 'eth1'))
+ self.assertFalse(vyos_net_name.is_available({'a': 'eth0'}, 'eth0'))
+
+ def test_vyos_net_name_mod_ifname(self):
+ self.assertEqual(vyos_net_name.mod_ifname('e5'), 'eth3')
+ self.assertEqual(vyos_net_name.mod_ifname('e2'), 'eth0')
+ self.assertEqual(vyos_net_name.mod_ifname('e1'), 'eth1')
+ self.assertEqual(vyos_net_name.mod_ifname('wlan0'), 'wlan0')
+
+
+class TestUnmatchedCandidates(unittest.TestCase):
+ """Field report: an unconfigured NIC that happens to be squatting on a
+ DIFFERENT interface's configured hw-id slot this boot (e.g. due to
+ probe-order scrambling) was excluded here just because
+ compute_rename_plan() already scheduled it for eviction. That skipped
+ it past match_pending_nodes() entirely, so a squatter that was really
+ the exact NIC a pending node should reclaim - or one that made a
+ reclaim genuinely ambiguous - fell through to ordinary bootstrap
+ naming and got a permanent hw-id with no ambiguity check at all.
+ Squatters must stay in this pool; only their eviction destination
+ (still computed by compute_rename_plan()) is a fallback, not a reason
+ to hide them from reclaim/ambiguity handling.
+ """
+
+ def test_squatter_is_still_a_candidate(self):
+ configured = {'m0': 'eth1'}
+ current = {'eth1': 'squatter-mac', 'eth9': 'm0'}
+ existing_plan = {'eth1': 'eth3', 'eth9': 'eth1'}
+ candidates = resolver.unmatched_candidates(configured, current, existing_plan)
+ self.assertEqual(candidates, [('squatter-mac', 'eth1')])
+
+ def test_rightful_owner_still_excluded(self):
+ configured = {'m0': 'eth0'}
+ current = {'eth0': 'm0'}
+ candidates = resolver.unmatched_candidates(configured, current, {})
+ self.assertEqual(candidates, [])
+
+
+class TestComputeRenamePlan(unittest.TestCase):
+ """The authoritative rename plan is what fixes the multi-vendor NIC
+ boot race: it is driven purely by MAC address, independent of
+ ifindex/probe order, and must never produce a colliding assignment.
+ """
+
+ def test_noop_when_already_correct(self):
+ configured = {'m0': 'eth0', 'm1': 'eth1'}
+ current = {'eth0': 'm0', 'eth1': 'm1'}
+ self.assertEqual(resolver.compute_rename_plan(configured, current), {})
+
+ def test_rename_independent_of_probe_order(self):
+ # m0/m1 landed on unrelated names this boot (e.g. a different
+ # vendor driver won the probe race) - the plan must still point
+ # them at their configured targets regardless.
+ configured = {'m0': 'eth0', 'm1': 'eth1'}
+ current = {'eth5': 'm0', 'eth3': 'm1'}
+ plan = resolver.compute_rename_plan(configured, current)
+ self.assertEqual(plan, {'eth5': 'eth0', 'eth3': 'eth1'})
+
+ def test_full_swap_cycle(self):
+ # eth0 and eth1 need to trade names - a naive from->to rename
+ # would collide; the two-phase scratch-name staging must not.
+ configured = {'m0': 'eth1', 'm1': 'eth0'}
+ current = {'eth0': 'm0', 'eth1': 'm1'}
+ plan = resolver.compute_rename_plan(configured, current)
+ self.assertEqual(plan, {'eth0': 'eth1', 'eth1': 'eth0'})
+
+ def test_squatter_relocated(self):
+ # an unconfigured interface currently sits on a name a configured
+ # hw-id needs - it must be moved out of the way, not left there.
+ configured = {'m0': 'eth0'}
+ current = {'eth0': 'unconfigured-mac', 'eth5': 'm0'}
+ plan = resolver.compute_rename_plan(configured, current)
+ self.assertEqual(plan['eth5'], 'eth0')
+ self.assertNotEqual(plan['eth0'], 'eth0')
+ self.assertTrue(plan['eth0'].startswith('eth'))
+
+ def test_squatter_never_relocated_onto_a_missing_hwid_target(self):
+ # macMissing's hardware hasn't shown up this boot (removed/faulty/
+ # slow driver), so its reserved target 'eth1' never appears in
+ # `current` and the main hw-id loop skips it entirely (source is
+ # None) - the squatter being relocated off of 'eth0' must still not
+ # be handed 'eth1', or it would need relocating again the moment
+ # that hardware actually appears.
+ configured = {'macA': 'eth0', 'macMissing': 'eth1'}
+ current = {'eth0': 'squatterMac', 'eth2': 'macA'}
+ plan = resolver.compute_rename_plan(configured, current)
+ self.assertEqual(plan.get('eth2'), 'eth0')
+ self.assertNotEqual(plan.get('eth0'), 'eth1')
+
+ def test_squatter_never_relocated_onto_a_pending_hwid_less_node(self):
+ # 'eth1' has no hw-id at all (e.g. `delete interfaces ethernet
+ # eth1 hw-id`, the documented remediation for a replaced NIC) but
+ # its config node - and settings like address - still exist. A
+ # squatter being relocated off of 'eth0' must not be handed 'eth1'
+ # either, or it would permanently steal the exact slot the
+ # original NIC needs to reclaim.
+ configured = {'macA': 'eth0'}
+ pending = {'ethernet': {'eth1'}, 'wireless': set()}
+ current = {'eth0': 'squatterMac', 'eth2': 'macA'}
+ plan = resolver.compute_rename_plan(configured, current, pending)
+ self.assertEqual(plan.get('eth2'), 'eth0')
+ self.assertNotEqual(plan.get('eth0'), 'eth1')
+
+ def test_missing_hardware_no_crash_no_entry(self):
+ # hw-id configured but the NIC never showed up (missing/faulty) -
+ # must be skipped, not crash or incorrectly assign another interface.
+ configured = {'m0': 'eth0', 'missing-mac': 'eth9'}
+ current = {'eth5': 'm0'}
+ plan = resolver.compute_rename_plan(configured, current)
+ self.assertEqual(plan, {'eth5': 'eth0'})
+
+ def test_rename_introducing_a_gap(self):
+ # user edits the CLI so m2's node is renamed eth2 -> eth10 while
+ # keeping the same hw-id, leaving eth5..eth9 unused.
+ configured = {
+ 'm0': 'eth0', 'm1': 'eth1', 'm2': 'eth10',
+ 'm3': 'eth3', 'm4': 'eth4',
+ }
+ current = {
+ 'eth0': 'm0', 'eth1': 'm1', 'eth2': 'm2',
+ 'eth3': 'm3', 'eth4': 'm4',
+ }
+ plan = resolver.compute_rename_plan(configured, current)
+ self.assertEqual(plan, {'eth2': 'eth10'})
+
+ def test_rename_introducing_a_gap_with_scrambled_fastpath(self):
+ # same as above, but this boot's cosmetic udev fast-path lands m2
+ # on an unrelated name and coincidentally puts a different NIC on
+ # the now-meaningless old 'eth2'.
+ configured = {
+ 'm0': 'eth0', 'm1': 'eth1', 'm2': 'eth10',
+ 'm3': 'eth3', 'm4': 'eth4',
+ }
+ current = {
+ 'eth0': 'm0', 'eth1': 'm1', 'eth7': 'm2',
+ 'eth2': 'm3', 'eth4': 'm4',
+ }
+ plan = resolver.compute_rename_plan(configured, current)
+ self.assertEqual(plan.get('eth7'), 'eth10')
+ self.assertEqual(plan.get('eth2'), 'eth3')
+ self.assertNotIn('eth2', configured.values())
+
+
+class TestWaitForSettle(unittest.TestCase):
+ """There is no specific MAC to wait for when bootstrap-naming hardware
+ without a configured hw-id, so settling is detected as N consecutive
+ identical discover_physical_interfaces() snapshots instead.
+ """
+
+ def test_returns_after_n_consecutive_stable_polls(self):
+ snapshots = [
+ {'eth0': 'm0'}, # still changing
+ {'eth0': 'm0', 'eth1': 'm1'}, # still changing
+ {'eth0': 'm0', 'eth1': 'm1'}, # 1st stable repeat
+ {'eth0': 'm0', 'eth1': 'm1'}, # 2nd stable repeat -> settled
+ ]
+ with mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=snapshots):
+ result = resolver.wait_for_settle(
+ {}, timeout=5, poll=0, stable_polls=3)
+ self.assertEqual(result, {'eth0': 'm0', 'eth1': 'm1'})
+
+ def test_bounded_by_timeout_when_never_stable(self):
+ counter = iter(range(1000))
+
+ def ever_changing(*_a, **_kw):
+ return {'eth0': f'm{next(counter)}'}
+
+ clock = iter([0, 0.1, 20, 20])
+ with mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=ever_changing), \
+ mock.patch('time.monotonic', side_effect=lambda: next(clock)):
+ result = resolver.wait_for_settle(
+ {}, timeout=10, poll=0, stable_polls=3)
+ # must return SOMETHING (the last snapshot seen), not hang or crash
+ self.assertIn('eth0', result)
+
+ def test_stable_polls_of_one_returns_on_initial_snapshot(self):
+ with mock.patch.object(resolver, 'discover_physical_interfaces') as m:
+ result = resolver.wait_for_settle(
+ {'eth0': 'm0'}, timeout=5, poll=0, stable_polls=1)
+ m.assert_not_called()
+ self.assertEqual(result, {'eth0': 'm0'})
+
+
+class TestIsWirelessInterface(unittest.TestCase):
+ """Wireless detection must be race-free - it has to be correct even
+ when the interface's current name is a leftover from the cosmetic
+ udev-time fast path and doesn't look like a wlan name at all.
+ """
+
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+ # phy80211 is always a symlink to a real sysfs node in practice -
+ # exists() follows symlinks, so the test target must actually exist
+ self.phy_target = os.path.join(self.tmp, 'phy0')
+ os.mkdir(self.phy_target)
+
+ def test_phy80211_present_is_wireless(self):
+ path = os.path.join(self.tmp, 'wlan0')
+ os.mkdir(path)
+ os.symlink(self.phy_target, os.path.join(path, 'phy80211'))
+ self.assertTrue(resolver.is_wireless_interface('wlan0', self.tmp))
+
+ def test_no_phy80211_is_not_wireless(self):
+ os.mkdir(os.path.join(self.tmp, 'eth0'))
+ self.assertFalse(resolver.is_wireless_interface('eth0', self.tmp))
+
+ def test_detection_independent_of_misleading_name(self):
+ # named like a leftover cosmetic ethernet guess, but it's a real
+ # wifi card - must still be classified wireless
+ path = os.path.join(self.tmp, 'eth7')
+ os.mkdir(path)
+ os.symlink(self.phy_target, os.path.join(path, 'phy80211'))
+ self.assertTrue(resolver.is_wireless_interface('eth7', self.tmp))
+
+
+class TestPcieDistance(unittest.TestCase):
+ """PCIe hop-depth is what lets bootstrap naming reflect physical slot
+ position instead of raw MAC magnitude - a NIC wired straight to the
+ root complex must not sort after an add-in card just because its MAC
+ happens to be numerically higher.
+ """
+
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+ # fake /sys/devices tree, separate from the fake /sys/class/net
+ # entries, so symlink targets can point at realistic PCI-bus-shaped
+ # paths the way a real /sys/class/net/<if>/device symlink does
+ self.devices_root = os.path.join(self.tmp, 'devices')
+ os.makedirs(self.devices_root)
+
+ def _make_iface_pointing_at(self, name, *path_segments):
+ """Create <tmp>/<name>/device as a symlink to a fake sysfs device
+ node at devices_root/<path_segments...>.
+ """
+ target = os.path.join(self.devices_root, *path_segments)
+ os.makedirs(target, exist_ok=True)
+ iface_path = os.path.join(self.tmp, name)
+ os.mkdir(iface_path)
+ os.symlink(target, os.path.join(iface_path, 'device'))
+
+ def test_shallow_device_one_pci_segment(self):
+ # e.g. a NIC directly under the root complex: pci0000:00/0000:00:1f.6
+ self._make_iface_pointing_at('eth0', 'pci0000:00', '0000:00:1f.6')
+ self.assertEqual(resolver.pcie_distance('eth0', self.tmp), 1)
+
+ def test_deep_device_behind_bridges(self):
+ # e.g. an add-in card behind two cascaded PCIe bridges
+ self._make_iface_pointing_at(
+ 'eth1', 'pci0000:00', '0000:00:1c.0',
+ '0000:01:00.0', '0000:02:04.0')
+ self.assertEqual(resolver.pcie_distance('eth1', self.tmp), 3)
+
+ def test_virtio_indirection_not_miscounted(self):
+ # verified-real virtio_net layout: device -> virtioN node, whose
+ # OWN parent is the true PCI BDF - the virtioN segment itself must
+ # be skipped, not counted as a hop.
+ self._make_iface_pointing_at(
+ 'eth2', 'pci0000:00', '0000:00:12.0', 'virtio2', 'net')
+ self.assertEqual(resolver.pcie_distance('eth2', self.tmp), 1)
+
+ def test_missing_device_symlink_returns_sentinel(self):
+ os.mkdir(os.path.join(self.tmp, 'eth3')) # no 'device' entry at all
+ self.assertEqual(resolver.pcie_distance('eth3', self.tmp),
+ resolver.PCIE_DISTANCE_UNKNOWN)
+
+ def test_broken_device_symlink_returns_sentinel(self):
+ iface_path = os.path.join(self.tmp, 'eth4')
+ os.mkdir(iface_path)
+ os.symlink(os.path.join(self.devices_root, 'does-not-exist'),
+ os.path.join(iface_path, 'device'))
+ self.assertEqual(resolver.pcie_distance('eth4', self.tmp),
+ resolver.PCIE_DISTANCE_UNKNOWN)
+
+ def test_no_pci_segment_at_all_returns_sentinel(self):
+ # simulated USB NIC: resolved path has no PCI BDF component
+ self._make_iface_pointing_at('eth5', 'usb1', '1-1', '1-1:1.0')
+ self.assertEqual(resolver.pcie_distance('eth5', self.tmp),
+ resolver.PCIE_DISTANCE_UNKNOWN)
+
+
+class TestComputeBootstrapPlan(unittest.TestCase):
+ """Bootstrap naming is what makes a box's very first boot (before any
+ hw-id exists) deterministic, the same way hw-id makes every boot after
+ that deterministic - it must depend only on PCIe topology and MAC
+ rank, never on whatever name the racy cosmetic fast-path happened to
+ assign. pcie_distance is held constant by default in these tests, so
+ they exercise pure MAC-rank ordering (see TestPcieDistance for the
+ topology-ordering cases specifically).
+ """
+
+ def setUp(self):
+ patcher = mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False)
+ self.is_wireless = patcher.start()
+ self.addCleanup(patcher.stop)
+
+ distance_patcher = mock.patch.object(resolver, 'pcie_distance',
+ return_value=0)
+ self.pcie_distance = distance_patcher.start()
+ self.addCleanup(distance_patcher.stop)
+
+ def test_sorted_by_mac_independent_of_current_names(self):
+ # names are in the OPPOSITE order of their MACs
+ current = {'eth5': 'bb', 'eth2': 'aa'}
+ plan = resolver.compute_bootstrap_plan({}, current, {})
+ self.assertEqual(plan, {'eth2': 'eth0', 'eth5': 'eth1'})
+
+ def test_noop_when_already_in_canonical_rank_order(self):
+ current = {'eth0': 'aa', 'eth1': 'bb'}
+ plan = resolver.compute_bootstrap_plan({}, current, {})
+ self.assertEqual(plan, {})
+
+ def test_configured_targets_are_never_overwritten(self):
+ configured = {'m0': 'eth0'}
+ current = {'eth0': 'm0', 'eth9': 'm2', 'eth8': 'm1'}
+ plan = resolver.compute_bootstrap_plan(configured, current, {})
+ self.assertNotIn('eth0', plan.values())
+ self.assertEqual(plan.get('eth8'), 'eth1')
+ self.assertEqual(plan.get('eth9'), 'eth2')
+
+ def test_new_interface_backfills_a_gap_with_no_pending_node(self):
+ # eth1 is a gap between two hw-id'd interfaces, but nothing in
+ # config.boot reserves it (no hw-id, no pending node either - the
+ # interface was fully deleted at some point, or never existed) -
+ # a new, unrelated NIC discovered this boot is free to take it,
+ # the same as it would on a system with no config for it at all.
+ configured = {'m0': 'eth0', 'm2': 'eth2'}
+ current = {'eth0': 'm0', 'eth2': 'm2', 'eth9': 'new-mac'}
+ plan = resolver.compute_bootstrap_plan(configured, current, {})
+ self.assertEqual(plan, {'eth9': 'eth1'})
+
+ def test_multiple_new_interfaces_backfill_gaps_in_order(self):
+ configured = {'m0': 'eth0', 'm2': 'eth2'}
+ current = {
+ 'eth0': 'm0', 'eth2': 'm2',
+ 'ethB': 'bb', 'ethA': 'aa',
+ }
+ plan = resolver.compute_bootstrap_plan(configured, current, {})
+ # lower mac 'aa' takes the gap at eth1 first, 'bb' continues to
+ # the next free slot, eth3 - not "skip past the gap to eth3/eth4"
+ self.assertEqual(plan, {'ethA': 'eth1', 'ethB': 'eth3'})
+
+ def test_ethernet_and_wireless_numbered_independently(self):
+ self.is_wireless.side_effect = lambda name: name == 'radio0'
+ current = {'ifaceB': 'bb', 'radio0': 'aa'}
+ plan = resolver.compute_bootstrap_plan({}, current, {})
+ self.assertEqual(plan.get('radio0'), 'wlan0')
+ self.assertEqual(plan.get('ifaceB'), 'eth0')
+
+ def test_squatter_already_in_existing_plan_excluded(self):
+ # 'eth0' is a squatter the hw-id plan already decided to move away
+ # (existing_plan key) - it IS still bootstrap candidacy (see
+ # unmatched_candidates()), but must never be handed a second,
+ # conflicting assignment on top of its existing_plan eviction.
+ existing_plan = {'eth0': 'eth3'}
+ current = {'eth0': 'unconfigured-mac', 'eth7': 'm0'}
+ plan = resolver.compute_bootstrap_plan({'m0': 'eth1'}, current,
+ existing_plan)
+ self.assertNotIn('eth0', plan)
+
+ def test_bootstrap_targets_never_collide_with_existing_plan_values(self):
+ # existing_plan says an interface currently named 'oldname' is
+ # moving to 'eth0' - nothing else in `current` is named 'eth0' or
+ # 'eth*' at all, so if that target isn't reserved, find_available()
+ # would freely hand 'eth0' to the bootstrap candidate too.
+ existing_plan = {'oldname': 'eth0'}
+ current = {'oldname': 'm0', 'newnic': 'unconfigured-mac'}
+ plan = resolver.compute_bootstrap_plan({'m0': 'eth0'}, current,
+ existing_plan)
+ self.assertNotEqual(plan.get('newnic'), 'eth0')
+
+ def test_smaller_pcie_distance_wins_over_higher_mac(self):
+ # 'bb' has the numerically higher MAC but sits closer to the root
+ # complex - it must be named first despite losing on MAC alone.
+ current = {'eth5': 'bb', 'eth2': 'aa'}
+ self.pcie_distance.side_effect = lambda name: {'eth5': 0, 'eth2': 3}[name]
+ plan = resolver.compute_bootstrap_plan({}, current, {})
+ self.assertEqual(plan, {'eth5': 'eth0', 'eth2': 'eth1'})
+
+ def test_same_pcie_distance_falls_back_to_mac_order(self):
+ current = {'eth5': 'bb', 'eth2': 'aa'}
+ self.pcie_distance.side_effect = lambda name: 2 # tie for both
+ plan = resolver.compute_bootstrap_plan({}, current, {})
+ self.assertEqual(plan, {'eth2': 'eth0', 'eth5': 'eth1'})
+
+ def test_unknown_pcie_distance_sorts_last(self):
+ # 'aa' has the numerically lowest MAC but an undeterminable bus
+ # position (e.g. a USB NIC) - it must not jump the queue.
+ current = {'eth9': 'cc', 'eth5': 'bb', 'ethX': 'aa'}
+ self.pcie_distance.side_effect = lambda name: {
+ 'eth9': 1, 'eth5': 2, 'ethX': resolver.PCIE_DISTANCE_UNKNOWN,
+ }[name]
+ plan = resolver.compute_bootstrap_plan({}, current, {})
+ self.assertEqual(plan, {'eth9': 'eth0', 'eth5': 'eth1', 'ethX': 'eth2'})
+
+ def test_pcie_distance_ordering_independent_of_ethernet_wireless_split(self):
+ # distance-based ordering applies within each type group separately,
+ # same as MAC does today - a wlan candidate's distance must not
+ # affect eth numbering or vice versa.
+ self.is_wireless.side_effect = lambda name: name == 'radio0'
+ current = {'ifaceB': 'bb', 'radio0': 'aa'}
+ self.pcie_distance.side_effect = lambda name: {
+ 'ifaceB': 5, 'radio0': 0,
+ }[name]
+ plan = resolver.compute_bootstrap_plan({}, current, {})
+ self.assertEqual(plan.get('radio0'), 'wlan0')
+ self.assertEqual(plan.get('ifaceB'), 'eth0')
+
+
+class TestComputeBootstrapPlanPendingSlots(unittest.TestCase):
+ """A pending node's name (see get_pending_hwid_nodes()) carries no
+ special reservation in compute_bootstrap_plan() at all - it is just
+ another name with nothing configured for it, exactly like a numeric
+ gap. main() is what attributes a candidate landing on such a name
+ back to a "reclaim" afterward, by cross-checking it against
+ get_pending_hwid_nodes() - not this function's job (see main()).
+ """
+
+ def setUp(self):
+ patcher = mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False)
+ self.is_wireless = patcher.start()
+ self.addCleanup(patcher.stop)
+
+ distance_patcher = mock.patch.object(resolver, 'pcie_distance',
+ return_value=0)
+ self.pcie_distance = distance_patcher.start()
+ self.addCleanup(distance_patcher.stop)
+
+ def test_open_name_filled_like_an_ordinary_gap(self):
+ # 'eth1' has nothing configured for it - whether that is because
+ # its whole node was deleted or just its hw-id makes no
+ # difference here, it is simply the lowest open name.
+ configured = {'m0': 'eth0'}
+ current = {'eth0': 'm0', 'eth9': 'new-mac'}
+ plan = resolver.compute_bootstrap_plan(configured, current, {})
+ self.assertEqual(plan.get('eth9'), 'eth1')
+
+ def test_ascending_fill_has_no_preference_between_open_names(self):
+ # two open names (eth1, eth4 - the reason either is open makes no
+ # difference) and two candidates - the lower-sorted candidate
+ # takes the lower-numbered open name, unconditionally.
+ configured = {'m0': 'eth0', 'm2': 'eth2', 'm3': 'eth3'}
+ current = {'eth0': 'm0', 'eth2': 'm2', 'eth3': 'm3',
+ 'eth9': 'bb', 'eth8': 'aa'}
+ plan = resolver.compute_bootstrap_plan(configured, current, {})
+ self.assertEqual(plan.get('eth8'), 'eth1')
+ self.assertEqual(plan.get('eth9'), 'eth4')
+
+ def test_closed_subset_recovers_original_slot_assignment(self):
+ # the vyos-build check-qemu-install --ifnametest shape, at the
+ # function level: two NICs' original slots opened up in the same
+ # boot (one via full node deletion, one via hw-id-only deletion).
+ # PCIe distance and MAC are static per-NIC properties, so the two
+ # freed candidates' relative sort order is identical to their
+ # original first-boot assignment order - re-running the same
+ # ascending sort over just the two open names recovers each one's
+ # own original slot, with no pending-specific logic involved.
+ configured = {
+ '00:00:5e:00:53:00': 'eth0', '00:00:5e:00:53:01': 'eth1',
+ '00:00:5e:00:53:03': 'eth3', '00:00:5e:00:53:05': 'eth5',
+ '00:00:5e:00:53:06': 'eth6', '00:00:5e:00:53:07': 'eth7',
+ }
+ current = {name: mac for mac, name in configured.items()}
+ current.update({'eth9': '00:00:5e:00:53:02',
+ 'eth8': '00:00:5e:00:53:04'})
+ plan = resolver.compute_bootstrap_plan(configured, current, {})
+ self.assertEqual(plan.get('eth9'), 'eth2')
+ self.assertEqual(plan.get('eth8'), 'eth4')
+
+ def test_open_names_currently_squatted_by_a_rightful_mover_are_not_taken(self):
+ # real boot reproduction (via the resolver's own status file,
+ # captured mid-failure): the cosmetic fast-path this boot had
+ # every configured mac sitting on some OTHER configured mac's
+ # target name (a full rotation), including the two open names
+ # (eth1 - pending, eth6 - fully deleted) both currently squatted
+ # by macs that are about to move elsewhere via a rightful-owner
+ # move. existing_plan's SOURCE names looked "taken" at the exact
+ # moment this function ran, but safe_bulk_rename()'s two-phase
+ # staging vacates every source before any target is claimed - so
+ # they must not count as taken, or the two real candidates get
+ # needlessly pushed to fresh eth8/eth9-style slots instead of
+ # their own open names.
+ configured = {
+ '00:00:5e:00:53:00': 'eth0', '00:00:5e:00:53:02': 'eth2',
+ '00:00:5e:00:53:03': 'eth3', '00:00:5e:00:53:04': 'eth4',
+ '00:00:5e:00:53:05': 'eth5', '00:00:5e:00:53:07': 'eth7',
+ }
+ current = {
+ 'eth4': '00:00:5e:00:53:00', 'eth6': '00:00:5e:00:53:02',
+ 'eth7': '00:00:5e:00:53:03', 'eth2': '00:00:5e:00:53:04',
+ 'eth3': '00:00:5e:00:53:05', 'eth1': '00:00:5e:00:53:07',
+ 'eth0': '00:00:5e:00:53:06', # unconfigured - squats on eth0
+ 'eth5': '00:00:5e:00:53:01', # unconfigured - squats on eth5
+ }
+ existing_plan = resolver.compute_rename_plan(configured, current,
+ {'ethernet': {'eth1'},
+ 'wireless': set()})
+ plan = resolver.compute_bootstrap_plan(configured, current,
+ existing_plan)
+ self.assertEqual(plan.get('eth5'), 'eth1')
+ self.assertEqual(plan.get('eth0'), 'eth6')
+
+
+class TestSafeBulkRename(unittest.TestCase):
+ """The two-phase rename must accurately report what actually happened -
+ a phase-2 (scratch -> final target) failure must not be reported as a
+ successful rename, and must not leave the interface down indefinitely
+ under a name nothing else knows about.
+ """
+
+ def setUp(self):
+ patcher = mock.patch.object(resolver, 'get_ifindex',
+ side_effect=lambda name: name)
+ patcher.start()
+ self.addCleanup(patcher.stop)
+
+ def test_all_renames_succeed(self):
+ with mock.patch.object(resolver, 'run', return_value=0):
+ applied = resolver.safe_bulk_rename({'eth5': 'eth0'})
+ self.assertEqual(applied, {'eth5': 'eth0'})
+
+ def test_phase_two_failure_is_not_reported_as_applied(self):
+ # 'ip link set dev vyetheth5 name eth0' fails (exit 1); everything
+ # else (down, the phase-1 rename to scratch) succeeds
+ def fake_run(command, *_a, **_kw):
+ return 1 if command == 'ip link set dev vyetheth5 name eth0' else 0
+
+ with mock.patch.object(resolver, 'run', side_effect=fake_run):
+ applied = resolver.safe_bulk_rename({'eth5': 'eth0'})
+
+ self.assertNotIn('eth5', applied)
+ self.assertNotEqual(applied.get('eth5'), 'eth0')
+
+ def test_phase_two_failure_brings_scratch_name_back_up(self):
+ calls = []
+
+ def fake_run(command, *_a, **_kw):
+ calls.append(command)
+ return 1 if command == 'ip link set dev vyetheth5 name eth0' else 0
+
+ with mock.patch.object(resolver, 'run', side_effect=fake_run):
+ resolver.safe_bulk_rename({'eth5': 'eth0'})
+
+ self.assertIn('ip link set dev vyetheth5 up', calls)
+ self.assertNotIn('ip link set dev eth0 up', calls)
+
+ def test_one_failure_does_not_affect_other_renames_in_the_batch(self):
+ def fake_run(command, *_a, **_kw):
+ return 1 if command == 'ip link set dev vyetheth5 name eth0' else 0
+
+ with mock.patch.object(resolver, 'run', side_effect=fake_run):
+ applied = resolver.safe_bulk_rename({'eth5': 'eth0', 'eth9': 'eth1'})
+
+ self.assertNotIn('eth5', applied)
+ self.assertEqual(applied.get('eth9'), 'eth1')
+
+
+class TestGetPermanentMac(unittest.TestCase):
+ """Part of the original race is reading a MAC before the driver has
+ programmed the permanent address. Verify the ethtool-first,
+ sysfs-fallback behaviour, including drivers that report an all-zero
+ 'unsupported' permanent address instead of failing outright.
+ """
+
+ def test_ethtool_permanent_address(self):
+ with mock.patch.object(
+ resolver, 'rc_cmd',
+ return_value=(0, 'Permanent address: aa:bb:cc:dd:ee:ff\n')):
+ self.assertEqual(resolver.get_permanent_mac('eth0'),
+ 'aa:bb:cc:dd:ee:ff')
+
+ def test_ethtool_zero_address_falls_back_to_sysfs(self):
+ with mock.patch.object(
+ resolver, 'rc_cmd',
+ return_value=(0, 'Permanent address: 00:00:00:00:00:00\n')), \
+ mock.patch('pathlib.Path.read_text',
+ return_value='11:22:33:44:55:66\n'):
+ self.assertEqual(resolver.get_permanent_mac('eth0'),
+ '11:22:33:44:55:66')
+
+ def test_ethtool_unsupported_falls_back_to_sysfs(self):
+ with mock.patch.object(
+ resolver, 'rc_cmd',
+ return_value=(1, 'Operation not supported')), \
+ mock.patch('pathlib.Path.read_text',
+ return_value='11:22:33:44:55:66\n'):
+ self.assertEqual(resolver.get_permanent_mac('eth0'),
+ '11:22:33:44:55:66')
+
+ def test_no_mac_available_returns_empty(self):
+ with mock.patch.object(resolver, 'rc_cmd', return_value=(1, '')), \
+ mock.patch('pathlib.Path.read_text', side_effect=OSError):
+ self.assertEqual(resolver.get_permanent_mac('eth0'), '')
+
+
+class TestDiscoverPhysicalInterfaces(unittest.TestCase):
+ """Only interfaces with a real bus 'device' link are candidates -
+ bridges/bonds/VLANs/veth must be excluded regardless of name, since
+ this replaces the racy udev DRIVERS=="?*" check with a stable,
+ post-settle sysfs read.
+ """
+
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+
+ def make_iface(name, mac=None, has_device=True):
+ path = os.path.join(self.tmp, name)
+ os.mkdir(path)
+ if has_device:
+ os.mkdir(os.path.join(path, 'device'))
+ if mac:
+ with open(os.path.join(path, 'address'), 'w') as f:
+ f.write(mac + '\n')
+
+ make_iface('eth0', mac='aa:aa:aa:aa:aa:00')
+ make_iface('eth1', mac='aa:aa:aa:aa:aa:01')
+ make_iface('br0', mac='aa:aa:aa:aa:aa:99', has_device=False)
+ make_iface('lo', has_device=False)
+
+ def test_only_physical_interfaces_with_mac(self):
+ with mock.patch.object(resolver, 'rc_cmd', return_value=(1, '')):
+ found = resolver.discover_physical_interfaces(self.tmp)
+ self.assertEqual(found, {
+ 'eth0': 'aa:aa:aa:aa:aa:00',
+ 'eth1': 'aa:aa:aa:aa:aa:01',
+ })
+ self.assertNotIn('br0', found)
+ self.assertNotIn('lo', found)
+
+
+class TestSyncRescanHints(unittest.TestCase):
+ """New/unconfigured hardware must still be surfaced to
+ vyos-interface-rescan.py, already-configured interfaces must not, and
+ hints left under a pre-rename name must not linger once the resolver
+ has moved that interface to its hw-id target.
+ """
+
+ def setUp(self):
+ self.tmp = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+ self._orig_dir = resolver.vyos_udev_dir
+ resolver.vyos_udev_dir = self.tmp
+ self.addCleanup(setattr, resolver, 'vyos_udev_dir', self._orig_dir)
+
+ def test_writes_hint_for_unconfigured_interface(self):
+ resolver.sync_rescan_hints({'eth5': 'new-mac'}, configured={})
+ hint = os.path.join(self.tmp, 'eth5')
+ self.assertTrue(os.path.isfile(hint))
+ with open(hint) as f:
+ self.assertEqual(f.read(), 'new-mac')
+
+ def test_no_hint_for_configured_interface(self):
+ resolver.sync_rescan_hints({'eth0': 'm0'}, configured={'m0': 'eth0'})
+ self.assertFalse(os.path.exists(os.path.join(self.tmp, 'eth0')))
+
+ def test_stale_hint_removed_after_rename(self):
+ # udev's cosmetic fast-path wrote a hint under the pre-rename name;
+ # the resolver then moved that interface to its hw-id target.
+ stale = os.path.join(self.tmp, 'eth7')
+ with open(stale, 'w') as f:
+ f.write('m2')
+ resolver.sync_rescan_hints({'eth10': 'm2'}, configured={'m2': 'eth10'},
+ stale_names={'eth7'})
+ self.assertFalse(os.path.exists(stale))
+
+
+
+class TestMainFirstBootBootstrap(unittest.TestCase):
+ """Direct regression guard for the gap this feature closes: main()
+ used to leave a fully-unconfigured system's interfaces at whatever
+ racy cosmetic name they already had, with no settle wait at all, and
+ freeze that into the rescan hints vyos-interface-rescan.py persists.
+ """
+
+ def setUp(self):
+ self.udev_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.udev_dir, ignore_errors=True)
+ self._orig_udev_dir = resolver.vyos_udev_dir
+ resolver.vyos_udev_dir = self.udev_dir
+ self.addCleanup(setattr, resolver, 'vyos_udev_dir', self._orig_udev_dir)
+
+ status_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, status_dir, ignore_errors=True)
+ self._orig_status_file = resolver.status_file
+ resolver.status_file = resolver.Path(status_dir) / 'status.json'
+ self.addCleanup(setattr, resolver, 'status_file', self._orig_status_file)
+
+ def test_empty_configfile_bootstrap_renames_and_hints_new_names(self):
+ # a tiny fake kernel: name -> mac, mutated by simulated 'ip link'
+ # calls so discover_physical_interfaces() reflects renames for real
+ state = {'eth5': 'bb', 'eth2': 'aa'} # cosmetic fast-path leftover
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value={}), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value={'ethernet': set(), 'wireless': set()}), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ hints = set(os.listdir(self.udev_dir))
+ # renamed into MAC-rank order: lower mac 'aa' -> eth0, higher 'bb' -> eth1
+ self.assertEqual(hints, {'eth0', 'eth1'})
+ with open(os.path.join(self.udev_dir, 'eth0')) as f:
+ self.assertEqual(f.read(), 'aa')
+ with open(os.path.join(self.udev_dir, 'eth1')) as f:
+ self.assertEqual(f.read(), 'bb')
+
+ def test_pending_node_still_settles_when_its_hardware_is_slow_to_appear(self):
+ # field report: a pending node ended up permanently bound to a
+ # fresh eth8/eth9-style name despite its own hardware genuinely
+ # being present - because that hardware (a slower driver, exactly
+ # the multi-vendor-NIC race this whole feature addresses) simply
+ # had not shown up in the FIRST, unsettled discover_physical_
+ # interfaces() snapshot wait_for_hardware() returns as soon as all
+ # already-CONFIGURED macs are found. Since that snapshot showed
+ # nothing unconfigured yet, main()'s gate into wait_for_settle()
+ # never even ran, giving the slow hardware zero extra time to
+ # appear before the boot gave up. A pending node in play must
+ # force wait_for_settle() to run regardless of what the early
+ # snapshot shows, so slower hardware still gets its chance.
+ configured = {'m0': 'eth0'}
+ pending = {'ethernet': {'eth1'}, 'wireless': set()}
+
+ calls = {'n': 0}
+
+ def fake_discover(*_a, **_kw):
+ calls['n'] += 1
+ if calls['n'] <= 2:
+ return {'eth0': 'm0'} # m1's hardware not visible yet
+ return {'eth0': 'm0', 'eth9': 'm1'} # now it has appeared
+
+ state = {'eth0': 'm0', 'eth9': 'm1'}
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ self.assertEqual(state.get('eth1'), 'm1')
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['reclaimed'], {'m1': 'eth1'})
+ self.assertEqual(status['pending_unresolved'], [])
+
+ def test_reported_reproduction_end_to_end(self):
+ # exact reported scenario: eth0/eth2 keep their hw-id, eth1's hw-id
+ # was deleted (the documented "NIC replaced" remediation) but its
+ # node - and an address setting - stay in config.boot. The same
+ # physical NIC (mac 'm1') must come back as 'eth1', not bootstrap
+ # past the floor to 'eth3', and must still get a rescan hint under
+ # 'eth1' so vyos-interface-rescan.py can refill its hw-id.
+ configured = {'m0': 'eth0', 'm2': 'eth2'}
+ pending = {'ethernet': {'eth1'}, 'wireless': set()}
+ # cosmetic fast-path already placed everything at its final
+ # hw-id'd name except the replaced NIC, which landed on a racy name
+ state = {'eth0': 'm0', 'eth2': 'm2', 'eth9': 'm1'}
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ # the NIC is back at 'eth1', not bootstrapped past the floor to 'eth3'
+ self.assertEqual(state.get('eth1'), 'm1')
+ self.assertNotIn('eth3', state)
+
+ # still gets a rescan hint under its reclaimed name, so
+ # vyos-interface-rescan.py can write the real hw-id into the
+ # existing 'eth1' node (preserving its other settings, e.g. address)
+ hints = set(os.listdir(self.udev_dir))
+ self.assertEqual(hints, {'eth1'})
+ with open(os.path.join(self.udev_dir, 'eth1')) as f:
+ self.assertEqual(f.read(), 'm1')
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['reclaimed'], {'m1': 'eth1'})
+ self.assertEqual(status['pending_unresolved'], [])
+
+ def test_fully_deleted_interfaces_backfill_their_gaps_end_to_end(self):
+ # reported scenario: eth3, eth6 and eth7 were fully deleted (not
+ # just their hw-id - the whole node, so get_pending_hwid_nodes()
+ # has nothing for them either) while eth0,1,2,4,5,8,9 stay
+ # configured. The three now-unconfigured NICs must backfill the
+ # gaps at 3/6/7 in MAC order, not bootstrap past the highest
+ # configured index (9) to 10/11/12.
+ configured = {
+ 'm0': 'eth0', 'm1': 'eth1', 'm2': 'eth2', 'm4': 'eth4',
+ 'm5': 'eth5', 'm8': 'eth8', 'm9': 'eth9',
+ }
+ pending = {'ethernet': set(), 'wireless': set()}
+ state = {name: mac for mac, name in configured.items()}
+ # realistic macs, deliberately NOT alphabetically matching their
+ # old slot, so the test can't pass by coincidence
+ state.update({'ethX': 'ff:ff:ff:ff:ff:07',
+ 'ethY': 'aa:aa:aa:aa:aa:03',
+ 'ethZ': 'cc:cc:cc:cc:cc:06'})
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ # the three gaps got backfilled in ascending MAC order, not pushed
+ # past the highest configured index to eth10/eth11/eth12
+ self.assertEqual(state.get('eth3'), 'aa:aa:aa:aa:aa:03')
+ self.assertEqual(state.get('eth6'), 'cc:cc:cc:cc:cc:06')
+ self.assertEqual(state.get('eth7'), 'ff:ff:ff:ff:ff:07')
+ self.assertNotIn('eth10', state)
+ self.assertNotIn('eth11', state)
+ self.assertNotIn('eth12', state)
+
+ def test_stray_leftover_interface_still_fills_both_open_names(self):
+ # field report: deleting only eth7's hw-id and rebooting produced
+ # "could not be safely auto-matched" instead of a clean reclaim,
+ # because a stray interface left over from an earlier, unrelated
+ # boot (e.g. a scratch vyethN name stuck after a failed rename)
+ # was also present, making this a 1 pending/2 candidate case. Both
+ # names now reliably get a real hw-id either way - the lower-
+ # sorted candidate takes the lower-numbered open name (the
+ # ordinary gap at eth3), the higher-sorted one takes the pending
+ # node - nothing is left unresolved, and there is no way to know
+ # (or need to know) which candidate was "really" eth7's own past
+ # hardware once its hw-id is gone.
+ configured = {
+ 'm0': 'eth0', 'm1': 'eth1', 'm2': 'eth2', 'm4': 'eth4',
+ 'm5': 'eth5', 'm6': 'eth6',
+ }
+ pending = {'ethernet': {'eth7'}, 'wireless': set()}
+ state = {name: mac for mac, name in configured.items()}
+ state.update({'eth9': 'aa:bb:cc:dd:ee:07', # lower-sorted candidate
+ 'vyeth13': 'ff:ff:ff:ff:ff:99'}) # higher-sorted candidate
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ # the lower-sorted candidate fills the ordinary gap (eth3), the
+ # higher-sorted one fills the pending node (eth7) - both real,
+ # both deterministic, nothing left unresolved
+ self.assertEqual(state.get('eth3'), 'aa:bb:cc:dd:ee:07')
+ self.assertEqual(state.get('eth7'), 'ff:ff:ff:ff:ff:99')
+
+ hints = set(os.listdir(self.udev_dir))
+ self.assertEqual(hints, {'eth3', 'eth7'})
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['pending_unresolved'], [])
+ self.assertEqual(status['reclaimed'], {'ff:ff:ff:ff:ff:99': 'eth7'})
+
+ def test_second_pending_node_fills_lowest_numbered_first_when_hardware_is_short(self):
+ # two pending nodes (eth1 and eth7), but only one candidate showed
+ # up this boot - a hardware shortage, not ambiguity. The lower-
+ # numbered node fills; the higher-numbered one stays genuinely
+ # unresolved and reported, since there simply isn't enough
+ # hardware to satisfy both.
+ configured = {'m0': 'eth0'}
+ pending = {'ethernet': {'eth1', 'eth7'}, 'wireless': set()}
+ state = {'eth0': 'm0', 'eth9': 'aa:bb:cc:dd:ee:07'}
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ self.assertEqual(state.get('eth1'), 'aa:bb:cc:dd:ee:07')
+ self.assertNotIn('eth7', state)
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['pending_unresolved'], ['eth7'])
+ self.assertEqual(status['reclaimed'], {'aa:bb:cc:dd:ee:07': 'eth1'})
+
+ def test_two_candidates_for_one_pending_node_fill_deterministically(self):
+ # field report: deleting eth2's hw-id (leaving its node in place)
+ # produced "0 unconfigured candidates" on a later boot, because an
+ # EARLIER boot with 2 candidates present had left the pending node
+ # unresolved and its rightful hardware un-hinted, orphaning it.
+ # Per the deterministic-fill policy, one pending node with two
+ # candidates now reliably resolves: with only 'eth0' configured,
+ # 'eth1' is also a genuinely open name here, so the lower-sorted
+ # candidate fills it first and the higher-sorted one fills the
+ # pending node ('eth2') - nothing is left unresolved or un-hinted
+ # either way.
+ configured = {'m0': 'eth0'}
+ pending = {'ethernet': {'eth2'}, 'wireless': set()}
+ state = {
+ 'eth0': 'm0',
+ 'racyA': 'aa:bb:cc:dd:ee:02', # lower-sorted candidate
+ 'racyB': 'ff:ff:ff:ff:ff:99', # higher-sorted candidate
+ }
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ # the lower-sorted candidate (racyA) fills the lower open name
+ # (eth1); the higher-sorted one (racyB) reclaims the pending node
+ self.assertEqual(state.get('eth1'), 'aa:bb:cc:dd:ee:02')
+ self.assertEqual(state.get('eth2'), 'ff:ff:ff:ff:ff:99')
+
+ hints = set(os.listdir(self.udev_dir))
+ self.assertEqual(hints, {'eth1', 'eth2'})
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['pending_unresolved'], [])
+ self.assertEqual(status['reclaimed'], {'ff:ff:ff:ff:ff:99': 'eth2'})
+
+ def test_two_pending_nodes_two_candidates_both_fill_deterministically(self):
+ # two NICs' hw-id were deleted before the same reboot (eth0 and
+ # eth2, nodes left in place), and both came back racily named.
+ # Sorted pending names (eth0 < eth2) pair with sorted candidates
+ # (by MAC here, since pcie_distance ties) - both fill in one boot,
+ # with no unresolved node left behind.
+ configured = {
+ 'm1': 'eth1', 'm3': 'eth3', 'm4': 'eth4',
+ 'm5': 'eth5', 'm6': 'eth6', 'm7': 'eth7',
+ }
+ pending = {'ethernet': {'eth0', 'eth2'}, 'wireless': set()}
+ state = {name: mac for mac, name in configured.items()}
+ state.update({
+ 'racyA': 'aa:bb:cc:dd:ee:00',
+ 'racyB': 'aa:bb:cc:dd:ee:02',
+ })
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ self.assertEqual(state.get('eth0'), 'aa:bb:cc:dd:ee:00')
+ self.assertEqual(state.get('eth2'), 'aa:bb:cc:dd:ee:02')
+
+ hints = set(os.listdir(self.udev_dir))
+ self.assertEqual(hints, {'eth0', 'eth2'})
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['pending_unresolved'], [])
+ self.assertEqual(status['reclaimed'], {
+ 'aa:bb:cc:dd:ee:00': 'eth0',
+ 'aa:bb:cc:dd:ee:02': 'eth2',
+ })
+
+ def test_squatting_pending_candidate_still_reclaims_correctly(self):
+ # field report: deleting eth7's hw-id (node left in place) produced
+ # a "0 unconfigured candidates" boot even though eth7's real
+ # hardware was present, and it later turned up permanently bound
+ # to a fresh eth8/eth9-style name. Root cause: eth7's hardware
+ # (mac07) happened to be squatting on eth2's configured hw-id slot
+ # this boot (probe-order scrambling) - compute_rename_plan()
+ # already scheduled it for eviction, which used to make
+ # unmatched_candidates() skip it entirely, so it never reached the
+ # ascending fill and fell through to a stale, unrelated fallback
+ # destination instead. A squatting candidate must still reclaim
+ # its pending node exactly like a non-squatting one would, when it
+ # is the only open name and the only candidate this boot.
+ configured = {
+ 'm0': 'eth0', 'm1': 'eth1', 'm2': 'eth2', 'm3': 'eth3',
+ 'm4': 'eth4', 'm5': 'eth5', 'm6': 'eth6',
+ }
+ pending = {'ethernet': {'eth7'}, 'wireless': set()}
+ state = {name: mac for mac, name in configured.items()}
+ state.update({
+ 'eth2': 'aa:bb:cc:dd:ee:07', # eth7's hardware squats on eth2
+ 'eth9': 'm2', # m2's rightful NIC, racy this boot
+ })
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ # m2 still lands on its own configured slot, and the squatter
+ # reclaims 'eth7' - not some unrelated fresh bootstrap slot
+ self.assertEqual(state.get('eth2'), 'm2')
+ self.assertEqual(state.get('eth7'), 'aa:bb:cc:dd:ee:07')
+ self.assertNotIn('eth8', state)
+ self.assertNotIn('eth9', state)
+
+ hints = set(os.listdir(self.udev_dir))
+ self.assertEqual(hints, {'eth7'})
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['reclaimed'], {'aa:bb:cc:dd:ee:07': 'eth7'})
+ self.assertEqual(status['pending_unresolved'], [])
+
+ def test_squatting_candidate_and_extra_candidate_fill_ascending(self):
+ # same squatter-eviction shape as the test above, but with a
+ # second, unrelated candidate also present this boot, and exactly
+ # two open names (one ordinary gap, one pending node) for the two
+ # candidates - a closed system. The lower-sorted candidate (the
+ # squatter, evicted off of a configured slot) takes the lower-
+ # numbered open name (the gap); the higher-sorted one takes the
+ # higher-numbered one (the pending node) - both get a real,
+ # deterministic hw-id, one of them attributed as a reclaim.
+ configured = {
+ 'm0': 'eth0', 'm1': 'eth1', 'm2': 'eth2',
+ 'm3': 'eth3', 'm4': 'eth4', 'm5': 'eth5',
+ }
+ pending = {'ethernet': {'eth7'}, 'wireless': set()}
+ state = {name: mac for mac, name in configured.items()}
+ state.update({
+ 'eth3': 'aa:bb:cc:dd:ee:07', # squats on m3's configured slot
+ 'eth9': 'm3', # m3's rightful NIC, racy this boot
+ 'racyC': 'ff:ff:ff:ff:ff:99', # unrelated leftover candidate
+ })
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ # m3 still lands on its own configured slot; the squatter takes
+ # the lower-numbered open gap, the stray takes the pending node
+ self.assertEqual(state.get('eth3'), 'm3')
+ self.assertEqual(state.get('eth6'), 'aa:bb:cc:dd:ee:07')
+ self.assertEqual(state.get('eth7'), 'ff:ff:ff:ff:ff:99')
+
+ hints = set(os.listdir(self.udev_dir))
+ self.assertEqual(hints, {'eth6', 'eth7'})
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['pending_unresolved'], [])
+ self.assertEqual(status['reclaimed'], {'ff:ff:ff:ff:ff:99': 'eth7'})
+
+ def test_gap_backfill_and_pending_reclaim_coexist_in_the_same_boot(self):
+ # the exact vyos-build check-qemu-install --ifnametest shape: one
+ # interface's whole config node is fully deleted (a free numeric
+ # gap) while a DIFFERENT interface's hw-id alone is deleted (a
+ # pending node) in the very same reboot
+ # (del_idx, hwid_idx = random.sample(range(8), 2)). Both must
+ # resolve independently in one boot: the pending node (eth2)
+ # reliably gets a real hw-id, and the fully-deleted gap (eth5)
+ # backfills with whatever hardware is left - no manual step
+ # needed for either.
+ configured = {
+ 'm0': 'eth0', 'm1': 'eth1', 'm3': 'eth3', 'm4': 'eth4',
+ 'm6': 'eth6', 'm8': 'eth8', 'm9': 'eth9',
+ }
+ pending = {'ethernet': {'eth2'}, 'wireless': set()}
+ state = {name: mac for mac, name in configured.items()}
+ state.update({
+ 'eth7': 'aa:bb:cc:dd:ee:02', # eth2's real hardware, racy this boot
+ 'eth5': 'ff:ff:ff:ff:ff:05', # eth5 fully deleted, its own hardware freed
+ })
+
+ def fake_discover(*_a, **_kw):
+ return dict(state)
+
+ def fake_run(command, *_a, **_kw):
+ parts = command.split()
+ if 'name' in parts:
+ old = parts[parts.index('dev') + 1]
+ new = parts[parts.index('name') + 1]
+ if old in state:
+ state[new] = state.pop(old)
+ return 0
+
+ with mock.patch.object(resolver, 'get_configfile_interfaces',
+ return_value=configured), \
+ mock.patch.object(resolver, 'get_pending_hwid_nodes',
+ return_value=pending), \
+ mock.patch.object(resolver, 'discover_physical_interfaces',
+ side_effect=fake_discover), \
+ mock.patch.object(resolver, 'is_wireless_interface',
+ return_value=False), \
+ mock.patch.object(resolver, 'pcie_distance', return_value=0), \
+ mock.patch.object(resolver, 'run', side_effect=fake_run), \
+ mock.patch('time.sleep'):
+ resolver.main()
+
+ self.assertEqual(state.get('eth2'), 'aa:bb:cc:dd:ee:02')
+ self.assertEqual(state.get('eth5'), 'ff:ff:ff:ff:ff:05')
+
+ status = json.loads(resolver.status_file.read_text())
+ self.assertEqual(status['pending_unresolved'], [])
+ self.assertEqual(status['reclaimed'], {'aa:bb:cc:dd:ee:02': 'eth2'})
+
+
+class TestWriteStatus(unittest.TestCase):
+ """A pending node that couldn't be safely auto-matched must be
+ reported so vyos-router can surface a boot-time warning - it must
+ never regress into another silent, buried-syslog-only failure like
+ the one this whole fix closes.
+ """
+
+ def setUp(self):
+ status_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, status_dir, ignore_errors=True)
+ self._orig_status_file = resolver.status_file
+ resolver.status_file = resolver.Path(status_dir) / 'status.json'
+ self.addCleanup(setattr, resolver, 'status_file', self._orig_status_file)
+
+ def _read_status(self):
+ return json.loads(resolver.status_file.read_text())
+
+ def test_pending_unresolved_reported_when_no_match(self):
+ pending = {'ethernet': {'eth1', 'eth4'}, 'wireless': set()}
+ resolver.write_status({}, {}, set(), {}, pending=pending, reclaimed={})
+ status = self._read_status()
+ self.assertEqual(status['pending_unresolved'], ['eth1', 'eth4'])
+
+ def test_reclaimed_not_reported_as_unresolved(self):
+ pending = {'ethernet': {'eth1'}, 'wireless': set()}
+ reclaimed = {'m1': 'eth1'}
+ resolver.write_status({}, {}, set(), {}, pending=pending,
+ reclaimed=reclaimed)
+ status = self._read_status()
+ self.assertEqual(status['pending_unresolved'], [])
+ self.assertEqual(status['reclaimed'], {'m1': 'eth1'})
+
+ def test_no_pending_no_crash(self):
+ resolver.write_status({}, {}, set(), {})
+ status = self._read_status()
+ self.assertEqual(status['pending_unresolved'], [])
+ self.assertEqual(status['reclaimed'], {})
+
+ def test_unconfigured_candidates_reported_for_diagnosis(self):
+ # the full unmatched_candidates() list from main(), so an
+ # ambiguous reclaim (zero or multiple candidates for a pending
+ # node) is diagnosable from this one file alone - which physical
+ # interfaces were actually competing - without needing a separate
+ # `ip -br link show` or `show configuration commands` to
+ # reconstruct the same picture by hand.
+ pending = {'ethernet': {'eth7'}, 'wireless': set()}
+ candidates = [('aa:bb:cc:dd:ee:07', 'eth9'), ('aa:bb:cc:dd:ee:99', 'vyeth13')]
+ resolver.write_status({}, {}, set(), {}, pending=pending,
+ reclaimed={}, candidates=candidates)
+ status = self._read_status()
+ self.assertEqual(status['pending_unresolved'], ['eth7'])
+ self.assertEqual(status['unconfigured_candidates'], {
+ 'eth9': 'aa:bb:cc:dd:ee:07',
+ 'vyeth13': 'aa:bb:cc:dd:ee:99',
+ })
+
+ def test_unconfigured_candidates_defaults_empty(self):
+ resolver.write_status({}, {}, set(), {})
+ status = self._read_status()
+ self.assertEqual(status['unconfigured_candidates'], {})
+
+
+class TestBootOrdering(unittest.TestCase):
+ """Regression guard for the config.boot availability corner case:
+ /opt/vyatta/etc/config/config.boot does not exist until vyos-router
+ mounts (and, for an encrypted config volume, decrypts) the config
+ directory itself. vyos-net-name-resolve must be started explicitly
+ from inside vyos-router after that point, and must not be
+ reintroduced as an independently-scheduled unit ordered before
+ vyos-router/network-pre.target/cloud-init.
+ """
+
+ def setUp(self):
+ path = os.path.join(_here, '../init/vyos-router')
+ with open(path) as f:
+ self.lines = f.readlines()
+
+ def _first_index(self, needle):
+ for i, line in enumerate(self.lines):
+ if needle in line:
+ return i
+ self.fail(f"'{needle}' not found in src/init/vyos-router")
+
+ def test_resolver_runs_after_migrate_and_before_config_apply(self):
+ # search for call SITES, not the function definitions further up
+ migrate = self._first_index('migrate_bootfile || overall_status=1')
+ resolve = self._first_index(
+ 'systemctl start vyos-net-name-resolve.service')
+ update_iface = self._first_index('update_interface_config || overall_status=1')
+ load_boot = self._first_index('disabled configure || load_bootfile')
+
+ self.assertLess(migrate, resolve,
+ 'resolver must run after config.boot is migrated/current')
+ self.assertLess(resolve, update_iface,
+ 'resolver must run before new-hardware rescan consumes its hints')
+ self.assertLess(resolve, load_boot,
+ 'resolver must run before the CLI config is applied')
+
+ def test_service_unit_not_independently_ordered_before_vyos_router(self):
+ path = os.path.join(_here, '../systemd/vyos-net-name-resolve.service')
+ with open(path) as f:
+ unit = f.read()
+ self.assertNotIn('Before=vyos-router.service', unit)
+ self.assertNotIn('Before=network-pre.target', unit)
+ self.assertNotIn('WantedBy=', unit)
+
+
+if __name__ == '__main__':
+ unittest.main()