apt: Add deb822 format support for disable_suites in user-data

Provide two new functions to handle disabling deb822 format files
disable_suites_deb822 and disable_deb822_section_without_suites.

When features.APT_DEB822_SOURCE_LIST_FILE is set, deb822 format
apt sources will be rendered and written to /etc/apt/source.list.d/.

A deb822 source file can have suites defined in a single entry:

  Types: deb
  URIs: https://ppa.launchpadcontent.net/something
  Suites: mantic mantic-updates
  Components: main

When disable_suites matches any suite in a deb822 source,
disable_suites_deb822 will preserve the original Suites line as a
comment and redact any active configured Suites from the commented
line. The result when user-data provides disable_suites: [mantic] is:

  Types: deb
  URIs: https://ppa.launchpadcontent.net/something
  # cloud-init disable_suites redacted: Suites: mantic mantic-updates
  Suites: mantic
  Components: main

If all applicable Suites are disabled by cloud-init for an entry,
cloud-init will disabled the entire entry like the following:

  ## Entry disabled by cloud-init, due to disable_suites
  # disabled by cloud-init: Types: deb
  # disabled by cloud-init: URIs: https://ppa.launchpadcontent.net/...
  # disabled by cloud-init: Suites: mantic mantic-updates
  # disabled by cloud-init: Components: main

cloudinit/config/cc_apt_configure.py

@@ -13,7 +13,7 @@
 import pathlib
 import re
 import signal
-from textwrap import dedent
+from textwrap import dedent, indent
 from typing import Dict
 
 import apt_pkg
@@ -34,6 +34,7 @@
 APT_LOCAL_KEYS = "/etc/apt/trusted.gpg"
 APT_TRUSTED_GPG_DIR = "/etc/apt/trusted.gpg.d/"
 CLOUD_INIT_GPG_DIR = "/etc/apt/cloud-init.gpg.d/"
+DISABLE_SUITES_REDACT_PREFIX = "# cloud-init disable_suites redacted: "
 
 frequency = PER_INSTANCE
 distros = ["ubuntu", "debian"]
@@ -413,13 +414,78 @@ def map_known_suites(suite):
     return retsuite
 
 
-def disable_suites(disabled, src, release):
+def disable_deb822_section_without_suites(deb822_entry: str) -> str:
+    """If no active Suites, disable this deb822 source."""
+    if not re.findall(r"\nSuites:[ \t]+([\w-]+)", deb822_entry):
+        # No Suites remaining in this entry, disable full entry
+        # Reconstitute commented Suites line to original as we disable entry
+        deb822_entry = re.sub(r"\nSuites:.*", "", deb822_entry)
+        deb822_entry = re.sub(
+            rf"{DISABLE_SUITES_REDACT_PREFIX}", "", deb822_entry
+        )
+        return (
+            "## Entry disabled by cloud-init, due to disable_suites\n"
+            + indent(deb822_entry, "# disabled by cloud-init: ")
+        )
+    return deb822_entry
+
+
+def disable_suites_deb822(disabled, src, release) -> str:
+    """reads the deb822 format config and comment disabled suites"""
+    new_src = []
+    disabled_suite_names = [
+        templater.render_string(map_known_suites(suite), {"RELEASE": release})
+        for suite in disabled
+    ]
+    LOG.debug("Disabling suites %s as %s", disabled, disabled_suite_names)
+    new_deb822_entry = ""
+    for line in src.splitlines():
+        if line.startswith("#"):
+            if new_deb822_entry:
+                new_deb822_entry += f"{line}\n"
+            else:
+                new_src.append(line)
+            continue
+        if not line or line.isspace():
+            # Validate/disable deb822 entry upon whitespace
+            if new_deb822_entry:
+                new_src.append(
+                    disable_deb822_section_without_suites(new_deb822_entry)
+                )
+                new_deb822_entry = ""
+            new_src.append(line)
+            continue
+        new_line = line
+        if not line.startswith("Suites:"):
+            new_deb822_entry += line + "\n"
+            continue
+        # redact all disabled suite names
+        if disabled_suite_names:
+            # Redact any matching Suites from line
+            orig_suites = line.split()[1:]
+            new_suites = [
+                suite
+                for suite in orig_suites
+                if suite not in disabled_suite_names
+            ]
+            if new_suites != orig_suites:
+                new_deb822_entry += f"{DISABLE_SUITES_REDACT_PREFIX}{line}\n"
+                new_line = f"Suites: {' '.join(new_suites)}"
+        new_deb822_entry += new_line + "\n"
+    if new_deb822_entry:
+        new_src.append(disable_deb822_section_without_suites(new_deb822_entry))
+    return "\n".join(new_src)
+
+
+def disable_suites(disabled, src, release) -> str:
     """reads the config for suites to be disabled and removes those
     from the template"""
     if not disabled:
         return src
 
     retsrc = src
+    if is_deb822_format(src):
+        return disable_suites_deb822(disabled, src, release)
     for suite in disabled:
         suite = map_known_suites(suite)
         releasesuite = templater.render_string(suite, {"RELEASE": release})

tests/unittests/config/test_apt_source_v3.py

@@ -1438,3 +1438,105 @@ def test_dpkg_reconfigure_not_done_on_no_data(self, m_subp):
     def test_dpkg_reconfigure_not_done_if_no_cleaners(self, m_subp):
         cc_apt_configure.dpkg_reconfigure(["pkgfoo", "pkgbar"])
         m_subp.assert_not_called()
+
+
+DEB822_SINGLE_SUITE = """\
+Types: deb
+URIs: https://ppa.launchpadcontent.net/cloud-init-dev/daily/ubuntu/
+Suites: mantic  # Some comment
+Components: main
+"""
+
+DEB822_DISABLED_SINGLE_SUITE = """\
+## Entry disabled by cloud-init, due to disable_suites
+# disabled by cloud-init: Types: deb
+# disabled by cloud-init: URIs: https://ppa.launchpadcontent.net/cloud-init-dev/daily/ubuntu/
+# disabled by cloud-init: Suites: mantic  # Some comment
+# disabled by cloud-init: Components: main
+"""
+
+DEB822_SINGLE_SECTION_TWO_SUITES = """\
+Types: deb
+URIs: https://ppa.launchpadcontent.net/cloud-init-dev/daily/ubuntu/
+Suites: mantic mantic-updates
+Components: main
+"""
+
+DEB822_SINGLE_SECTION_TWO_SUITES_DISABLE_ONE = """\
+Types: deb
+URIs: https://ppa.launchpadcontent.net/cloud-init-dev/daily/ubuntu/
+# cloud-init disable_suites redacted: Suites: mantic mantic-updates
+Suites: mantic-updates
+Components: main
+"""
+
+DEB822_SUITE_2 = """
+# APT Suite 2
+Types: deb
+URIs: https://ppa.launchpadcontent.net/cloud-init-dev/daily/ubuntu/
+Suites: mantic-backports
+Components: main
+"""
+
+
+DEB822_DISABLED_SINGLE_SUITE = """\
+## Entry disabled by cloud-init, due to disable_suites
+# disabled by cloud-init: Types: deb
+# disabled by cloud-init: URIs: https://ppa.launchpadcontent.net/cloud-init-dev/daily/ubuntu/
+# disabled by cloud-init: Suites: mantic  # Some comment
+# disabled by cloud-init: Components: main
+"""
+
+DEB822_DISABLED_MULTIPLE_SUITES = """\
+## Entry disabled by cloud-init, due to disable_suites
+# disabled by cloud-init: Types: deb
+# disabled by cloud-init: URIs: https://ppa.launchpadcontent.net/cloud-init-dev/daily/ubuntu/
+# disabled by cloud-init: Suites: mantic mantic-updates
+# disabled by cloud-init: Components: main
+"""
+
+
+class TestDisableSuitesDeb822:
+    @pytest.mark.parametrize(
+        "disabled_suites,src,expected",
+        (
+            pytest.param(
+                [],
+                DEB822_SINGLE_SUITE,
+                DEB822_SINGLE_SUITE,
+                id="empty_suites_nochange",
+            ),
+            pytest.param(
+                ["$RELEASE-updates"],
+                DEB822_SINGLE_SUITE,
+                DEB822_SINGLE_SUITE,
+                id="no_matching_suites_nochange",
+            ),
+            pytest.param(
+                ["$RELEASE"],
+                DEB822_SINGLE_SUITE,
+                DEB822_DISABLED_SINGLE_SUITE,
+                id="matching_all_suites_disables_whole_section",
+            ),
+            pytest.param(
+                ["$RELEASE"],
+                DEB822_SINGLE_SECTION_TWO_SUITES + DEB822_SUITE_2,
+                DEB822_SINGLE_SECTION_TWO_SUITES_DISABLE_ONE
+                + "\n"
+                + DEB822_SUITE_2,
+                id="matching_some_suites_redacts_matches_and_comments_orig",
+            ),
+            pytest.param(
+                ["$RELEASE", "$RELEASE-updates"],
+                DEB822_SINGLE_SECTION_TWO_SUITES + DEB822_SUITE_2,
+                DEB822_DISABLED_MULTIPLE_SUITES + "\n" + DEB822_SUITE_2,
+                id="matching_all_suites_disables_specific_section",
+            ),
+        ),
+    )
+    def test_disable_deb822_suites_disables_proper_suites(
+        self, disabled_suites, src, expected
+    ):
+        assert expected == cc_apt_configure.disable_suites_deb822(
+            disabled_suites, src, "mantic"
+        )