network: don't run netplan apply if no autoinstall and nm is enabled

Signed-off-by: Olivier Gayot <olivier.gayot@canonical.com>
Co-authored-by: Dan Bungert <daniel.bungert@canonical.com>
(cherry picked from commit 9d32ded2f9)

subiquity/models/network.py

@@ -14,8 +14,10 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import logging
+import subprocess
 
 from subiquitycore.models.network import NetworkModel as CoreNetworkModel
+from subiquitycore.utils import arun_command
 
 log = logging.getLogger('subiquity.models.network')
 
@@ -72,3 +74,16 @@ class NetworkModel(CoreNetworkModel):
             return ['wpasupplicant']
         else:
             return []
+
+    async def is_nm_enabled(self):
+        try:
+            cp = await arun_command(("nmcli", "networking"), check=True)
+        except subprocess.CalledProcessError as exc:
+            log.warning("failed to run nmcli networking,"
+                        " considering NetworkManager disabled.")
+            log.debug("stderr: %s", exc.stderr)
+            return False
+        except FileNotFoundError:
+            return False
+
+        return cp.stdout == "enabled\n"

subiquity/models/tests/test_network.py

@@ -0,0 +1,47 @@
+# Copyright 2023 Canonical, Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# 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 Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import subprocess
+import unittest
+from unittest import mock
+
+from subiquity.models.network import (
+    NetworkModel,
+)
+
+
+class TestNetworkModel(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.model = NetworkModel()
+
+    async def test_is_nm_enabled(self):
+        with mock.patch("subiquity.models.network.arun_command") as arun:
+            arun.return_value = subprocess.CompletedProcess([], 0)
+            arun.return_value.stdout = "enabled\n"
+            self.assertTrue(await self.model.is_nm_enabled())
+
+        with mock.patch("subiquity.models.network.arun_command") as arun:
+            arun.return_value = subprocess.CompletedProcess([], 0)
+            arun.return_value.stdout = "disabled\n"
+            self.assertFalse(await self.model.is_nm_enabled())
+
+        with mock.patch("subiquity.models.network.arun_command") as arun:
+            arun.side_effect = FileNotFoundError
+            self.assertFalse(await self.model.is_nm_enabled())
+
+        with mock.patch("subiquity.models.network.arun_command") as arun:
+            arun.side_effect = subprocess.CalledProcessError(
+                    1, [], None, "error")
+            self.assertFalse(await self.model.is_nm_enabled())

subiquity/server/controllers/network.py

@@ -239,13 +239,26 @@ class NetworkController(BaseNetworkController, SubiquityController):
 
     @with_context()
     async def apply_autoinstall_config(self, context):
+        want_apply_config = True
         if self.ai_data is None:
-            with context.child("wait_initial_config"):
-                await self.initial_config
-            self.update_initial_configs()
-            self.apply_config(context)
-        with context.child("wait_for_apply"):
-            await self.apply_config_task.wait()
+            if not await self.model.is_nm_enabled():
+                with context.child("wait_initial_config"):
+                    await self.initial_config
+                self.update_initial_configs()
+                self.apply_config(context)
+            else:
+                log.debug("NetworkManager is enabled and no network"
+                          " autoinstall section was found. Not applying"
+                          " network settings.")
+                want_apply_config = False
+        if want_apply_config:
+            with context.child("wait_for_apply"):
+                await self.apply_config_task.wait()
+        else:
+            # Make sure we have read at least once the routing table.
+            # Careful, the following is a blocking call. But running it in a
+            # separate thread without locking sounds unsafe too.
+            self.network_event_receiver.probe_default_routes()
         self.model.has_network = self.network_event_receiver.has_default_route
 
     async def _apply_config(self, *, context=None, silent=False):