Skip to content

Commit

Permalink
refactor: clean up diff, signatures, etc
Browse files Browse the repository at this point in the history
  • Loading branch information
@james-garner-canonical
james-garner-canonical committed Dec 4, 2024
1 parent 58c61be commit e52f8b4
Show file tree
Hide file tree
Showing 3 changed files with 180 additions and 257 deletions.
259 changes: 108 additions & 151 deletions lib/charms/operator_libs_linux/v0/apt.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,6 @@

import fileinput
import glob
import itertools
import logging
import os
import re
Expand Down Expand Up @@ -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:
Expand All @@ -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
Expand Down Expand Up @@ -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."""

Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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:
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Loading

0 comments on commit e52f8b4

Please sign in to comment.