Adding an exclusion option for restic

Author: MarcSchuhTNG

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "opsbox"
-version = "0.1.5"
+version = "0.1.6"
 description = "A comprehensive Python library for server operations including backup scripts, encrypted mail functionality, and utility tools"
 readme = "README.md"
 license = {text = "MIT"}

src/opsbox/backup/__init__.py

@@ -5,6 +5,7 @@
     BackupEnvironmentError,
     BackupError,
     ConfigurationError,
+    FolderNotFoundError,
     InvalidResticConfigError,
     MaintenanceError,
     NetworkUnreachableError,
@@ -31,6 +32,7 @@
     "BackupScript",
     "ConfigManager",
     "ConfigurationError",
+    "FolderNotFoundError",
     "InvalidResticConfigError",
     "MaintenanceError",
     "NetworkChecker",

src/opsbox/backup/exceptions.py

@@ -89,3 +89,7 @@ class DiffParsingError(BackupError):
 
 class InvalidSnapshotIDError(BackupError):
     """Raised when snapshot ID format is invalid."""
+
+
+class FolderNotFoundError(BackupError):
+    """Raised when a required folder or file is not found."""

src/opsbox/backup/restic_backup.py

@@ -55,7 +55,7 @@ class BackupScript:
     """Refactored backup script with improved architecture and error handling."""
 
     MIN_SNAPSHOTS_FOR_DIFF = 2
-    MAX_FILES_IN_EMAIL = 100
+    MAX_FILES_IN_EMAIL = 200
 
     def __init__(
         self,

src/opsbox/rsync/config.example.yaml

@@ -43,5 +43,9 @@ rsync_options:
   # Show progress during transfer
   progress: true
 
+# Optional: Path to file containing patterns to exclude from sync (one pattern per line)
+# If specified, the file must exist or a FolderNotFoundError will be raised
+exclude_file: null # e.g., "/home/ben/exclude_me.txt"
+
 # Optional: Title for the rsync operation (defaults to "Default rsync title")
 rsync_title: "Default rsync title"

src/opsbox/rsync/rsync_manager.py

@@ -18,6 +18,7 @@
 
 from opsbox.backup.exceptions import (
     ConfigurationError,
+    FolderNotFoundError,
     NetworkUnreachableError,
     SSHKeyNotFoundError,
     UserDoesNotExistError,
@@ -74,6 +75,7 @@ class RsyncConfig:
     rsync_title: str = "Default rsync title"
     rsync_options: dict[str, Any] = field(default_factory=dict)
     default_user: str = field(default_factory=lambda: getpass.getuser())
+    exclude_file: Path | None = None
 
     def __post_init__(self) -> None:
         """Validate configuration after initialization."""
@@ -113,6 +115,9 @@ def _validate_paths(self) -> None:
         if not self.email_settings_path.exists():
             error_msg = f"Email settings file not found: {self.email_settings_path}"
             raise ConfigurationError(error_msg)
+        if self.exclude_file is not None and not self.exclude_file.exists():
+            error_msg = f"Exclude file not found: {self.exclude_file}"
+            raise FolderNotFoundError(error_msg)
 
 
 class RsyncManager:
@@ -207,6 +212,7 @@ def _load_config(config_path: Path) -> RsyncConfig:
 
         try:
             # Create RsyncConfig instance with validation
+            exclude_file = config_data.get("exclude_file")
             return RsyncConfig(
                 rsync_source=config_data["rsync_source"],
                 rsync_target=config_data["rsync_target"],
@@ -220,6 +226,7 @@ def _load_config(config_path: Path) -> RsyncConfig:
                 rsync_options=config_data.get("rsync_options", {}),
                 default_user=config_data.get("default_user", getpass.getuser()),
                 rsync_title=config_data.get("rsync_title", "Default rsync title"),
+                exclude_file=Path(exclude_file) if exclude_file else None,
             )
 
         except KeyError as e:
@@ -291,6 +298,10 @@ def _build_rsync_command(self) -> list[str]:
         if self.config.rsync_options.get("progress", False):
             cmd.append("--progress")
 
+        # Add exclude file if specified
+        if self.config.exclude_file is not None:
+            cmd.extend(["--exclude-from", str(self.config.exclude_file)])
+
         # Add log file
         cmd.extend(["--log-file", str(self.log_file_path)])
 

tests/rsync/test_rsync_manager.py

@@ -9,6 +9,7 @@
 
 from opsbox.backup.exceptions import (
     ConfigurationError,
+    FolderNotFoundError,
     NetworkUnreachableError,
     SSHKeyNotFoundError,
 )
@@ -177,6 +178,52 @@ def test_config_email_settings_path_not_exists(self) -> None:
 
         assert "Email settings file not found" in str(exc_info.value)
 
+    def test_config_exclude_file_not_exists(self) -> None:
+        """Test raises FolderNotFoundError when exclude_file doesn't exist."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            email_settings = self._create_test_email_settings(temp_path)
+
+            with pytest.raises(FolderNotFoundError):
+                RsyncConfig(
+                    rsync_source="user@host:/source",
+                    rsync_target="/target",
+                    email_settings_path=email_settings,
+                    exclude_file=Path("/nonexistent/exclude.txt"),
+                )
+
+    def test_config_exclude_file_optional(self) -> None:
+        """Test that exclude_file is optional and None is valid."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            email_settings = self._create_test_email_settings(temp_path)
+
+            config = RsyncConfig(
+                rsync_source="user@host:/source",
+                rsync_target="/target",
+                email_settings_path=email_settings,
+                exclude_file=None,
+            )
+
+            assert config.exclude_file is None
+
+    def test_config_exclude_file_valid(self) -> None:
+        """Test that exclude_file is accepted when file exists."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            email_settings = self._create_test_email_settings(temp_path)
+            exclude_file = temp_path / "exclude.txt"
+            exclude_file.write_text("*.tmp\n*.log\n")
+
+            config = RsyncConfig(
+                rsync_source="user@host:/source",
+                rsync_target="/target",
+                email_settings_path=email_settings,
+                exclude_file=exclude_file,
+            )
+
+            assert config.exclude_file == exclude_file
+
 
 class TestRsyncManagerInitialization:
     """Test cases for RsyncManager initialization."""
@@ -349,6 +396,89 @@ def test_initialization_missing_required_field(self) -> None:
 
             assert "Missing required configuration field" in str(exc_info.value)
 
+    @patch("opsbox.rsync.rsync_manager.configure_logging")
+    @patch("opsbox.rsync.rsync_manager.EncryptedMail")
+    @patch("opsbox.rsync.rsync_manager.LockManager")
+    @patch("opsbox.rsync.rsync_manager.NetworkChecker")
+    @patch("opsbox.rsync.rsync_manager.SSHManager")
+    def test_initialization_with_exclude_file(
+        self,
+        mock_ssh_manager: MagicMock,
+        mock_network_checker: MagicMock,
+        mock_lock_manager: MagicMock,
+        mock_encrypted_mail: MagicMock,
+        mock_configure_logging: MagicMock,
+    ) -> None:
+        """Test initializes successfully with exclude_file in config."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            email_settings = self._create_test_email_settings(temp_path)
+            exclude_file = temp_path / "exclude.txt"
+            exclude_file.write_text("*.tmp\n*.log\n")
+
+            config_data = {
+                "rsync_source": "user@host:/source",
+                "rsync_target": str(temp_path / "target"),
+                "email_settings_path": str(email_settings),
+                "exclude_file": str(exclude_file),
+            }
+            config_file = self._create_test_config_file(
+                temp_path,
+                email_settings,
+                config_data,
+            )
+
+            target_dir = temp_path / "target"
+            target_dir.mkdir()
+
+            mock_logger = Mock()
+            mock_configure_logging.return_value = mock_logger
+
+            manager = RsyncManager(config_path=config_file, log_level="INFO")
+
+            assert manager.config.exclude_file == exclude_file
+
+    @patch("opsbox.rsync.rsync_manager.configure_logging")
+    @patch("opsbox.rsync.rsync_manager.EncryptedMail")
+    @patch("opsbox.rsync.rsync_manager.LockManager")
+    @patch("opsbox.rsync.rsync_manager.NetworkChecker")
+    @patch("opsbox.rsync.rsync_manager.SSHManager")
+    def test_initialization_with_exclude_file_not_exists(
+        self,
+        mock_ssh_manager: MagicMock,
+        mock_network_checker: MagicMock,
+        mock_lock_manager: MagicMock,
+        mock_encrypted_mail: MagicMock,
+        mock_configure_logging: MagicMock,
+    ) -> None:
+        """Test raises FolderNotFoundError when exclude_file in config doesn't exist."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            email_settings = self._create_test_email_settings(temp_path)
+
+            config_data = {
+                "rsync_source": "user@host:/source",
+                "rsync_target": str(temp_path / "target"),
+                "email_settings_path": str(email_settings),
+                "exclude_file": "/nonexistent/exclude.txt",
+            }
+            config_file = self._create_test_config_file(
+                temp_path,
+                email_settings,
+                config_data,
+            )
+
+            target_dir = temp_path / "target"
+            target_dir.mkdir()
+
+            mock_logger = Mock()
+            mock_configure_logging.return_value = mock_logger
+
+            with pytest.raises(ConfigurationError) as exc_info:
+                RsyncManager(config_path=config_file, log_level="INFO")
+
+            assert "Exclude file not found" in str(exc_info.value)
+
 
 class TestRsyncCommandBuilding:
     """Test cases for rsync command building."""
@@ -463,6 +593,69 @@ def test_build_rsync_command_with_log_file(self) -> None:
             log_file_index = cmd.index("--log-file")
             assert cmd[log_file_index + 1] == str(manager.log_file_path)
 
+    def test_build_rsync_command_with_exclude_file(self) -> None:
+        """Test includes exclude-from option when exclude_file is configured."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            exclude_file = temp_path / "exclude.txt"
+            exclude_file.write_text("*.tmp\n*.log\n")
+
+            config_data = {
+                "rsync_source": "user@host:/source",
+                "rsync_target": str(temp_path / "target"),
+                "email_settings_path": str(temp_path / "email_settings.yaml"),
+                "exclude_file": str(exclude_file),
+            }
+            manager = self._create_test_manager(temp_path, config_data)
+
+            cmd = manager._build_rsync_command()
+
+            assert "--exclude-from" in cmd
+            exclude_index = cmd.index("--exclude-from")
+            assert cmd[exclude_index + 1] == str(exclude_file)
+
+    def test_build_rsync_command_without_exclude_file(self) -> None:
+        """Test does not include exclude-from option when exclude_file is not configured."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            manager = self._create_test_manager(temp_path)
+
+            cmd = manager._build_rsync_command()
+
+            assert "--exclude-from" not in cmd
+
+    def test_build_rsync_command_with_exclude_file_and_other_options(self) -> None:
+        """Test exclude-from works correctly with other rsync options."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir)
+            exclude_file = temp_path / "exclude.txt"
+            exclude_file.write_text("*.tmp\n*.log\n")
+
+            config_data = {
+                "rsync_source": "user@host:/source",
+                "rsync_target": str(temp_path / "target"),
+                "email_settings_path": str(temp_path / "email_settings.yaml"),
+                "exclude_file": str(exclude_file),
+                "rsync_options": {
+                    "chown": "user:group",
+                    "delete": True,
+                    "progress": True,
+                },
+            }
+            manager = self._create_test_manager(temp_path, config_data)
+
+            cmd = manager._build_rsync_command()
+
+            # Verify all options are present
+            assert "--exclude-from" in cmd
+            assert "--chown" in cmd
+            assert "--delete" in cmd
+            assert "--progress" in cmd
+            # Verify exclude-from comes before log-file (as per implementation)
+            exclude_index = cmd.index("--exclude-from")
+            log_file_index = cmd.index("--log-file")
+            assert exclude_index < log_file_index
+
 
 class TestRsyncUtilityFunctions:
     """Test cases for utility functions."""