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.