Repository: osbuild/osbuild
Pull request #1615 (2024-02-27 16:31:47)

util: Consolidate parse functions into util
 - Move functions to the 'util' to centralize common functionality, reducing code duplication and improving maintainability across the codebase.

bcl: Looks like a good idea, just need to check the lint errors :)
mvo5: Hey @ravanelli I really like this PR and there are only a few small comments to address. Will you have time for them? If not I'm happy to just do the small fixups myself to land this (as I think it's a really nice cleanup, thanks for that!) :)
ravanelli: > Hey @ravanelli I really like this PR and there are only a few small comments to address. Will you have time for them? If not I'm happy to just do the small fixups myself to land this (as I think it's a really nice cleanup, thanks for that!) :)

@mvo5 sorry about the delay, we got some fires with the podman v5 release that I needed to focus on. I hope to address the last comments  by Monday.
mvo5: > > Hey @ravanelli I really like this PR and there are only a few small comments to address. Will you have time for them? If not I'm happy to just do the small fixups myself to land this (as I think it's a really nice cleanup, thanks for that!) :)
> 
> @mvo5 sorry about the delay, we got some fires with the podman v5 release that I needed to focus on. I hope to address the last comments by Monday.

No problem at all! Just wanted to double check if I could help :)
ravanelli: > > > Hey @ravanelli I really like this PR and there are only a few small comments to address. Will you have time for them? If not I'm happy to just do the small fixups myself to land this (as I think it's a really nice cleanup, thanks for that!) :)
> > 
> > 
> > @mvo5 sorry about the delay, we got some fires with the podman v5 release that I needed to focus on. I hope to address the last comments by Monday.
> 
> No problem at all! Just wanted to double check if I could help :)

@mvo5 thank you so much for all the help, I had to rebase and got some conflicts to go back to the initial commit to fix it there, so I added a new commit with the requested changes,  if you don't mind.
ravanelli: @mvo5 Can you help me understand why the ci still failing? I did a new rebase again.
mvo5: > @mvo5 Can you help me understand why the ci still failing? I did a new rebase again.

This looks like flakiness, I added a retry.

Patch:
diff --git a/osbuild/util/parsing.py b/osbuild/util/parsing.py
index 303fbefa..f8fb2768 100644
--- a/osbuild/util/parsing.py
+++ b/osbuild/util/parsing.py
@@ -1,7 +1,9 @@
 """Helpers related to parsing"""
 
+import os
 import re
-from typing import Union
+from typing import Dict, Union
+from urllib.parse import ParseResult, urlparse
 
 
 def parse_size(s: str) -> Union[int, str]:
@@ -31,3 +33,75 @@ def parse_size(s: str) -> Union[int, str]:
                 return "unlimited"
 
     raise TypeError(f"invalid size value: '{s}'")
+
+
+def find_mount_root(url: ParseResult, args: Dict) -> os.PathLike:
+    """
+    Parses the mount URL to extract the root path.
+
+    Parameters:
+    - url (ParseResult): The ParseResult object obtained from urlparse.
+    - args (Dict):A dictionary containing arguments including mounts and
+    path information as passed by osbuild.api.arguments()
+    """
+    name = url.netloc
+    if name:
+        root = args["mounts"].get(name, {}).get("path")
+        if root is None:
+            raise ValueError(f"Unknown mount '{name}'")
+    else:
+        root = args["paths"]["mounts"]
+
+    return root
+
+
+def parse_input(url: ParseResult, args: Dict) -> os.PathLike:
+    """
+    Parses the input URL to extract the root path.
+
+    Parameters:
+    - url (ParseResult): The ParseResult object obtained from urlparse.
+    - args (Dict): A dictionary containing arguments including mounts and
+    path information as passed by osbuild.api.arguments()
+    """
+    name = url.netloc
+    root = args["inputs"].get(name, {}).get("path")
+    if root is None:
+        raise ValueError(f"Unknown input '{name}'")
+
+    return root
+
+
+def parse_location(location: str, args: Dict) -> str:
+    """
+    Parses the location URL to derive the corresponding file path.
+
+    Parameters:
+    - location (str): The location URL to be parsed.
+    - args (Dict): A dictionary containing arguments including mounts and
+    path information as passed by osbuild.api.arguments()
+    """
+
+    url = urlparse(location)
+
+    scheme = url.scheme
+    if scheme == "tree":
+        root = args["tree"]
+    elif scheme == "mount":
+        root = find_mount_root(url, args)
+    elif scheme == "input":
+        root = parse_input(url, args)
+    else:
+        raise ValueError(f"Unsupported scheme '{scheme}'")
+
+    if not url.path.startswith("/"):
+        raise ValueError(f"url.path from location must start with '/', got: {url.path}")
+
+    path = os.path.relpath(url.path, "/")
+    path = os.path.join(root, path)
+    path = os.path.normpath(path)
+
+    if url.path.endswith("/"):
+        path = os.path.join(path, ".")
+
+    return path
diff --git a/stages/org.osbuild.chattr b/stages/org.osbuild.chattr
index b1e6ad88..4b714c79 100755
--- a/stages/org.osbuild.chattr
+++ b/stages/org.osbuild.chattr
@@ -1,52 +1,15 @@
 #!/usr/bin/python3
-import os
 import subprocess
 import sys
-from typing import Dict
-from urllib.parse import ParseResult, urlparse
 
 import osbuild.api
-
-
-def parse_mount(url: ParseResult, args: Dict):
-    name = url.netloc
-    if name:
-        root = args["mounts"].get(name, {}).get("path")
-        if not root:
-            raise ValueError(f"Unknown mount '{name}'")
-    else:
-        root = args["paths"]["mounts"]
-
-    return root
-
-
-def parse_location(location, args):
-    url = urlparse(location)
-
-    scheme = url.scheme
-    if scheme == "tree":
-        root = args["tree"]
-    elif scheme == "mount":
-        root = parse_mount(url, args)
-    else:
-        raise ValueError(f"Unsupported scheme '{scheme}'")
-
-    assert url.path.startswith("/")
-
-    path = os.path.relpath(url.path, "/")
-    path = os.path.join(root, path)
-    path = os.path.normpath(path)
-
-    if url.path.endswith("/"):
-        path = os.path.join(path, ".")
-
-    return path
+from osbuild.util import parsing
 
 
 def main(args, options):
     for path, cmdargs in options["items"].items():
         immutable = cmdargs["immutable"]
-        dst = parse_location(path, args)
+        dst = parsing.parse_location(path, args)
         op = '+' if immutable else '-'
         subprocess.run(["chattr", f"{op}i", dst], check=True)
 
diff --git a/stages/org.osbuild.copy b/stages/org.osbuild.copy
index 90766a14..433e592f 100755
--- a/stages/org.osbuild.copy
+++ b/stages/org.osbuild.copy
@@ -1,65 +1,17 @@
 #!/usr/bin/python3
-import os
 import subprocess
 import sys
-from typing import Dict
-from urllib.parse import ParseResult, urlparse
 
 import osbuild.api
-
-
-def parse_mount(url: ParseResult, args: Dict):
-    name = url.netloc
-    if name:
-        root = args["mounts"].get(name, {}).get("path")
-        if not root:
-            raise ValueError(f"Unknown mount '{name}'")
-    else:
-        root = args["paths"]["mounts"]
-
-    return root
-
-
-def parse_input(url: ParseResult, args: Dict):
-    name = url.netloc
-    root = args["inputs"].get(name, {}).get("path")
-    if root is None:
-        raise ValueError(f"Unknown input '{name}'")
-
-    return root
-
-
-def parse_location(location, args):
-    url = urlparse(location)
-
-    scheme = url.scheme
-    if scheme == "tree":
-        root = args["tree"]
-    elif scheme == "mount":
-        root = parse_mount(url, args)
-    elif scheme == "input":
-        root = parse_input(url, args)
-    else:
-        raise ValueError(f"Unsupported scheme '{scheme}'")
-
-    assert url.path.startswith("/")
-
-    path = os.path.relpath(url.path, "/")
-    path = os.path.join(root, path)
-    path = os.path.normpath(path)
-
-    if url.path.endswith("/"):
-        path = os.path.join(path, ".")
-
-    return path
+from osbuild.util import parsing
 
 
 def main(args, options):
     items = options["paths"]
 
     for path in items:
-        src = parse_location(path["from"], args)
-        dst = parse_location(path["to"], args)
+        src = parsing.parse_location(path["from"], args)
+        dst = parsing.parse_location(path["to"], args)
         remove_destination = path.get("remove_destination", False)
 
         print(f"copying '{src}' -> '{dst}'")
diff --git a/stages/org.osbuild.kernel-cmdline.bls-append b/stages/org.osbuild.kernel-cmdline.bls-append
index eb6dcf44..72873498 100755
--- a/stages/org.osbuild.kernel-cmdline.bls-append
+++ b/stages/org.osbuild.kernel-cmdline.bls-append
@@ -1,30 +1,19 @@
 #!/usr/bin/python3
 import sys
-from urllib.parse import urlparse
 
 import osbuild.api
-from osbuild.util import bls
+from osbuild.util import bls, parsing
 
 
-def main(paths, tree, options):
+def main(args, options):
     kopts = options.get("kernel_opts", [])
     bootpath = options.get("bootpath", "tree:///boot")
 
-    url = urlparse(bootpath)
-    scheme = url.scheme
-    if scheme == "tree":
-        root = tree
-    elif scheme == "mount":
-        root = paths["mounts"]
-    else:
-        raise ValueError(f"Unsupported scheme '{scheme}'")
-
-    assert url.path.startswith("/")
-    bootroot = root + url.path
+    bootroot = parsing.parse_location(bootpath, args)
     bls.options_append(bootroot, kopts)
 
 
 if __name__ == '__main__':
-    args = osbuild.api.arguments()
-    r = main(args["paths"], args["tree"], args["options"])
+    _args = osbuild.api.arguments()
+    r = main(_args, _args["options"])
     sys.exit(r)
diff --git a/tox.ini b/tox.ini
index dbaf7a94..505719c5 100644
--- a/tox.ini
+++ b/tox.ini
@@ -28,7 +28,7 @@ setenv =
     LINTABLES_EXCLUDES = "*.json,*.sh"
     LINTABLES_EXCLUDES_RE = ".*\.json$,.*\.sh"
     TYPEABLES = osbuild
-    TYPEABLES_STRICT = ./osbuild/main_cli.py
+    TYPEABLES_STRICT = ./osbuild/main_cli.py ./osbuild/util/parsing.py
 
 passenv =
     TEST_CATEGORY


Test patch:
diff --git a/test/mod/test_util_parsing.py b/test/mod/test_util_parsing.py
index faaa31a5..e501310c 100644
--- a/test/mod/test_util_parsing.py
+++ b/test/mod/test_util_parsing.py
@@ -37,3 +37,41 @@ def test_parse_size():
         else:
             res = parsing.parse_size(s)
             assert res == num, f"{s} parsed as {res} (wanted {num})"
+
+
+def test_parse_location_mounts():
+    args = {
+        "paths": {
+            "mounts": "/run/osbuild/mounts",
+        },
+        "mounts": {
+            "root": {
+                "path": "/run/osbuild/mounts/.",
+            },
+        },
+    }
+    location = "mount://root/"
+    path = parsing.parse_location(location, args)
+    assert path == "/run/osbuild/mounts/."
+
+
+def test_parse_location_tree():
+    args = {
+        "tree": "/run/osbuild/tree",
+    }
+    location = "tree:///disk.img"
+    path = parsing.parse_location(location, args)
+    assert path == "/run/osbuild/tree/disk.img"
+
+
+def test_parse_location_inputs():
+    args = {
+        "inputs": {
+            "tree": {
+                "path": "/run/osbuild/inputs/tree",
+            },
+        },
+    }
+    location = "input://tree/"
+    path = parsing.parse_location(location, args)
+    assert path == "/run/osbuild/inputs/tree/."