feat: add dnf charm library

.github/workflows/ci.yaml

@@ -35,7 +35,9 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v3
-      - uses: canonical/setup-lxd@90d76101915da56a42a562ba766b1a77019242fd
+      - uses: canonical/[email protected]
+        with:
+          channel: 5.9/stable
       - name: Install dependencies
         run: python -m pip install tox
       - name: Run integration tests

lib/charms/operator_libs_linux/v0/dnf.py

@@ -0,0 +1,207 @@
+# Copyright 2023 Canonical Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Abstractions for system's DNF package information and repositories."""
+
+import os
+import re
+import shutil
+import subprocess
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional, Union
+
+
+class Error(Exception):
+    """Raise when dnf encounters an execution error."""
+
+
+class _PackageState(Enum):
+    INSTALLED = "installed"
+    AVAILABLE = "available"
+    ABSENT = "absent"
+
+
+@dataclass(frozen=True)
+class PackageInfo:
+    """Dataclass representing DNF package information."""
+
+    name: str
+    arch: str = None
+    epoch: str = None
+    version: str = None
+    release: str = None
+    repo: str = None
+    _state: _PackageState = _PackageState.ABSENT
+
+    @property
+    def installed(self) -> bool:
+        """Determine if package is marked 'installed'."""
+        return self._state == _PackageState.INSTALLED
+
+    @property
+    def available(self) -> bool:
+        """Determine if package is marked 'available'."""
+        return self._state == _PackageState.AVAILABLE
+
+    @property
+    def absent(self) -> bool:
+        """Determine if package is marked 'absent'."""
+        return self._state == _PackageState.ABSENT
+
+    @property
+    def full_version(self) -> Optional[str]:
+        """Get full version of package."""
+        if self.absent:
+            return None
+
+        full_version = [self.version, f"-{self.release}"]
+        if self.epoch:
+            full_version.insert(0, f"{self.epoch}:")
+
+        return "".join(full_version)
+
+
+def version() -> str:
+    """Get version of `dnf` executable."""
+    return _dnf("--version").splitlines()[0]
+
+
+def installed() -> bool:
+    """Determine if the `dnf` executable is available on PATH."""
+    return shutil.which("dnf") is not None
+
+
+def upgrade(*packages: Optional[str]) -> None:
+    """Upgrade one or more packages.
+
+    Args:
+        *packages (Optional[str]):
+            Packages to upgrade on system. If packages is omitted,
+            upgrade all packages on the system.
+    """
+    _dnf("upgrade", *packages)
+
+
+def install(*packages: Union[str, os.PathLike]) -> None:
+    """Install one or more packages.
+
+    Args:
+        *packages (Union[str, os.PathLike]): Packages to install on the system.
+    """
+    if not packages:
+        raise TypeError("No packages specified.")
+    _dnf("install", *packages)
+
+
+def remove(*packages: str) -> None:
+    """Remove one or more packages from the system.
+
+    Args:
+        *packages (str): Packages to remove from system.
+    """
+    if not packages:
+        raise TypeError("No packages specified.")
+    _dnf("remove", *packages)
+
+
+def fetch(package: str) -> PackageInfo:
+    """Fetch information about a package.
+
+    Args:
+        package (str): Package to get information about.
+
+    Returns:
+        PackageInfo: Information about package.
+
+    Notes:
+        `package` needs to exactly match the name of the package that you are fetching.
+        For example, if working with the `python2` package on select EL distributions,
+        `dnf.install("python2")` will succeed, but `dnf.fetch("python2")` will return
+        the package in ABSENT state. This is because the name of the python2 package is
+        python2.7, not python2. To get info about the python2 package, you need to use
+        its exact name: `dnf.fetch("python2.7")`.
+    """
+    try:
+        stdout = _dnf("list", "-q", package)
+        # Only take top two lines of output.
+        status, info = stdout.splitlines()[:2]
+
+        # Check if package is in states INSTALLED or AVAILABLE. If not, mark absent.
+        if "Installed" in status:
+            state = _PackageState.INSTALLED
+        elif "Available" in status:
+            state = _PackageState.AVAILABLE
+        else:
+            raise Error(f"{package} is not installed or available but {status} instead.")
+
+        pkg_name, pkg_version, pkg_repo = info.split()
+        name, arch = pkg_name.rsplit(".", 1)
+
+        # Version should be good, but if not mark absent since package
+        # is probably in a bad state then.
+        version_match = re.match(r"(?:(.*):)?(.*)-(.*)", pkg_version)
+        if not version_match:
+            raise Error(f"Bad version {pkg_version}. Marking {package} absent.")
+        else:
+            epoch, version, release = version_match.groups()
+
+        return PackageInfo(
+            name=name,
+            arch=arch,
+            epoch=epoch,
+            version=version,
+            release=release,
+            repo=pkg_repo[1:] if pkg_repo.startswith("@") else pkg_repo,
+            _state=state,
+        )
+    except Error:
+        return PackageInfo(name=package)
+
+
+def add_repo(repo: str) -> None:  # pragma: no cover
+    """Add a new repository to DNF.
+
+    Args:
+        repo (str): URL of new repository to add.
+    """
+    if not fetch("dnf-plugins-core").installed:
+        install("dnf-plugins-core")
+    _dnf("config-manager", "--add-repo", repo)
+
+
+def _dnf(*args: str) -> str:
+    """Execute a DNF command.
+
+    Args:
+        *args (str): Arguments to pass to `dnf` executable.
+
+    Raises:
+        Error: Raised if DNF command execution fails.
+
+    Returns:
+        str: Captured stdout of executed DNF command.
+    """
+    try:
+        return subprocess.run(
+            ["dnf", "-y", *args],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            check=True,
+        ).stdout.strip("\n")
+    except FileNotFoundError:
+        raise Error(f"dnf not found on PATH {os.getenv('PATH')}")
+    except subprocess.CalledProcessError as e:
+        raise Error(f"{e} Reason:\n{e.stderr}")

tests/integration/test_dnf.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Integration tests for dnf charm library."""
+
+import charms.operator_libs_linux.v0.dnf as dnf
+
+
+def test_install_package() -> None:
+    dnf.install("epel-release")
+    dnf.install("slurm-slurmd", "slurm-slurmctld", "slurm-slurmdbd", "slurm-slurmrestd")
+    assert dnf.fetch("epel-release").installed
+    assert dnf.fetch("slurm-slurmd").installed
+    assert dnf.fetch("slurm-slurmctld").installed
+    assert dnf.fetch("slurm-slurmdbd").installed
+    assert dnf.fetch("slurm-slurmrestd").installed
+
+
+def test_query_dnf_and_package() -> None:
+    assert dnf.version() == "4.14.0"
+    package = dnf.fetch("slurm-slurmd")
+    assert package.installed
+    assert not package.available
+    assert not package.absent
+    assert package.name == "slurm-slurmd"
+    assert package.arch == "x86_64"
+    assert package.epoch is None
+    assert package.version == "22.05.6"
+    assert package.release == "3.el9"
+    assert package.full_version == "22.05.6-3.el9"
+    assert package.repo == "epel"
+
+
+def test_remove_package() -> None:
+    dnf.remove("slurm-slurmdbd")
+    dnf.remove("slurm-slurmd", "slurm-slurmrestd", "slurm-slurmctld")
+    assert dnf.fetch("slurm-slurmdbd").available
+    assert dnf.fetch("slurm-slurmd").available
+    assert dnf.fetch("slurm-slurmrestd").available
+    assert dnf.fetch("slurm-slurmctld").available
+
+
+def test_check_absent() -> None:
+    bogus = dnf.fetch("nuccitheboss")
+    assert not bogus.installed
+    assert not bogus.available
+    assert bogus.absent
+    assert bogus.name == "nuccitheboss"
+    assert bogus.arch is None
+    assert bogus.epoch is None
+    assert bogus.version is None
+    assert bogus.release is None
+    assert bogus.full_version is None
+    assert bogus.repo is None
+
+
+def test_add_repo() -> None:
+    dnf.add_repo("http://mirror.stream.centos.org/9-stream/HighAvailability/x86_64/os")
+    dnf.install("pacemaker")
+    assert dnf.fetch("pacemaker").installed