From c10203545832b7ac9997791b6d34311f0c327651 Mon Sep 17 00:00:00 2001 From: James Garner Date: Tue, 3 Dec 2024 14:33:23 +1300 Subject: [PATCH] feat: use sourceslist (one-per-line) format for add-apt-repository --- lib/charms/operator_libs_linux/v0/apt.py | 112 ++++++++++++++--------- 1 file changed, 68 insertions(+), 44 deletions(-) diff --git a/lib/charms/operator_libs_linux/v0/apt.py b/lib/charms/operator_libs_linux/v0/apt.py index f678018..3888fc3 100644 --- a/lib/charms/operator_libs_linux/v0/apt.py +++ b/lib/charms/operator_libs_linux/v0/apt.py @@ -941,8 +941,8 @@ def __init__( uri: str, release: str, groups: List[str], - filename: Optional[str] = "", - gpg_key_filename: Optional[str] = "", + filename: str = "", + gpg_key_filename: str = "", options: Optional[Dict[str, str]] = None, ): self._enabled = enabled @@ -954,6 +954,24 @@ 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.""" @@ -995,6 +1013,16 @@ 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. + """ + 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.""" @@ -1022,7 +1050,8 @@ def make_options_string(self) -> str: if not options: return "" - return "[{}] ".format(" ".join("{}={}".format(k, v) for k, v in options.items())) + pairs = ("{}={}".format(k, v) for k, v in sorted(options.items())) + return "[{}] ".format(" ".join(pairs)) @staticmethod def prefix_from_uri(uri: str) -> str: @@ -1041,38 +1070,19 @@ def from_repo_line(repo_line: str, write_file: Optional[bool] = True) -> "Debian repo_line: a string representing a repository entry write_file: boolean to enable writing the new repo to disk """ - repo = RepositoryMapping._parse(repo_line, "UserInput") - fname = "{}-{}.list".format( - DebianRepository.prefix_from_uri(repo.uri), repo.release.replace("/", "-") + repo = RepositoryMapping._parse( # pyright: ignore[reportPrivateUsage] + repo_line, "UserInput" ) - repo.filename = fname - - options = repo.options if repo.options else {} - if repo.gpg_key: - options["signed-by"] = repo.gpg_key - - # For Python 3.5 it's required to use sorted in the options dict in order to not have - # different results in the order of the options between executions. - options_str = ( - "[{}] ".format(" ".join(["{}={}".format(k, v) for k, v in sorted(options.items())])) - if options - else "" - ) - - repo_str = ( - "{}".format("#" if not repo.enabled else "") - + "{} {}{} ".format(repo.repotype, options_str, repo.uri) - + "{} {}\n".format(repo.release, " ".join(repo.groups)) - ) - + repo.filename = repo._make_filename() + repo_str = repo._to_line() + "\n" if write_file: logger.info( "DebianRepository.from_repo_line('%s', write_file=True)\nWriting to '%s':\n%s", repo_line, - fname, + repo.filename, repo_str, ) - with open(fname, "wb") as f: + with open(repo.filename, "wb") as f: f.write(repo_str.encode("utf-8")) return repo @@ -1307,8 +1317,8 @@ def load_deb822(self, filename: str) -> None: repos, errors = self._parse_deb822_lines(f, filename=filename) for repo in repos: - repo_identifier = "{}-{}-{}".format(repo.repotype, repo.uri, repo.release) - self._repository_map[repo_identifier] = repo + identifier = repo._get_identifier() # pyright: ignore[reportPrivateUsage] + self._repository_map[identifier] = repo if errors: logger.debug( @@ -1462,7 +1472,8 @@ def add( # noqa: D417 # undocumented-param: default_filename intentionally und self._add_apt_repository(repo, update_cache=update_cache, remove=False) if update_cache: return RepositoryMapping() - self._repository_map["{}-{}-{}".format(repo.repotype, repo.uri, repo.release)] = repo + identifier = repo._get_identifier() # pyright: ignore[reportPrivateUsage] + self._repository_map[identifier] = repo return self def _remove(self, repo: DebianRepository, update_cache: bool = False) -> "RepositoryMapping": @@ -1494,25 +1505,28 @@ def _add_apt_repository( update_cache: bool = False, remove: bool = False, ) -> None: + # DEBUGGING START + from itertools import chain + from pathlib import Path + + logger.info("BEFORE add-apt-repository") + for path in chain( + [Path("/etc/apt/sources.list")], + Path("/etc/apt/sources.list.d").iterdir(), + ): + if path.is_file(): + logger.info("%s: \n%s", str(path), path.read_text()) + else: + logger.info(str(path)) + # DEBUGGING END if not repo.enabled: raise ValueError("{repo}.enabled is {value}".format(repo=repo, value=repo.enabled)) cmd = [ "apt-add-repository", "--yes", - f"--uri={repo.uri}", + "--sourceslist", + repo._to_line(), # pyright: ignore[reportPrivateUsage] ] - ## FIXME: trying to compute pocket seems like a dead end -- add by repo line format instead? - # _, _, pocket = repo.release.partition("-") - # if pocket: - # cmd.append(f"--pocket={pocket}") - if repo.repotype == "deb-src": - cmd.append("--enable-source") - for component in repo.groups: - cmd.append(f"--component={component}") - if not update_cache: - cmd.append("--no-update") - if remove: - cmd.append("--remove") logger.info(" ".join(cmd)) try: subprocess.run(cmd, check=True, capture_output=True) @@ -1524,6 +1538,16 @@ def _add_apt_repository( e.stderr.decode(), ) raise + # DEBUGGING + logger.info("AFTER add-apt-repository") + for path in chain( + [Path("/etc/apt/sources.list")], + Path("/etc/apt/sources.list.d").iterdir(), + ): + if path.is_file(): + logger.info("%s: \n%s", str(path), path.read_text()) + else: + logger.info(str(path)) def disable(self, repo: DebianRepository) -> None: """Remove a repository by disabling it in the source file.