0.7.9-89-gbf7723e8-0ubuntu1 (patches unapplied)

Imported using git-ubuntu import.

3 files changed, 95 insertions, 4 deletions

diff --git a/tests/unittests/test_datasource/test_digitalocean.py b/tests/unittests/test_datasource/test_digitalocean.py
index 9be6bc19..61d6e001 100644
--- a/tests/unittests/test_datasource/test_digitalocean.py
+++ b/tests/unittests/test_datasource/test_digitalocean.py
@@ -194,7 +194,12 @@ class TestDataSourceDigitalOcean(TestCase):
 
 class TestNetworkConvert(TestCase):
 
-    def _get_networking(self):
+    @mock.patch('cloudinit.net.get_interfaces_by_mac')
+    def _get_networking(self, m_get_by_mac):
+        m_get_by_mac.return_value = {
+            '04:01:57:d1:9e:01': 'ens1', '04:01:57:d1:9e:02': 'ens2',
+            'b8:ae:ed:75:5f:9a': 'enp0s25',
+            'ae:cc:08:7c:88:00': 'meta2p1'}
         netcfg = digitalocean.convert_network_configuration(
             DO_META['interfaces'], DO_META['dns']['nameservers'])
         self.assertIn('config', netcfg)
@@ -302,10 +307,15 @@ class TestNetworkConvert(TestCase):
         self.assertEqual(ipv4_def.get('netmask'), subn_def.get('netmask'))
         self.assertNotIn('gateway', subn_def)
 
-    def test_convert_without_private(self):
+    @mock.patch('cloudinit.net.get_interfaces_by_mac')
+    def test_convert_without_private(self, m_get_by_mac):
+        m_get_by_mac.return_value = {
+            'b8:ae:ed:75:5f:9a': 'enp0s25',
+            'ae:cc:08:7c:88:00': 'meta2p1'}
         netcfg = digitalocean.convert_network_configuration(
             DO_META_2['interfaces'], DO_META_2['dns']['nameservers'])
 
+        # print(netcfg)
         byname = {}
         for i in netcfg['config']:
             if 'name' in i:
diff --git a/tests/unittests/test_datasource/test_opennebula.py b/tests/unittests/test_datasource/test_opennebula.py
index a266e952..bce66125 100644
--- a/tests/unittests/test_datasource/test_opennebula.py
+++ b/tests/unittests/test_datasource/test_opennebula.py
@@ -195,7 +195,9 @@ class TestOpenNebulaDataSource(TestCase):
             self.assertTrue('userdata' in results)
             self.assertEqual(USER_DATA, results['userdata'])
 
-    def test_hostname(self):
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_hostname(self, m_get_phys_by_mac):
+        m_get_phys_by_mac.return_value = {'02:00:0a:12:01:01': 'eth0'}
         for k in ('HOSTNAME', 'PUBLIC_IP', 'IP_PUBLIC', 'ETH0_IP'):
             my_d = os.path.join(self.tmp, k)
             populate_context_dir(my_d, {k: PUBLIC_IP})
@@ -205,11 +207,14 @@ class TestOpenNebulaDataSource(TestCase):
             self.assertTrue('local-hostname' in results['metadata'])
             self.assertEqual(PUBLIC_IP, results['metadata']['local-hostname'])
 
-    def test_network_interfaces(self):
+    @mock.patch(DS_PATH + ".get_physical_nics_by_mac")
+    def test_network_interfaces(self, m_get_phys_by_mac):
+        m_get_phys_by_mac.return_value = {'02:00:0a:12:01:01': 'eth0'}
         populate_context_dir(self.seed_dir, {'ETH0_IP': '1.2.3.4'})
         results = ds.read_context_disk_dir(self.seed_dir)
 
         self.assertTrue('network-interfaces' in results)
+        self.assertTrue('1.2.3.4' in results['network-interfaces'])
 
     def test_find_candidates(self):
         def my_devs_with(criteria):
diff --git a/tests/unittests/test_net.py b/tests/unittests/test_net.py
index bfd04ba0..9cc5e4ab 100644
--- a/tests/unittests/test_net.py
+++ b/tests/unittests/test_net.py
@@ -1461,6 +1461,82 @@ class TestNetRenderers(CiTestCase):
                           priority=['sysconfig', 'eni'])
 
 
+class TestGetInterfacesByMac(CiTestCase):
+    _data = {'devices': ['enp0s1', 'enp0s2', 'bond1', 'bridge1',
+                         'bridge1-nic', 'tun0'],
+             'bonds': ['bond1'],
+             'bridges': ['bridge1'],
+             'own_macs': ['enp0s1', 'enp0s2', 'bridge1-nic', 'bridge1'],
+             'macs': {'enp0s1': 'aa:aa:aa:aa:aa:01',
+                      'enp0s2': 'aa:aa:aa:aa:aa:02',
+                      'bond1': 'aa:aa:aa:aa:aa:01',
+                      'bridge1': 'aa:aa:aa:aa:aa:03',
+                      'bridge1-nic': 'aa:aa:aa:aa:aa:03',
+                      'tun0': None}}
+    data = {}
+
+    def _se_get_devicelist(self):
+        return self.data['devices']
+
+    def _se_get_interface_mac(self, name):
+        return self.data['macs'][name]
+
+    def _se_is_bridge(self, name):
+        return name in self.data['bridges']
+
+    def _se_interface_has_own_mac(self, name):
+        return name in self.data['own_macs']
+
+    def _mock_setup(self):
+        self.data = copy.deepcopy(self._data)
+        mocks = ('get_devicelist', 'get_interface_mac', 'is_bridge',
+                 'interface_has_own_mac')
+        self.mocks = {}
+        for n in mocks:
+            m = mock.patch('cloudinit.net.' + n,
+                           side_effect=getattr(self, '_se_' + n))
+            self.addCleanup(m.stop)
+            self.mocks[n] = m.start()
+
+    def test_raise_exception_on_duplicate_macs(self):
+        self._mock_setup()
+        self.data['macs']['bridge1-nic'] = self.data['macs']['enp0s1']
+        self.assertRaises(RuntimeError, net.get_interfaces_by_mac)
+
+    def test_excludes_any_without_mac_address(self):
+        self._mock_setup()
+        ret = net.get_interfaces_by_mac()
+        self.assertIn('tun0', self._se_get_devicelist())
+        self.assertNotIn('tun0', ret.values())
+
+    def test_excludes_stolen_macs(self):
+        self._mock_setup()
+        ret = net.get_interfaces_by_mac()
+        self.mocks['interface_has_own_mac'].assert_has_calls(
+            [mock.call('enp0s1'), mock.call('bond1')], any_order=True)
+        self.assertEqual(
+            {'aa:aa:aa:aa:aa:01': 'enp0s1', 'aa:aa:aa:aa:aa:02': 'enp0s2',
+             'aa:aa:aa:aa:aa:03': 'bridge1-nic'},
+            ret)
+
+    def test_excludes_bridges(self):
+        self._mock_setup()
+        # add a device 'b1', make all return they have their "own mac",
+        # set everything other than 'b1' to be a bridge.
+        # then expect b1 is the only thing left.
+        self.data['macs']['b1'] = 'aa:aa:aa:aa:aa:b1'
+        self.data['devices'].append('b1')
+        self.data['bonds'] = []
+        self.data['own_macs'] = self.data['devices']
+        self.data['bridges'] = [f for f in self.data['devices'] if f != "b1"]
+        ret = net.get_interfaces_by_mac()
+        self.assertEqual({'aa:aa:aa:aa:aa:b1': 'b1'}, ret)
+        self.mocks['is_bridge'].assert_has_calls(
+            [mock.call('bridge1'), mock.call('enp0s1'), mock.call('bond1'),
+             mock.call('b1')],
+            any_order=True)
+
+
 def _gzip_data(data):
     with io.BytesIO() as iobuf:
         gzfp = gzip.GzipFile(mode="wb", fileobj=iobuf)