net: exclude OVS internal interfaces in get_interfaces (#829)

`get_interfaces` is used to in two ways, broadly: firstly, to determine
the available interfaces when converting cloud network configuration
formats to cloud-init's network configuration formats; and, secondly, to
ensure that any interfaces which are specified in network configuration
are (a) available, and (b) named correctly.  The first of these is
unaffected by this commit, as no clouds support Open vSwitch
configuration in their network configuration formats.

For the second, we check that MAC addresses of physical devices are
unique.  In some OVS configurations, there are OVS-created devices which
have duplicate MAC addresses, either with each other or with physical
devices.  As these interfaces are created by OVS, we can be confident
that (a) they will be available when appropriate, and (b) that OVS will
name them correctly.  As such, this commit excludes any OVS-internal
interfaces from the set of interfaces returned by `get_interfaces`.

LP: #1912844

1 files changed, 119 insertions, 0 deletions

diff --git a/cloudinit/net/tests/test_init.py b/cloudinit/net/tests/test_init.py
index 0535387a..946f8ee2 100644
--- a/cloudinit/net/tests/test_init.py
+++ b/cloudinit/net/tests/test_init.py
@@ -391,6 +391,10 @@ class TestGetDeviceList(CiTestCase):
         self.assertCountEqual(['eth0', 'eth1'], net.get_devicelist())
 
 
+@mock.patch(
+    "cloudinit.net.is_openvswitch_internal_interface",
+    mock.Mock(return_value=False),
+)
 class TestGetInterfaceMAC(CiTestCase):
 
     def setUp(self):
@@ -1224,6 +1228,121 @@ class TestNetFailOver(CiTestCase):
         self.assertFalse(net.is_netfailover(devname, driver))
 
 
+class TestOpenvswitchIsInstalled:
+    """Test cloudinit.net.openvswitch_is_installed.
+
+    Uses the ``clear_lru_cache`` local autouse fixture to allow us to test
+    despite the ``lru_cache`` decorator on the unit under test.
+    """
+
+    @pytest.fixture(autouse=True)
+    def clear_lru_cache(self):
+        net.openvswitch_is_installed.cache_clear()
+
+    @pytest.mark.parametrize(
+        "expected,which_return", [(True, "/some/path"), (False, None)]
+    )
+    @mock.patch("cloudinit.net.subp.which")
+    def test_mirrors_which_result(self, m_which, expected, which_return):
+        m_which.return_value = which_return
+        assert expected == net.openvswitch_is_installed()
+
+    @mock.patch("cloudinit.net.subp.which")
+    def test_only_calls_which_once(self, m_which):
+        net.openvswitch_is_installed()
+        net.openvswitch_is_installed()
+        assert 1 == m_which.call_count
+
+
+@mock.patch("cloudinit.net.subp.subp", return_value=("", ""))
+class TestGetOVSInternalInterfaces:
+    """Test cloudinit.net.get_ovs_internal_interfaces.
+
+    Uses the ``clear_lru_cache`` local autouse fixture to allow us to test
+    despite the ``lru_cache`` decorator on the unit under test.
+    """
+    @pytest.fixture(autouse=True)
+    def clear_lru_cache(self):
+        net.get_ovs_internal_interfaces.cache_clear()
+
+    def test_command_used(self, m_subp):
+        """Test we use the correct command when we call subp"""
+        net.get_ovs_internal_interfaces()
+
+        assert [
+            mock.call(net.OVS_INTERNAL_INTERFACE_LOOKUP_CMD)
+        ] == m_subp.call_args_list
+
+    def test_subp_contents_split_and_returned(self, m_subp):
+        """Test that the command output is appropriately mangled."""
+        stdout = "iface1\niface2\niface3\n"
+        m_subp.return_value = (stdout, "")
+
+        assert [
+            "iface1",
+            "iface2",
+            "iface3",
+        ] == net.get_ovs_internal_interfaces()
+
+    def test_database_connection_error_handled_gracefully(self, m_subp):
+        """Test that the error indicating OVS is down is handled gracefully."""
+        m_subp.side_effect = ProcessExecutionError(
+            stderr="database connection failed"
+        )
+
+        assert [] == net.get_ovs_internal_interfaces()
+
+    def test_other_errors_raised(self, m_subp):
+        """Test that only database connection errors are handled."""
+        m_subp.side_effect = ProcessExecutionError()
+
+        with pytest.raises(ProcessExecutionError):
+            net.get_ovs_internal_interfaces()
+
+    def test_only_runs_once(self, m_subp):
+        """Test that we cache the value."""
+        net.get_ovs_internal_interfaces()
+        net.get_ovs_internal_interfaces()
+
+        assert 1 == m_subp.call_count
+
+
+@mock.patch("cloudinit.net.get_ovs_internal_interfaces")
+@mock.patch("cloudinit.net.openvswitch_is_installed")
+class TestIsOpenVSwitchInternalInterface:
+    def test_false_if_ovs_not_installed(
+        self, m_openvswitch_is_installed, _m_get_ovs_internal_interfaces
+    ):
+        """Test that OVS' absence returns False."""
+        m_openvswitch_is_installed.return_value = False
+
+        assert not net.is_openvswitch_internal_interface("devname")
+
+    @pytest.mark.parametrize(
+        "detected_interfaces,devname,expected_return",
+        [
+            ([], "devname", False),
+            (["notdevname"], "devname", False),
+            (["devname"], "devname", True),
+            (["some", "other", "devices", "and", "ours"], "ours", True),
+        ],
+    )
+    def test_return_value_based_on_detected_interfaces(
+        self,
+        m_openvswitch_is_installed,
+        m_get_ovs_internal_interfaces,
+        detected_interfaces,
+        devname,
+        expected_return,
+    ):
+        """Test that the detected interfaces are used correctly."""
+        m_openvswitch_is_installed.return_value = True
+        m_get_ovs_internal_interfaces.return_value = detected_interfaces
+        assert expected_return == net.is_openvswitch_internal_interface(
+            devname
+        )
+
+
 class TestIsIpAddress:
     """Tests for net.is_ip_address.