Deb822 default sources list d

cloudinit/config/cc_apt_configure.py

@@ -18,8 +18,6 @@
 from textwrap import dedent, indent
 from typing import Dict
 
-import apt_pkg
-
 from cloudinit import features, gpg, subp, templater, util
 from cloudinit.cloud import Cloud
 from cloudinit.config import Config
@@ -511,7 +509,7 @@ def disable_suites(disabled, src, release) -> str:
         return src
 
     retsrc = src
-    if is_deb822_format(src):
+    if is_deb822_sources_format(src):
         return disable_suites_deb822(disabled, src, release)
     for suite in disabled:
         suite = map_known_suites(suite)
@@ -551,7 +549,7 @@ def add_mirror_keys(cfg, cloud, target):
             add_apt_key(mirror, cloud, target, file_name=key)
 
 
-def is_deb822_format(apt_src_content: str) -> bool:
+def is_deb822_sources_format(apt_src_content: str) -> bool:
     """Simple check for deb822 format for apt source content
 
     Only validates that minimal required keys are present in the file, which
@@ -562,16 +560,18 @@ def is_deb822_format(apt_src_content: str) -> bool:
     Return True if content looks like it is deb822 formatted APT source.
     """
     # TODO(At jammy EOL: use aptsources.sourceslist.Deb822SourceEntry.invalid)
-    try:
-        apt_src = safeyaml.load(apt_src_content)
-    except (safeyaml.YAMLError, TypeError, ValueError):
+    if re.findall(r"^(deb |deb-src )", apt_src_content, re.M):
         return False
-    if not isinstance(apt_src, dict):
-        return False
-
-    # Are all required deb822 keys present
-    required_keys = set(["Types", "Suites", "Components", "URIs"])
-    return required_keys == required_keys.intersection(apt_src.keys())
+    if re.findall(
+        r"^(Types: |Suites: |Components: |URIs: )", apt_src_content, re.M
+    ):
+        return True
+    # Did not match any required deb822 format keys
+    LOG.warning(
+        "apt.sources_list value does not match either deb822 source keys or"
+        " deb/deb-src list keys. Assuming APT deb/deb-src list format."
+    )
+    return False
 
 
 DEFAULT_APT_CFG = {
@@ -656,7 +656,7 @@ def generate_sources_list(cfg, release, mirrors, cloud):
 
     rendered = templater.render_string(tmpl, params)
     if tmpl:
-        if is_deb822_format(rendered):
+        if is_deb822_sources_format(rendered):
             if aptsrc_file == apt_sources_list:
                 LOG.debug(
                     "Provided 'sources_list' user-data is deb822 format,"

tests/unittests/config/test_apt_configure_sources_list_v3.py

@@ -243,7 +243,7 @@ def test_apt_v3_source_list_ubuntu_snappy(self, mocker):
             ),
         ),
     )
-    def test_apt_v3_source_list_psm(
+    def test_apt_v3_source_list_psm_deb822_feature_aware(
         self,
         deb822,
         tmpl_file,
@@ -253,7 +253,11 @@ def test_apt_v3_source_list_psm(
         tmpdir,
         mocker,
     ):
-        """test_apt_v3_source_list_psm - Test specifying prim+sec mirrors"""
+        """test_apt_v3_source_list_psm - Test specifying prim+sec mirrors
+
+        Assert APT_DEB822_SOURCE_LIST_FILE is taken into account when
+        determining which sources.list.tmpl source file to read.
+        """
         pm = "http://test.ubuntu.com/ubuntu/"
         sm = "http://testsec.ubuntu.com/ubuntu/"
         cfg = {
@@ -307,10 +311,15 @@ def test_apt_v3_source_list_psm(
             ),
         ),
     )
-    def test_apt_v3_srcl_custom(
+    def test_apt_v3_srcl_custom_deb822_feature_aware(
         self, deb822, cfg, apt_file, expected, mocker, tmpdir
     ):
-        """test_apt_v3_srcl_custom - Test rendering a custom source template"""
+        """test_apt_v3_srcl_custom - Test rendering a custom source template
+
+        Also take into account deb822 feature flag to assert writing the
+        appropriate deb822 /etc/apt/sources.list.d/*list or *source file based
+        on content and deb822 feature flag.
+        """
         mycloud = get_cloud("debian")
 
         self.deb822 = mocker.patch.object(

tests/unittests/config/test_apt_source_v3.py

@@ -5,6 +5,7 @@
 This tries to call all in the new v3 format and cares about new features
 """
 import glob
+import logging
 import os
 import pathlib
 import re
@@ -1596,3 +1597,72 @@ def test_use_defaults_or_apt_config_dump(
         subp.side_effect = subp_side_effect
         assert expected == cc_apt_configure.get_apt_cfg()
         subp.assert_called_once_with(["apt-config", "dump"])
+
+
+class TestIsDeb822SourcesFormat:
+    @pytest.mark.parametrize(
+        "content,is_deb822,warnings",
+        (
+            pytest.param(
+                "#Something\ndeb-src http://url lunar multiverse\n",
+                False,
+                [],
+                id="any_deb_src_is_not_deb822",
+            ),
+            pytest.param(
+                "#Something\ndeb http://url lunar multiverse\n",
+                False,
+                [],
+                id="any_deb_url_is_not_deb822",
+            ),
+            pytest.param(
+                "#Something\ndeb http://url lunar multiverse\nTypes: deb\n",
+                False,
+                [],
+                id="even_some_deb822_fields_not_deb822_if_any_deb_line",
+            ),
+            pytest.param(
+                "#Something\nTypes: deb\n",
+                True,
+                [],
+                id="types_deb822_keys_and_no_deb_or_deb_src_is_deb822",
+            ),
+            pytest.param(
+                "#Something\nURIs: http://url\n",
+                True,
+                [],
+                id="uris_deb822_keys_and_no_deb_or_deb_src_is_deb822",
+            ),
+            pytest.param(
+                "#Something\nSuites: http://url\n",
+                True,
+                [],
+                id="suites_deb822_keys_and_no_deb_deb_src_is_deb822",
+            ),
+            pytest.param(
+                "#Something\nComponents: http://url\n",
+                True,
+                [],
+                id="components_deb822_keys_and_no_deb_deb_src_is_deb822",
+            ),
+            pytest.param(
+                "#Something neither deb/deb-src nor deb822\n",
+                False,
+                [
+                    "apt.sources_list value does not match either deb822"
+                    " source keys or deb/deb-src list keys. Assuming APT"
+                    " deb/deb-src list format."
+                ],
+                id="neither_deb822_keys_nor_deb_deb_src_warn_and_not_deb822",
+            ),
+        ),
+    )
+    def test_is_deb822_format_prefers_non_deb822(
+        self, content, is_deb822, warnings, caplog
+    ):
+        with caplog.at_level(logging.WARNING):
+            assert is_deb822 is cc_apt_configure.is_deb822_sources_format(
+                content
+            )
+        for warning in warnings:
+            assert warning in caplog.text