diff --git a/lib/charms/operator_libs_linux/v0/apt.py b/lib/charms/operator_libs_linux/v0/apt.py index 78bfbe3..e62020c 100644 --- a/lib/charms/operator_libs_linux/v0/apt.py +++ b/lib/charms/operator_libs_linux/v0/apt.py @@ -102,7 +102,6 @@ import fileinput import glob -import itertools import logging import os import re @@ -807,22 +806,17 @@ def _add( def remove_package( - package_names: Union[str, List[str]], - autoremove: bool = False, + package_names: Union[str, List[str]] ) -> Union[DebianPackage, List[DebianPackage]]: """Remove package(s) from the system. Args: package_names: the name of a package - autoremove: run `apt autoremove` after uninstalling packages. - You probably want to do this if you're removing a metapackage, - otherwise the concrete packages will still be there. - False by default for backwards compatibility. Raises: - PackageNotFoundError if the package is not found. + TypeError: if no packages are provided """ - packages = [] + packages: List[DebianPackage] = [] package_names = [package_names] if isinstance(package_names, str) else package_names if not package_names: @@ -836,9 +830,6 @@ def remove_package( except PackageNotFoundError: logger.info("package '%s' was requested for removal, but it was not installed.", p) - if autoremove: - subprocess.run(["apt", "autoremove"], check=True) - # the list of packages will be empty when no package is removed logger.debug("packages: '%s'", packages) return packages[0] if len(packages) == 1 else packages @@ -925,35 +916,6 @@ class InvalidSourceError(Error): """Exceptions for invalid source entries.""" -class MissingRequiredKeyError(InvalidSourceError): - """Missing a required value in a source file.""" - - def __init__(self, message: str = "", *, file: str, line: Optional[int], key: str) -> None: - super().__init__(message, file, line, key) - self.file = file - self.line = line - self.key = key - - -class BadValueError(InvalidSourceError): - """Bad value for an entry in a source file.""" - - def __init__( - self, - message: str = "", - *, - file: str, - line: Optional[int], - key: str, - value: str, - ) -> None: - super().__init__(message, file, line, key, value) - self.file = file - self.line = line - self.key = key - self.value = value - - class GPGKeyError(Error): """Exceptions for GPG keys.""" @@ -984,24 +946,6 @@ def __init__( self._gpg_key_filename = gpg_key_filename self._options = options - def _get_identifier(self) -> str: - """Return str identifier derived from repotype, uri, and release. - - Private method used to produce the identifiers used by RepositoryMapping. - """ - return "{}-{}-{}".format(self.repotype, self.uri, self.release) - - def _to_line(self) -> str: - """Return the one-per-line format repository definition.""" - return "{prefix}{repotype} {options}{uri} {release} {groups}".format( - prefix="" if self.enabled else "#", - repotype=self.repotype, - options=self.make_options_string(), - uri=self.uri, - release=self.release, - groups=" ".join(self.groups), - ) - @property def enabled(self): """Return whether or not the repository is enabled.""" @@ -1043,17 +987,6 @@ def filename(self, fname: str) -> None: raise InvalidSourceError("apt source filenames should end in .list or .sources!") self._filename = fname - def _make_filename(self) -> str: - """Construct a filename from uri and release. - - For internal use when a filename isn't set. - Should match the filename written to by add-apt-repository. - """ - return "{}-{}.list".format( - DebianRepository.prefix_from_uri(self.uri), - self.release.replace("/", "-"), - ) - @property def gpg_key(self): """Returns the path to the GPG key for this repository.""" @@ -1100,18 +1033,28 @@ def from_repo_line(repo_line: str, write_file: Optional[bool] = True) -> "Debian Args: repo_line: a string representing a repository entry write_file: boolean to enable writing the new repo to disk. True by default. - Results in calling `add-apt-repository --no-update --sourceslist $repo_line` + Expect it to result in an add-apt-repository call under the hood, like: + add-apt-repository --no-update --sourceslist="$repo_line" """ repo = RepositoryMapping._parse( # pyright: ignore[reportPrivateUsage] repo_line, filename="UserInput" # temp filename ) repo.filename = repo._make_filename() if write_file: - RepositoryMapping._add_apt_repository( # pyright: ignore[reportPrivateUsage]) - repo, update_cache=False - ) + _add_repository(repo) return repo + def _make_filename(self) -> str: + """Construct a filename from uri and release. + + For internal use when a filename isn't set. + Should match the filename written to by add-apt-repository. + """ + return "{}-{}.list".format( + DebianRepository.prefix_from_uri(self.uri), + self.release.replace("/", "-"), + ) + def disable(self) -> None: """Remove this repository by disabling it in the source file. @@ -1253,6 +1196,26 @@ def _write_apt_gpg_keyfile(key_name: str, key_material: bytes) -> None: keyf.write(key_material) +def _repo_to_identifier(repo: DebianRepository) -> str: + """Return str identifier derived from repotype, uri, and release. + + Private method used to produce the identifiers used by RepositoryMapping. + """ + return "{}-{}-{}".format(repo.repotype, repo.uri, repo.release) + + +def _repo_to_line(repo: DebianRepository) -> str: + """Return the one-per-line format repository definition.""" + return "{prefix}{repotype} {options}{uri} {release} {groups}".format( + prefix="" if repo.enabled else "#", + repotype=repo.repotype, + options=repo.make_options_string(), + uri=repo.uri, + release=repo.release, + groups=" ".join(repo.groups), + ) + + class RepositoryMapping(Mapping[str, DebianRepository]): """An representation of known repositories. @@ -1343,8 +1306,7 @@ def load_deb822(self, filename: str) -> None: repos, errors = self._parse_deb822_lines(f, filename=filename) for repo in repos: - identifier = repo._get_identifier() # pyright: ignore[reportPrivateUsage] - self._repository_map[identifier] = repo + self._repository_map[_repo_to_identifier(repo)] = repo if errors: self._last_errors = errors @@ -1468,95 +1430,29 @@ def _parse(line: str, filename: str) -> DebianRepository: raise InvalidSourceError("An invalid sources line was found in %s!", filename) def add( # noqa: D417 # undocumented-param: default_filename intentionally undocumented - self, - repo: DebianRepository, - default_filename: Optional[bool] = False, - update_cache: bool = False, - ) -> "RepositoryMapping": + self, repo: DebianRepository, default_filename: Optional[bool] = False + ) -> None: """Add a new repository to the system using add-apt-repository. Args: repo: a DebianRepository object where repo.enabled is True - update_cache: if True, apt-add-repository will update the package cache - and then a new RepositoryMapping object will be returned. - If False, then apt-add-repository is run with the --no-update option. - An entry for the repo is added to this RepositoryMapping before returning it. - Don't forget to call `apt.update` manually before installing any packages! - - Returns: - self, or a new RepositoryMapping object if update_cache is True - raises: ValueError: if repo.enabled is False CalledProcessError: if there's an error running apt-add-repository WARNING: Does not associate the repository with a signing key. - Use the `import_key` method to add a signing key globally. + Use `import_key` to add a signing key globally. + + WARNING: Don't forget to call `apt.update` before installing any packages! + Or call `apt.add_package` with `update_cache=True`. WARNING: the default_filename keyword argument is provided for backwards compatibility only. It is not used, and was not used in the previous revision of this library. """ - self._add_apt_repository(repo, update_cache=update_cache, remove=False) - if update_cache: - return RepositoryMapping() - identifier = repo._get_identifier() # pyright: ignore[reportPrivateUsage] - self._repository_map[identifier] = repo - return self - - def _remove(self, repo: DebianRepository, update_cache: bool = False) -> "RepositoryMapping": - """Use add-apt-repository to remove a repository. - - Args: - repo: a DebianRepository object where repo.enabled is True - update_cache: if True, apt-add-repository will update the package cache - and then a new RepositoryMapping object will be returned. - If False, then apt-add-repository is run with the --no-update option. - Any entry for the repo is removed from this RepositoryMapping before returning it. - Don't forget to call `apt.update` manually before installing any packages! - - Returns: - self, or a new RepositoryMapping object if update_cache is True - - raises: - ValueError: if repo.enabled is False - CalledProcessError: if there's an error running apt-add-repository - """ - self._add_apt_repository(repo, update_cache=update_cache, remove=True) - if update_cache: - return RepositoryMapping() - identifier = repo._get_identifier() # pyright: ignore[reportPrivateUsage] - self._repository_map.pop(identifier, None) - return self - - @staticmethod - def _add_apt_repository( - repo: DebianRepository, - update_cache: bool = False, - remove: bool = False, - ) -> None: if not repo.enabled: raise ValueError("{repo}.enabled is {value}".format(repo=repo, value=repo.enabled)) - line = repo._to_line() # pyright: ignore[reportPrivateUsage] - cmd = [ - "add-apt-repository", - "--yes", - "--sourceslist=" + line, - ] - if remove: - cmd.append("--remove") - if not update_cache: - cmd.append("--no-update") - logger.info("%s", cmd) - try: - subprocess.run(cmd, check=True, capture_output=True) - except CalledProcessError as e: - logger.error( - "subprocess.run(%s):\nstdout:\n%s\nstderr:\n%s", - cmd, - e.stdout.decode(), - e.stderr.decode(), - ) - raise + _add_repository(repo) + self._repository_map[_repo_to_identifier(repo)] = repo def disable(self, repo: DebianRepository) -> None: """Remove a repository by disabling it in the source file. @@ -1567,6 +1463,36 @@ def disable(self, repo: DebianRepository) -> None: WARNING: This method does NOT alter the `.enabled` flag on the DebianRepository. """ repo.disable() + self._repository_map[_repo_to_identifier(repo)] = repo + # ^ adding to map on disable seems like a bug, but this is the previous behaviour + + +def _add_repository( + repo: DebianRepository, + remove: bool = False, + update_cache: bool = False, +) -> None: + line = _repo_to_line(repo) # pyright: ignore[reportPrivateUsage] + cmd = [ + "add-apt-repository", + "--yes", + "--sourceslist=" + line, + ] + if remove: + cmd.append("--remove") + if not update_cache: + cmd.append("--no-update") + logger.info("%s", cmd) + try: + subprocess.run(cmd, check=True, capture_output=True) + except CalledProcessError as e: + logger.error( + "subprocess.run(%s):\nstdout:\n%s\nstderr:\n%s", + cmd, + e.stdout.decode(), + e.stderr.decode(), + ) + raise class _Deb822Stanza: @@ -1613,6 +1539,35 @@ def get_gpg_key_filename(self) -> str: return self._gpg_key_filename +class MissingRequiredKeyError(InvalidSourceError): + """Missing a required value in a source file.""" + + def __init__(self, message: str = "", *, file: str, line: Optional[int], key: str) -> None: + super().__init__(message, file, line, key) + self.file = file + self.line = line + self.key = key + + +class BadValueError(InvalidSourceError): + """Bad value for an entry in a source file.""" + + def __init__( + self, + message: str = "", + *, + file: str, + line: Optional[int], + key: str, + value: str, + ) -> None: + super().__init__(message, file, line, key, value) + self.file = file + self.line = line + self.key = key + self.value = value + + def _iter_deb822_stanzas(lines: Iterable[str]) -> Iterator[List[Tuple[int, str]]]: """Given lines from a deb822 format file, yield a stanza of lines. @@ -1767,6 +1722,8 @@ def _deb822_options_to_repos( gpg_key_filename=gpg_key_file, options=options, ) - for repotype, uri, suite in itertools.product(repotypes, uris, suites) + for repotype in repotypes + for uri in uris + for suite in suites ) return repos, (gpg_key_file, gpg_key_from_stanza) diff --git a/tests/integration/test_apt.py b/tests/integration/test_apt.py index ea91ea6..7feb0cd 100644 --- a/tests/integration/test_apt.py +++ b/tests/integration/test_apt.py @@ -65,11 +65,10 @@ def test_install_hardware_observer_ssacli(): apt.add_package(self.pkg, update_cache=True) """ line = "deb http://downloads.linux.hpe.com/SDR/repo/mcp stretch/current non-free" - repo_id = apt.DebianRepository.from_repo_line( - line, write_file=False - )._get_identifier() # pyright: ignore[reportPrivateUsage] - repos_before = apt.RepositoryMapping() - assert repo_id not in repos_before + repo_id = apt._repo_to_identifier( # pyright: ignore[reportPrivateUsage] + apt.DebianRepository.from_repo_line(line, write_file=False) + ) + assert repo_id not in apt.RepositoryMapping() assert not get_command_path("ssacli") key_files: List[str] = [] # just for cleanup @@ -77,74 +76,68 @@ def test_install_hardware_observer_ssacli(): for key in HP_KEYS: key_file = apt.import_key(key) key_files.append(key_file) - repositories = apt.RepositoryMapping() + repos = apt.RepositoryMapping() repo = apt.DebianRepository.from_repo_line(line) # write_file=True by default - # repo added to system but repositories doesn't know about it yet - assert repo_id not in repositories - repositories.add(repo) # update_cache=False by default + assert repo_id in apt.RepositoryMapping() + # repo added to system but repos doesn't know about it yet + assert repo_id not in repos + repos.add(repo) + assert repo_id in repos # `add` call is redundant with `from_repo_line` from system pov - # but it does add an entry to the RepositoryMapping - assert repo_id in repositories + # but adds an entry to the RepositoryMapping apt.add_package("ssacli", update_cache=True) + # apt.update not required since update_cache=True assert get_command_path("ssacli") - # install succeed here as update_cache=True ## end steps ## cleanup for key_file in key_files: os.remove(key_file) - repos_clean = repositories._remove( # pyright: ignore[reportPrivateUsage] - repo, update_cache=True - ) - assert repo_id not in repos_clean + apt._add_repository(repo, remove=True) # pyright: ignore[reportPrivateUsage] + assert repo_id not in apt.RepositoryMapping() + apt.update() apt.remove_package("ssacli") assert not get_command_path("ssacli") def test_install_package_from_external_repository(): repo_id = "deb-https://repo.mongodb.org/apt/ubuntu-jammy/mongodb-org/8.0" - repos_before = apt.RepositoryMapping() - assert repo_id not in repos_before + repos = apt.RepositoryMapping() + assert repo_id not in repos assert not get_command_path("mongod") key = urlopen("https://www.mongodb.org/static/pgp/server-8.0.asc").read().decode() line = "deb [ arch=amd64,arm64 ] https://repo.mongodb.org/apt/ubuntu jammy/mongodb-org/8.0 multiverse" - ## this is one valid order of operations: - ## apt.import_key(...) # before add - ## apt.RepositoryMapping.add(repo, update_cache=True) - ## apt.add_package(...) key_file = apt.import_key(key) ## if: use implicit write_file repo = apt.DebianRepository.from_repo_line(line) # write_file=True by default - apt.update() - repos_after = apt.RepositoryMapping() + assert repo_id not in repos ## if: don't use implicit write_file # repo = apt.DebianRepository.from_repo_line(line, write_file=False) - # repos_after = repos_before.add(repo, update_cache=True) + # repos.add(repo) + # assert repo_id in repos ## fi - assert repo_id in repos_after + assert repo_id in apt.RepositoryMapping() + apt.update() apt.add_package("mongodb-org") assert get_command_path("mongod") subprocess.run(["mongod", "--version"], check=True) ## cleanup os.remove(key_file) - repos_after._remove(repo, update_cache=False) # pyright: ignore[reportPrivateUsage] - assert repo_id not in repos_after - repos_before_update = apt.RepositoryMapping() - assert repo_id not in repos_before_update - # ^ although apt update hasn't been called yet, we've already modified files on disk + apt._add_repository(repo, remove=True) # pyright: ignore[reportPrivateUsage] + assert repo_id not in apt.RepositoryMapping() apt.update() - repos_clean = apt.RepositoryMapping() - assert repo_id not in repos_clean - apt.remove_package("mongodb-org", autoremove=True) # mongodb-org is a metapackage + apt.remove_package("mongodb-org") + assert get_command_path("mongod") # mongodb-org is a metapackage + subprocess.run(["apt", "autoremove"], check=True) assert not get_command_path("mongod") def test_install_higher_version_package_from_external_repository(): repo_id = "deb-https://ppa.launchpadcontent.net/inkscape.dev/stable/ubuntu/-jammy" - repos_before = apt.RepositoryMapping() - assert repo_id not in repos_before + repos = apt.RepositoryMapping() + assert repo_id not in repos # version before if not get_command_path("inkscape"): @@ -165,13 +158,9 @@ def test_install_higher_version_package_from_external_repository(): release="jammy", groups=["main"], ) - ## this is a different, valid order of operations: - ## apt.RepositoryMapping.add(..., update_cache=False) - ## apt.import_key(...) # before update but after add - ## apt.update() - ## apt.add_package(...) - repos_after = repos_before.add(repo) # update_cache=False by default - assert repo_id in repos_after + repos.add(repo) # update_cache=False by default + assert repo_id in repos + assert repo_id in apt.RepositoryMapping() key_file = apt.import_key(INKSCAPE_KEY) apt.update() apt.add_package("inkscape") @@ -186,10 +175,9 @@ def test_install_higher_version_package_from_external_repository(): ## cleanup os.remove(key_file) - repos_clean = repos_after._remove( # pyright: ignore[reportPrivateUsage] - repo, update_cache=True - ) - assert repo_id not in repos_clean + apt._add_repository(repo, remove=True) # pyright: ignore[reportPrivateUsage] + assert repo_id not in apt.RepositoryMapping() + apt.update() apt.remove_package("inkscape") assert not get_command_path("inkscape") diff --git a/tests/unit/test_deb822.py b/tests/unit/test_deb822.py index 0bad2df..ac575a4 100644 --- a/tests/unit/test_deb822.py +++ b/tests/unit/test_deb822.py @@ -317,33 +317,11 @@ def test_add_with_deb822(repo_mapping: apt.RepositoryMapping): assert len(repos) == 1 assert not errors [repo] = repos - identifier = repo._get_identifier() - # call add with update_cache=True + identifier = apt._repo_to_identifier(repo) with patch.object(apt.subprocess, "run") as mock_run_1: - with patch.object( - apt.RepositoryMapping, - "_apt_dir", - str(FAKE_APT_DIRS / "empty"), - ): - r1 = repo_mapping.add(repo, update_cache=True) - assert r1 is not repo_mapping - assert identifier not in repo_mapping # because it's the old mapping - assert identifier not in r1 # because we mocked out the subprocess call - mock_run_1.assert_called_once_with( - [ - "add-apt-repository", - "--yes", - "--sourceslist=deb http://nz.archive.ubuntu.com/ubuntu/ an/exact/path/ ", - ], - check=True, - capture_output=True, - ) - # call add with update_cache=False - with patch.object(apt.subprocess, "run") as mock_run_2: - r2 = repo_mapping.add(repo) # update_cache=False by default - assert r2 is repo_mapping + repo_mapping.add(repo) assert identifier in repo_mapping - mock_run_2.assert_called_once_with( + mock_run_1.assert_called_once_with( [ "add-apt-repository", "--yes",