network: fix Wi-Fi interfaces not listed in dry-run

subiquity/server/controllers/network.py

@@ -108,15 +108,7 @@ def wlan_support_install_state(self):
         return self.app.package_installer.state_for_pkg("wpasupplicant")
 
     async def _install_wpasupplicant(self):
-        if self.opts.dry_run:
-            await asyncio.sleep(10 / self.app.scale_factor)
-            a = "DONE"
-            for k in self.app.debug_flags:
-                if k.startswith("wlan_install="):
-                    a = k.split("=", 2)[1]
-            r = getattr(PackageInstallState, a)
-        else:
-            r = await self.app.package_installer.install_pkg("wpasupplicant")
+        r = await self.app.package_installer.install_pkg("wpasupplicant")
         log.debug("wlan_support_install_finished %s", r)
         self._call_clients("wlan_support_install_finished", r)
         if r == PackageInstallState.DONE:

subiquity/server/pkghelper.py

@@ -16,7 +16,7 @@
 import asyncio
 import logging
 import os
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 
 import apt
 
@@ -34,10 +34,9 @@ class PackageInstaller:
     by the server installer.
     """
 
-    def __init__(self, app):
+    def __init__(self):
         self.pkgs: Dict[str, asyncio.Task] = {}
         self._cache: Optional[apt.Cache] = None
-        self.app = app
 
     @property
     def cache(self):
@@ -58,7 +57,7 @@ def start_installing_pkg(self, pkgname: str) -> None:
         if pkgname not in self.pkgs:
             self.pkgs[pkgname] = asyncio.create_task(self._install_pkg(pkgname))
 
-    async def install_pkg(self, pkgname) -> PackageInstallState:
+    async def install_pkg(self, pkgname: str) -> PackageInstallState:
         self.start_installing_pkg(pkgname)
         return await self.pkgs[pkgname]
 
@@ -71,10 +70,6 @@ async def _install_pkg(self, pkgname: str) -> PackageInstallState:
         if binpkg.installed:
             log.debug("%s already installed", pkgname)
             return PackageInstallState.DONE
-        if self.app.opts.dry_run:
-            log.debug("dry-run apt-get install %s", pkgname)
-            await asyncio.sleep(2 / self.app.scale_factor)
-            return PackageInstallState.DONE
         if not binpkg.candidate.uri.startswith("cdrom:"):
             log.debug(
                 "%s not available from cdrom (rather %s)", pkgname, binpkg.candidate.uri
@@ -94,3 +89,46 @@ async def _install_pkg(self, pkgname: str) -> PackageInstallState:
             return PackageInstallState.DONE
         else:
             return PackageInstallState.FAILED
+
+
+class DryRunPackageInstaller(PackageInstaller):
+    def __init__(self, app) -> None:
+        super().__init__()
+        self.scale_factor: float = app.scale_factor
+        self.debug_flags: List[str] = app.debug_flags
+        self.package_specific_impl = {
+            "wpasupplicant": self._install_wpa_supplicant,
+        }
+
+    async def _install_wpa_supplicant(self) -> PackageInstallState:
+        """Special implementation for wpasupplicant (used by code related to
+        Wi-Fi interfaces)."""
+        await asyncio.sleep(10 / self.scale_factor)
+        status = "DONE"
+        for flag in self.debug_flags:
+            if flag.startswith("wlan_install="):
+                status = flag.split("=", 2)[1]
+        return getattr(PackageInstallState, status)
+
+    async def _install_pkg(self, pkgname: str) -> PackageInstallState:
+        if pkgname in self.package_specific_impl:
+            return await self.package_specific_impl[pkgname]()
+
+        log.debug("checking if %s is available", pkgname)
+        binpkg = self.cache.get(pkgname)
+        if not binpkg:
+            log.debug("%s not found", pkgname)
+            return PackageInstallState.NOT_AVAILABLE
+        if binpkg.installed:
+            log.debug("%s already installed", pkgname)
+            return PackageInstallState.DONE
+        log.debug("dry-run apt-get install %s", pkgname)
+        await asyncio.sleep(2 / self.scale_factor)
+        return PackageInstallState.DONE
+
+
+def get_package_installer(app):
+    if app.opts.dry_run:
+        return DryRunPackageInstaller(app)
+    else:
+        return PackageInstaller()

subiquity/server/server.py

@@ -44,7 +44,7 @@
 from subiquity.server.dryrun import DRConfig
 from subiquity.server.errors import ErrorController
 from subiquity.server.geoip import DryRunGeoIPStrategy, GeoIP, HTTPGeoIPStrategy
-from subiquity.server.pkghelper import PackageInstaller
+from subiquity.server.pkghelper import get_package_installer
 from subiquity.server.runner import get_command_runner
 from subiquity.server.snapdapi import make_api_client
 from subiquity.server.types import InstallerChannels
@@ -291,7 +291,7 @@ def __init__(self, opts, block_log_dir):
         self.event_syslog_id = "subiquity_event.{}".format(os.getpid())
         self.log_syslog_id = "subiquity_log.{}".format(os.getpid())
         self.command_runner = get_command_runner(self)
-        self.package_installer = PackageInstaller(self)
+        self.package_installer = get_package_installer(self)
 
         self.error_reporter = ErrorReporter(
             self.context.child("ErrorReporter"), self.opts.dry_run, self.root

subiquity/server/tests/test_pkghelper.py

@@ -0,0 +1,130 @@
+# 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/>.
+
+from typing import Optional
+from unittest.mock import Mock, patch
+
+import attr
+
+from subiquity.server.pkghelper import (
+    DryRunPackageInstaller,
+    PackageInstaller,
+    PackageInstallState,
+)
+from subiquity.server.pkghelper import log as PkgHelperLogger
+from subiquitycore.tests import SubiTestCase
+from subiquitycore.tests.mocks import make_app
+
+
+class MockPackage:
+    @attr.s(auto_attribs=True)
+    class Candidate:
+        uri: str
+
+    def __init__(
+        self, installed: bool, name: str, candidate_uri: Optional[str] = None
+    ) -> None:
+        self.installed = installed
+        self.name = name
+        if candidate_uri is None:
+            self.candidate = None
+        else:
+            self.candidate = self.Candidate(uri=candidate_uri)
+
+
+@patch("apt.Cache", Mock(return_value={}))
+class TestPackageInstaller(SubiTestCase):
+    def setUp(self):
+        self.pkginstaller = PackageInstaller()
+
+    async def test_install_pkg_not_found(self):
+        self.assertEqual(
+            await self.pkginstaller.install_pkg("sysvinit-core"),
+            PackageInstallState.NOT_AVAILABLE,
+        )
+
+    async def test_install_pkg_already_installed(self):
+        with patch.dict(
+            self.pkginstaller.cache,
+            {"util-linux": MockPackage(installed=True, name="util-linux")},
+        ):
+            self.assertEqual(
+                await self.pkginstaller.install_pkg("util-linux"),
+                PackageInstallState.DONE,
+            )
+
+    async def test_install_pkg_not_from_cdrom(self):
+        with patch.dict(
+            self.pkginstaller.cache,
+            {
+                "python3-attr": MockPackage(
+                    installed=False,
+                    name="python3-attr",
+                    candidate_uri="http://archive.ubuntu.com",
+                )
+            },
+        ):
+            self.assertEqual(
+                await self.pkginstaller.install_pkg("python3-attr"),
+                PackageInstallState.NOT_AVAILABLE,
+            )
+
+
+@patch("apt.Cache", Mock(return_value={}))
+@patch("subiquity.server.pkghelper.asyncio.sleep")
+class TestDryRunPackageInstaller(SubiTestCase):
+    def setUp(self):
+        app = make_app()
+        app.debug_flags = []
+        self.pkginstaller = DryRunPackageInstaller(app)
+
+    async def test_install_pkg(self, sleep):
+        with patch.dict(
+            self.pkginstaller.cache,
+            {"python3-attr": MockPackage(installed=False, name="python3-attr")},
+        ):
+            with self.assertLogs(PkgHelperLogger, "DEBUG") as debug:
+                self.assertEqual(
+                    await self.pkginstaller.install_pkg("python3-attr"),
+                    PackageInstallState.DONE,
+                )
+        sleep.assert_called_once()
+        self.assertIn(
+            "dry-run apt-get install %s", [record.msg for record in debug.records]
+        )
+
+    async def test_install_pkg_wpasupplicant_default_impl(self, sleep):
+        with patch.object(self.pkginstaller, "debug_flags", []):
+            self.assertEqual(
+                await self.pkginstaller.install_pkg("wpasupplicant"),
+                PackageInstallState.DONE,
+            )
+        sleep.assert_called_once()
+
+    async def test_install_pkg_wpasupplicant_done(self, sleep):
+        with patch.object(self.pkginstaller, "debug_flags", ["wlan_install=DONE"]):
+            self.assertEqual(
+                await self.pkginstaller.install_pkg("wpasupplicant"),
+                PackageInstallState.DONE,
+            )
+        sleep.assert_called_once()
+
+    async def test_install_pkg_wpasupplicant_failed(self, sleep):
+        with patch.object(self.pkginstaller, "debug_flags", ["wlan_install=FAILED"]):
+            self.assertEqual(
+                await self.pkginstaller.install_pkg("wpasupplicant"),
+                PackageInstallState.FAILED,
+            )
+        sleep.assert_called_once()