[Freeze] Fixed parsing CDROM apt sources

changelog/62474.fixed

@@ -0,0 +1 @@
+Fixed parsing CDROM apt sources

salt/modules/aptpkg.py

@@ -149,7 +149,15 @@ def _invalid(line):
         comment = line[idx + 1 :]
         line = line[:idx]
 
-    repo_line = line.strip().split()
+    cdrom_match = re.match(r"(.*)(cdrom:.*/)(.*)", line.strip())
+    if cdrom_match:
+        repo_line = (
+            [p.strip() for p in cdrom_match.group(1).split()]
+            + [cdrom_match.group(2).strip()]
+            + [p.strip() for p in cdrom_match.group(3).split()]
+        )
+    else:
+        repo_line = line.strip().split()
     if (
         not repo_line
         or repo_line[0] not in ["deb", "deb-src", "rpm", "rpm-src"]
@@ -1719,7 +1727,7 @@ def _get_opts(line):
     """
     Return all opts in [] for a repo line
     """
-    get_opts = re.search(r"\[.*\]", line)
+    get_opts = re.search(r"\[(.*=.*)\]", line)
     ret = {
         "arch": {"full": "", "value": "", "index": 0},
         "signedby": {"full": "", "value": "", "index": 0},
@@ -1851,7 +1859,6 @@ def list_repo_pkgs(*args, **kwargs):  # pylint: disable=unused-import
 
     ret = {}
     pkg_name = None
-    skip_pkg = False
     new_pkg = re.compile("^Package: (.+)")
     for line in salt.utils.itertools.split(out["stdout"], "\n"):
         if not line.strip():

tests/pytests/functional/states/pkgrepo/test_debian.py

@@ -69,6 +69,24 @@ def test_adding_repo_file_arch(pkgrepo, tmp_path):
         )
 
 
+@pytest.mark.requires_salt_states("pkgrepo.managed")
+def test_adding_repo_file_cdrom(pkgrepo, tmp_path):
+    """
+    test adding a repo file using pkgrepo.managed
+    The issue is that CDROM installs often have [] in the line, and we
+    should still add the repo even though it's not setting arch(for example)
+    """
+    repo_file = str(tmp_path / "cdrom.list")
+    repo_content = "deb cdrom:[Debian GNU/Linux 11.4.0 _Bullseye_ - Official amd64 NETINST 20220709-10:31]/ stable main"
+    pkgrepo.managed(name=repo_content, file=repo_file, clean_file=True)
+    with salt.utils.files.fopen(repo_file, "r") as fp:
+        file_content = fp.read()
+        assert (
+            file_content.strip()
+            == "deb cdrom:[Debian GNU/Linux 11.4.0 _Bullseye_ - Official amd64 NETINST 20220709-10:31]/ stable main"
+        )
+
+
 def system_aptsources_ids(value):
     return "{}(aptsources.sourceslist)".format(value.title())
 

tests/pytests/unit/modules/test_aptpkg.py

@@ -852,8 +852,9 @@ def test__skip_source():
     assert ret is False
 
 
-def test__parse_source():
-    cases = (
+@pytest.mark.parametrize(
+    "case",
+    (
         {"ok": False, "line": "", "invalid": True, "disabled": False},
         {"ok": False, "line": "#", "invalid": True, "disabled": True},
         {"ok": False, "line": "##", "invalid": True, "disabled": True},
@@ -880,19 +881,31 @@ def test__parse_source():
             "invalid": False,
             "disabled": False,
         },
-    )
+        {
+            "ok": True,
+            "line": (
+                "# deb cdrom:[Debian GNU/Linux 11.4.0 _Bullseye_ - Official amd64 NETINST 20220709-10:31]/ bullseye main\n"
+                "\n"
+                "deb http://httpredir.debian.org/debian bullseye main\n"
+                "deb-src http://httpredir.debian.org/debian bullseye main\n"
+            ),
+            "invalid": False,
+            "disabled": True,
+        },
+    ),
+)
+def test__parse_source(case):
     with patch.dict("sys.modules", {"aptsources.sourceslist": None}):
         importlib.reload(aptpkg)
         NoAptSourceEntry = aptpkg.SourceEntry
     importlib.reload(aptpkg)
 
-    for case in cases:
-        source = NoAptSourceEntry(case["line"])
-        ok = source._parse_sources(case["line"])
+    source = NoAptSourceEntry(case["line"])
+    ok = source._parse_sources(case["line"])
 
-        assert ok is case["ok"]
-        assert source.invalid is case["invalid"]
-        assert source.disabled is case["disabled"]
+    assert ok is case["ok"]
+    assert source.invalid is case["invalid"]
+    assert source.disabled is case["disabled"]
 
 
 def test_normalize_name():
@@ -948,16 +961,10 @@ def test_list_repos():
                 assert repos[source_uri][0]["uri"][-1] == "/"
 
 
-@pytest.mark.skipif(
-    HAS_APTSOURCES is False, reason="The 'aptsources' library is missing."
-)
 def test_expand_repo_def():
     """
     Checks results from expand_repo_def
     """
-    source_type = "deb"
-    source_uri = "http://cdn-aws.deb.debian.org/debian/"
-    source_line = "deb http://cdn-aws.deb.debian.org/debian/ stretch main\n"
     source_file = "/etc/apt/sources.list"
 
     # Valid source
@@ -987,6 +994,40 @@ def test_expand_repo_def():
     )
 
 
+def test_expand_repo_def_cdrom():
+    """
+    Checks results from expand_repo_def
+    """
+    source_file = "/etc/apt/sources.list"
+
+    # Valid source
+    repo = "# deb cdrom:[Debian GNU/Linux 11.4.0 _Bullseye_ - Official amd64 NETINST 20220709-10:31]/ bullseye main\n"
+    sanitized = aptpkg.expand_repo_def(repo=repo, file=source_file)
+    log.warning("SAN: %s", sanitized)
+
+    assert isinstance(sanitized, dict)
+    assert "uri" in sanitized
+
+    # Make sure last character in of the URI is still a /
+    assert sanitized["uri"][-1] == "/"
+
+    # Pass the architecture and make sure it is added the the line attribute
+    repo = "deb http://cdn-aws.deb.debian.org/debian/ stretch main\n"
+    sanitized = aptpkg.expand_repo_def(
+        repo=repo, file=source_file, architectures="amd64"
+    )
+
+    # Make sure line is in the dict
+    assert isinstance(sanitized, dict)
+    assert "line" in sanitized
+
+    # Make sure the architecture is in line
+    assert (
+        sanitized["line"]
+        == "deb [arch=amd64] http://cdn-aws.deb.debian.org/debian/ stretch main"
+    )
+
+
 def test_list_pkgs():
     """
     Test packages listing.