update_manager: use sysdeps_parser

Use the sysdeps_parser utility to parse the correct dependencies
out of the data returned by the configured system dependencies
json file.

Signed-off-by:  Eric Callahan <[email protected]>

moonraker/components/update_manager/app_deploy.py

@@ -10,13 +10,13 @@
 import hashlib
 import logging
 import re
-import distro
 import asyncio
 import importlib
 from .common import AppType, Channel
 from .base_deploy import BaseDeploy
 from ...utils import pip_utils
 from ...utils import json_wrapper as jsonw
+from ...utils.sysdeps_parser import SysDepsParser
 
 # Annotation imports
 from typing import (
@@ -34,9 +34,6 @@
     from ..machine import Machine
     from ..file_manager.file_manager import FileManager
 
-DISTRO_ALIASES = [distro.id()]
-DISTRO_ALIASES.extend(distro.like().split())
-
 class AppDeploy(BaseDeploy):
     def __init__(self, config: ConfigHelper, prefix: str) -> None:
         super().__init__(config, prefix=prefix)
@@ -263,51 +260,6 @@ async def restart_service(self) -> None:
                     svc = kconn.unit_name
                 await machine.do_service_action("restart", svc)
 
-    def _convert_version(self, version: str) -> Tuple[str | int, ...]:
-        version = version.strip()
-        ver_match = re.match(r"\d+(\.\d+)*((?:-|\.).+)?", version)
-        if ver_match is not None:
-            return tuple([
-                int(part) if part.isdigit() else part
-                for part in re.split(r"\.|-", version)
-            ])
-        return (version,)
-
-    def _parse_system_dep(self, full_spec: str) -> str | None:
-        parts = full_spec.split(";", maxsplit=1)
-        if len(parts) == 1:
-            return full_spec
-        dep_parts = re.split(r"(==|!=|<=|>=|<|>)", parts[1].strip())
-        if len(dep_parts) != 3 or dep_parts[0].strip().lower() != "distro_version":
-            logging.info(f"Invalid requirement specifier: {full_spec}")
-            return None
-        pkg_name = parts[0].strip()
-        operator = dep_parts[1].strip()
-        distro_ver = self._convert_version(distro.version())
-        req_version = self._convert_version(dep_parts[2].strip())
-        try:
-            if operator == "<":
-                if distro_ver < req_version:
-                    return pkg_name
-            elif operator == ">":
-                if distro_ver > req_version:
-                    return pkg_name
-            elif operator == "==":
-                if distro_ver == req_version:
-                    return pkg_name
-            elif operator == "!=":
-                if distro_ver != req_version:
-                    return pkg_name
-            elif operator == ">=":
-                if distro_ver >= req_version:
-                    return pkg_name
-            elif operator == "<=":
-                if distro_ver <= req_version:
-                    return pkg_name
-        except TypeError:
-            pass
-        return None
-
     async def _read_system_dependencies(self) -> List[str]:
         eventloop = self.server.get_event_loop()
         if self.system_deps_json is not None:
@@ -320,26 +272,8 @@ async def _read_system_dependencies(self) -> List[str]:
             except Exception:
                 logging.exception(f"Error reading system deps: {deps_json}")
                 return []
-            for distro_id in DISTRO_ALIASES:
-                if distro_id in dep_info:
-                    if not dep_info[distro_id]:
-                        self.log_info(
-                            f"Dependency file '{deps_json.name}' contains an empty "
-                            f"package definition for linux distro '{distro_id}'"
-                        )
-                        continue
-                    processed_deps: List[str] = []
-                    for dep in dep_info[distro_id]:
-                        parsed_dep = self._parse_system_dep(dep)
-                        if parsed_dep is not None:
-                            processed_deps.append(parsed_dep)
-                    return processed_deps
-            else:
-                self.log_info(
-                    f"Dependency file '{deps_json.name}' has no package definition "
-                    f" for linux distro '{DISTRO_ALIASES[0]}'"
-                )
-                return []
+            parser = SysDepsParser()
+            return parser.parse_dependencies(dep_info)
         # Fall back on install script if configured
         if self.install_script is None:
             return []

moonraker/components/update_manager/python_deploy.py

@@ -11,8 +11,9 @@
 from enum import Enum
 from ...utils.source_info import normalize_project_name, load_distribution_info
 from ...utils.versions import PyVersion, GitVersion
+from ...utils.sysdeps_parser import SysDepsParser
 from ...utils import pip_utils, json_wrapper
-from .app_deploy import AppDeploy, Channel, DISTRO_ALIASES
+from .app_deploy import AppDeploy, Channel
 
 # Annotation imports
 from typing import (
@@ -208,20 +209,8 @@ def _parse_system_dependencies(self, package_info: PackageInfo) -> List[str]:
         if rinfo is None:
             return []
         dep_info = rinfo.get("system_dependencies", {})
-        for distro_id in DISTRO_ALIASES:
-            if distro_id in dep_info:
-                if not dep_info[distro_id]:
-                    self.log_info(
-                        f"Package release_info contains an empty system "
-                        f"package definition for linux distro '{distro_id}'"
-                    )
-                return dep_info[distro_id]
-        else:
-            self.log_info(
-                "Package release_info has no package definition "
-                f" for linux distro '{DISTRO_ALIASES[0]}'"
-            )
-            return []
+        parser = SysDepsParser()
+        return parser.parse_dependencies(dep_info)
 
     async def _update_local_state(self) -> None:
         self.warnings.clear()