feat: use sourceslist (one-per-line) format for add-apt-repository

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.