diff --git a/subiquity/server/controllers/network.py b/subiquity/server/controllers/network.py index 463c8131..455d6337 100644 --- a/subiquity/server/controllers/network.py +++ b/subiquity/server/controllers/network.py @@ -109,14 +109,22 @@ class NetworkController(BaseNetworkController, SubiquityController): 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) + + async def fake_install(pkgname: str) -> PackageInstallState: + 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] + return getattr(PackageInstallState, a) + + install_coro = fake_install else: - r = await self.app.package_installer.install_pkg("wpasupplicant") + install_coro = None + + r = await self.app.package_installer.install_pkg( + "wpasupplicant", install_coro=install_coro + ) log.debug("wlan_support_install_finished %s", r) self._call_clients("wlan_support_install_finished", r) if r == PackageInstallState.DONE: diff --git a/subiquity/server/pkghelper.py b/subiquity/server/pkghelper.py index d27521d5..5e5828e4 100644 --- a/subiquity/server/pkghelper.py +++ b/subiquity/server/pkghelper.py @@ -16,7 +16,7 @@ import asyncio import logging import os -from typing import Dict, Optional +from typing import Callable, Dict, Optional import apt @@ -26,6 +26,9 @@ from subiquitycore.utils import arun_command log = logging.getLogger("subiquity.server.pkghelper") +InstallCoroutine = Optional[Callable[[str], PackageInstallState]] + + class PackageInstaller: """Install packages from the pool on the ISO in the live session. @@ -54,12 +57,22 @@ class PackageInstaller: else: return PackageInstallState.INSTALLING - def start_installing_pkg(self, pkgname: str) -> None: + def start_installing_pkg( + self, pkgname: str, *, install_coro: InstallCoroutine = None + ) -> None: if pkgname not in self.pkgs: - self.pkgs[pkgname] = asyncio.create_task(self._install_pkg(pkgname)) + if install_coro is None: + install_coro = self._install_pkg + self.pkgs[pkgname] = asyncio.create_task(install_coro(pkgname)) - async def install_pkg(self, pkgname) -> PackageInstallState: - self.start_installing_pkg(pkgname) + async def install_pkg( + self, pkgname: str, *, install_coro: InstallCoroutine = None + ) -> PackageInstallState: + """Install the requested package. The default implementation runs + apt-get (or will consider the package installed after two seconds in + dry-run mode). If a different implementation is wanted, one can specify + an alternative install coroutine""" + self.start_installing_pkg(pkgname, install_coro=install_coro) return await self.pkgs[pkgname] async def _install_pkg(self, pkgname: str) -> PackageInstallState: diff --git a/subiquity/server/tests/test_pkghelper.py b/subiquity/server/tests/test_pkghelper.py new file mode 100644 index 00000000..faa2d91e --- /dev/null +++ b/subiquity/server/tests/test_pkghelper.py @@ -0,0 +1,101 @@ +# 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 . + +from typing import Optional +from unittest.mock import Mock, patch + +import attr + +from subiquity.server.pkghelper import PackageInstaller, PackageInstallState +from subiquitycore.tests import SubiTestCase +from subiquitycore.tests.mocks import make_app + + +@patch("apt.Cache", Mock(return_value={})) +class TestPackageInstaller(SubiTestCase): + class Package: + @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) + + def setUp(self): + self.pkginstaller = PackageInstaller(make_app()) + + 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": self.Package(installed=True, name="util-linux")}, + ): + self.assertEqual( + await self.pkginstaller.install_pkg("util-linux"), + PackageInstallState.DONE, + ) + + async def test_install_pkg_dr_install(self): + with patch.dict( + self.pkginstaller.cache, + {"python3-attr": self.Package(installed=False, name="python3-attr")}, + ): + with patch("subiquity.server.pkghelper.asyncio.sleep") as sleep: + self.assertEqual( + await self.pkginstaller.install_pkg("python3-attr"), + PackageInstallState.DONE, + ) + sleep.assert_called_once() + + async def test_install_pkg_not_from_cdrom(self): + with patch.dict( + self.pkginstaller.cache, + { + "python3-attr": self.Package( + installed=False, + name="python3-attr", + candidate_uri="http://archive.ubuntu.com", + ) + }, + ): + with patch.object(self.pkginstaller.app.opts, "dry_run", False): + self.assertEqual( + await self.pkginstaller.install_pkg("python3-attr"), + PackageInstallState.NOT_AVAILABLE, + ) + + async def test_install_pkg_alternative_impl(self): + async def impl(pkgname: str) -> PackageInstallState: + return PackageInstallState.FAILED + + with patch.object(self.pkginstaller, "_install_pkg") as default_impl: + self.assertEqual( + await self.pkginstaller.install_pkg("python3-attr", install_coro=impl), + PackageInstallState.FAILED, + ) + + default_impl.assert_not_called()