Add file/directory management

src/decman/__init__.py

@@ -3,6 +3,153 @@ Module for writing system configurations for decman.
 """
 
 import typing
+import pwd
+import grp
+import shutil
+import os
+
+
+class File:
+    """
+    A simple file that gets copied to the target.
+    """
+
+    def __init__(
+        self,
+        source_file: typing.Optional[str] = None,
+        content: typing.Optional[str] = None,
+        bin_file: bool = False,
+        encoding: str = "utf-8",
+        owner: typing.Optional[str] = None,
+        group: typing.Optional[str] = None,
+        permissions: int = 0o644,
+    ):
+        if source_file is None and content is None:
+            raise ValueError("Both source_file and content cannot be None.")
+
+        if source_file is not None and content is not None:
+            raise ValueError("Both source_file and content cannot be set.")
+
+        self.source_file = source_file
+        self.content = content
+        self.permissions = permissions
+        self.bin_file = bin_file
+        self.encoding = encoding
+        self.uid = None
+        self.gid = None
+
+        if owner is not None:
+            self.uid = pwd.getpwnam(owner).pw_uid
+            self.gid = pwd.getpwnam(owner).pw_gid
+
+        if group is not None:
+            self.gid = grp.getgrnam(group).gr_gid
+
+    def copy_to(self,
+                target: str,
+                variables: typing.Optional[dict[str, str]] = None):
+        """
+        Copies the contents of this file to the target file.
+        """
+        if variables is None:
+            variables = {}
+
+        target_directory = os.path.dirname(target)
+        os.makedirs(target_directory, exist_ok=True)
+
+        self._write_content(target, variables)
+
+        if self.uid is not None:
+            assert self.gid is not None, "If uid is set, then gid is set."
+            os.chown(target, self.uid, self.gid)
+
+        os.chmod(target, self.permissions)
+
+    def _write_content(self, target: str, variables: dict[str, str]):
+        if self.source_file is not None and (self.bin_file
+                                             or len(variables) == 0):
+            shutil.copy(self.source_file, target)
+        elif self.bin_file and self.content is not None:
+            with open(target, "wb") as file:
+                file.write(self.content.encode(encoding=self.encoding))
+        elif self.source_file is not None:
+            with open(self.source_file, "rt", encoding=self.encoding) as src:
+                content = src.read()
+
+            for var, value in variables.items():
+                content = content.replace(var, value)
+
+            with open(target, "wt", encoding=self.encoding) as file:
+                file.write(content)
+        else:
+            assert self.content is not None, "Content should be set since source_file was not set."
+            content = self.content
+            for var, value in variables.items():
+                content = content.replace(var, value)
+
+            with open(target, "wt", encoding=self.encoding) as file:
+                file.write(content)
+
+
+class Directory:
+    """
+    Contents of this directory will be copied to the target.
+    """
+
+    def __init__(
+        self,
+        source_directory: str,
+        bin_files: bool = False,
+        encoding: str = "utf-8",
+        owner: typing.Optional[str] = None,
+        group: typing.Optional[str] = None,
+        permissions: int = 0o644,
+    ):
+        self.source_directory = source_directory
+        self.bin_files = bin_files
+        self.encoding = encoding
+        self.permissions = permissions
+
+        self.owner = owner
+        self.group = group
+        self.uid = None
+        self.gid = None
+
+        if owner is not None:
+            self.uid = pwd.getpwnam(owner).pw_uid
+            self.gid = pwd.getpwnam(owner).pw_gid
+
+        if group is not None:
+            self.gid = grp.getgrnam(group).gr_gid
+
+    def copy_to(
+            self,
+            target_directory: str,
+            variables: typing.Optional[dict[str, str]] = None) -> list[str]:
+        """
+        Copies the files in this directory to the target directory.
+
+        Returns all created files.
+        """
+        created = []
+        original_wd = os.getcwd()
+        try:
+            os.chdir(self.source_directory)
+            for src_dir, _, files in os.walk("."):
+                for src_file in files:
+                    src_path = os.path.join(src_dir, src_file)
+                    file = File(source_file=src_path,
+                                bin_file=self.bin_files,
+                                encoding=self.encoding,
+                                owner=self.owner,
+                                group=self.group,
+                                permissions=self.permissions)
+                    target = os.path.join(target_directory, src_path)
+                    created.append(target)
+                    file.copy_to(target, variables)
+        finally:
+            os.chdir(original_wd)
+        return created
 
 
 class UserPackage:
@@ -76,6 +223,25 @@ class Module:
         Override this method to run python code after the version of this module has changed.
         """
 
+    def files(self) -> dict[str, File]:
+        """
+        Override this method to return files that should be installed as a part of this module.
+        """
+        return {}
+
+    def directories(self) -> dict[str, Directory]:
+        """
+        Override this method to return directories that should be installed as a part of this module.
+        """
+        return {}
+
+    def file_variables(self) -> dict[str, str]:
+        """
+        Override this method to return variables that should replaced with a new value inside
+        this module's text files.
+        """
+        return {}
+
     def pacman_packages(self) -> list[str]:
         """
         Override this method to return pacman packages that should be installed as a part of this

src/decman/lib/__init__.py

@@ -134,6 +134,7 @@ class Store:
         self.enabled_systemd_units: list[str] = []
         self.enabled_user_systemd_units: list[tuple[str, str]] = []
         self.enabled_modules: dict[str, str] = {}
+        self.created_files: list[str] = []
         self.pkgbuild_latest_reviewed_commits: dict[str, str] = {}
         self._package_file_cache: dict[str, tuple[str, str]] = {}
 
@@ -170,6 +171,7 @@ class Store:
             "enabled_systemd_units": self.enabled_systemd_units,
             "enabled_user_systemd_units": self.enabled_user_systemd_units,
             "enabled_modules": self.enabled_modules,
+            "created_files": self.created_files,
             "package_file_cache": self._package_file_cache,
             "pkgbuild_git_commits": self.pkgbuild_latest_reviewed_commits
         }
@@ -208,6 +210,7 @@ class Store:
                     [],
                 )
                 store.enabled_modules = d.get("enabled_modules", {})
+                store.created_files = d.get("created_files", [])
                 store._package_file_cache = d.get("package_file_cache", {})
                 store.pkgbuild_latest_reviewed_commits = d.get(
                     "pkgbuild_git_commits",
@@ -243,6 +246,8 @@ class Source:
         ignored_packages: list[str],
         systemd_units: list[str],
         systemd_user_units: dict[str, list[str]],
+        files: dict[str, decman.File],
+        directories: dict[str, decman.Directory],
         modules: list[decman.Module],
     ):
         self.pacman_packages = pacman_packages
@@ -251,6 +256,8 @@ class Source:
         self.ignored_packages = ignored_packages
         self.systemd_units = systemd_units
         self.systemd_user_units = systemd_user_units
+        self.files = files
+        self.directories = directories
         self.modules = modules
 
     def run_on_enable(self, store: Store):
@@ -288,6 +295,55 @@ class Source:
             elif module.enabled and module.name not in store.enabled_modules:
                 module.after_version_change()
 
+    def create_all_files(self) -> list[str]:
+        """
+        Creates all files and returns them. The files created are based on the specified files,
+        directories and modules.
+        """
+        created_files = []
+
+        def install_files(files: dict[str, decman.File],
+                          variables: typing.Optional[dict[str, str]] = None):
+            for target, file in files.items():
+                created_files.append(target)
+                try:
+                    file.copy_to(target, variables)
+                    print_debug(f"Installing file to {target}.")
+                except OSError as e:
+                    raise UserFacingError(
+                        f"Failed to install file to {target}.") from e
+
+        def install_dirs(dirs: dict[str, decman.Directory],
+                         variables: typing.Optional[dict[str, str]] = None):
+            for target, directory in dirs.items():
+                try:
+                    print_debug(f"Installing directory to {target}.")
+                    directory.copy_to(target, variables)
+                except OSError as e:
+                    raise UserFacingError(
+                        f"Failed to install directory to {target}.") from e
+
+        install_files(self.files)
+        install_dirs(self.directories)
+
+        for module in self.modules:
+            if module.enabled:
+                install_files(module.files(), module.file_variables())
+                install_dirs(module.directories(), module.file_variables())
+
+        return created_files
+
+    def files_to_remove(self, store: Store,
+                        created_files: list[str]) -> list[str]:
+        """
+        Returns all files that should be removed.
+        """
+        to_remove = []
+        for path in store.created_files:
+            if path not in created_files:
+                to_remove.append(path)
+        return to_remove
+
     def units_to_enable(self, store: Store) -> list[str]:
         """
         Returns all systemd units that should be enabled.

tests/manual/src/f1.txt

@@ -0,0 +1,3 @@
+Simple text file with a %variable%
+
+twice: %another_variable%

tests/manual/src/f2.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+# This file should be executable.
+echo "Hello, world!"

tests/manual/src/srcdir/1

@@ -0,0 +1,3 @@
+1
+1
+1

tests/manual/src/srcdir/2

@@ -0,0 +1,3 @@
+2
+2
+2

tests/manual/src/srcdir/sub/s1

@@ -0,0 +1,3 @@
+s1
+s1
+s1

tests/manual/src/srcdir/sub/s2

@@ -0,0 +1,3 @@
+s2
+s2
+s2

tests/manual/test_file_creation.py

@@ -0,0 +1,61 @@
+import os
+import shutil
+import sys
+
+# This test is manual. You'll have to verify the results manually.
+
+# NOTE: Change this if you want to run this script.
+user = "kk"
+
+cd = os.path.dirname(os.path.abspath(__file__))
+os.chdir(cd)
+
+sys.path.append(os.path.join(cd, "../../src/."))
+
+from decman import File, Directory
+
+#if os.path.exists("/tmp/decman-files"):
+#    shutil.rmtree("/tmp/decman-files")
+#os.makedirs("/tmp/decman-files")
+
+f1 = File(source_file="src/f1.txt")
+f1.copy_to(
+    "/tmp/decman-files/f1.txt",
+    variables={
+        "%variable%": "123",
+        "%another_variable%": "456"
+    },
+)
+
+f2 = File(source_file="src/f2.sh", permissions=0o744)
+f2.copy_to("/tmp/decman-files/f2.sh", )
+
+f3 = File(content="%variable% doesn't work here.", bin_file=True)
+f3.copy_to(
+    "/tmp/decman-files/f3.txt",
+    variables={
+        "%variable%": "123",
+    },
+)
+
+f4 = File(content="%variable% works here.", bin_file=False, owner=user)
+f4.copy_to(
+    "/tmp/decman-files/f4.txt",
+    variables={
+        "%variable%": "123",
+    },
+)
+
+f5 = File(content="%variable% works here.",
+          bin_file=False,
+          owner=user,
+          group="root")
+f5.copy_to(
+    "/tmp/decman-files/f5.txt",
+    variables={
+        "%variable%": "123",
+    },
+)
+
+d = Directory("src/srcdir", bin_files=True)
+d.copy_to("/tmp/decman-files/targetdir")

tests/test_source_resolution.py

@@ -139,6 +139,8 @@ class TestSource(unittest.TestCase):
             systemd_units=["1.service", "2.timer"],
             systemd_user_units={"user": ["u1.service", "u2.timer"]},
             modules=modules,
+            files={},
+            directories={},
         )
 
         store = Store()
@@ -151,6 +153,7 @@ class TestSource(unittest.TestCase):
             "ExistingChanged": "1",
             "Disabled": "1",
         }
+        store.created_files = ["/test/file1", "/test/file2", "/test/file3"]
 
         currently_installed_packages = [
             "p1",
@@ -170,6 +173,12 @@ class TestSource(unittest.TestCase):
         self.store = store
         self.currently_installed_packages = currently_installed_packages
 
+    def test_files_to_remove(self):
+        created_files = ["/test/file1", "/test/file4"]
+        self.assertCountEqual(
+            self.source.files_to_remove(self.store, created_files),
+            ["/test/file2", "/test/file3"])
+
     def test_after_update_executed(self):
         self.source.run_after_update()