Skip to content

Commit

Permalink
refactor: clean up 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 090bad8
Show file tree
Hide file tree
Showing 3 changed files with 99 additions and 180 deletions.
169 changes: 60 additions & 109 deletions lib/charms/operator_libs_linux/v0/apt.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -806,23 +806,16 @@ def _add(
return name, False


def remove_package(
package_names: Union[str, List[str]],
autoremove: bool = False,
) -> Union[DebianPackage, List[DebianPackage]]:
def remove_package(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 +829,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 @@ -984,24 +974,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 @@ -1107,9 +1079,7 @@ def from_repo_line(repo_line: str, write_file: Optional[bool] = True) -> "Debian
)
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 disable(self) -> None:
Expand Down Expand Up @@ -1253,6 +1223,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 +1333,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 +1457,28 @@ 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.
WARNING: Don't forget to call `apt.update` before installing any packages!
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 +1489,35 @@ 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 # probably 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
82 changes: 36 additions & 46 deletions tests/integration/test_apt.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,86 +65,80 @@ 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
## begin steps
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()
# ^ although apt update hasn't been called yet, we've already modified files on disk
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()
## 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_before.add(repo)
## fi
assert repo_id in repos_after
apt.update()
assert repo_id in repos
assert repo_id in apt.RepositoryMapping()
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
apt._add_repository(repo, remove=True) # pyright: ignore[reportPrivateUsage]
assert repo_id not in apt.RepositoryMapping()
# ^ although apt update hasn't been called yet, we've already modified files on disk
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"):
Expand All @@ -165,13 +159,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")
Expand All @@ -186,10 +176,10 @@ 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()
# ^ although apt update hasn't been called yet, we've already modified files on disk
apt.update()
apt.remove_package("inkscape")
assert not get_command_path("inkscape")

Expand Down
Loading

0 comments on commit 090bad8

Please sign in to comment.