Fix linting

Makefile

@@ -1,17 +1,26 @@
-.PHONY: test
+.PHONY: test coverage dist lint fmt
+
 test:
 	# TODO: figure out why -s option is required for fabric remote connection
 	hatch run test -s -vv
 
-.PHONY: coverage
+coverage:
 	hatch run cov -s -vv
 
-.PHONY: dist
+lint:
+	hatch run lint:all
+
+fmt:
+	hatch run lint:fmt
+
+# dist:
+# 	# todo move all this to script in pyproject.toml
+# 	python3 -m pip install --upgrade build
+# 	python3 -m build
+# 	python3 -m pip install --upgrade twine
+# 	echo "CHECK THIS VERSION NUMBER AND CHANGE IF NECESSARY!"
+# 	cat ./src/pisync/__about__.py
+# 	python3 -m twine upload dist/*
+
 dist:
-	# todo move all this to script in pyproject.toml
-	python3 -m pip install --upgrade build
-	python3 -m build
-	python3 -m pip install --upgrade twine
-	echo "CHECK THIS VERSION NUMBER AND CHANGE IF NECESSARY!"
-	cat ./src/pisync/__about__.py
-	python3 -m twine upload dist/*
+	hatch publish

pyproject.toml

@@ -60,6 +60,8 @@ dependencies = [
   "black>=23.1.0",
   "mypy>=1.0.0",
   "ruff>=0.0.243",
+  "pytest",
+  "pisync"
 ]
 [tool.hatch.envs.lint.scripts]
 typing = "mypy --install-types --non-interactive {args:src/pisync tests}"
@@ -77,6 +79,12 @@ all = [
   "typing",
 ]
 
+[[tool.mypy.overrides]]
+module = [
+	"fabric"
+]
+ignore_missing_imports = true
+
 [tool.black]
 target-version = ["py37"]
 line-length = 120
@@ -121,6 +129,8 @@ ignore = [
   "S105", "S106", "S107",
   # Ignore complexity
   "C901", "PLR0911", "PLR0912", "PLR0913", "PLR0915",
+  # I do not know how to fix this `subprocess` call: check for execution of untrusted input
+  "S603"
 ]
 unfixable = [
   # Don't touch unused imports
@@ -137,6 +147,9 @@ ban-relative-imports = "all"
 # Tests can use magic values, assertions, and relative imports
 "tests/**/*" = ["PLR2004", "S101", "TID252"]
 
+# allow print statement in example script
+"./examples/run_backups.py" = ["T201"]
+
 [tool.coverage.run]
 source_pkgs = ["pisync", "tests"]
 branch = true

src/pisync/__init__.py

@@ -1,2 +1,4 @@
 from pisync.backup import backup
 from pisync.config import LocalConfig, RemoteConfig
+
+__all__ = ("backup", "LocalConfig", "RemoteConfig")

src/pisync/backup.py

@@ -16,10 +16,10 @@
 from pathlib import Path
 from typing import List
 
-from pisync.config.base_config import _BaseConfig
+from pisync.config.base_config import BackupType, BaseConfig
 
 
-def backup(config: _BaseConfig) -> str:
+def backup(config: BaseConfig) -> str:
     """
     Returns the path to the latest backup directory
     """
@@ -44,12 +44,14 @@ def backup(config: _BaseConfig) -> str:
         raise Exception(msg)
 
     if prev_backup_exists:
+        backup_method = BackupType.Incremental
         logging.info(f"Starting incremental backup from {config.resolve(config.link_dir)}")
     else:
+        backup_method = BackupType.Complete
         logging.info(f"No previous backup found at {config.destination_dir}")
-        logging.info(f"Starting a fresh complete backup from {config.source_dir} " "to {config.destination_dir}")
+        logging.info(f"Starting a fresh complete backup from {config.source_dir} to {config.destination_dir}")
 
-    rsync_command = config.get_rsync_command(latest_backup_path, previous_backup_exists=prev_backup_exists)
+    rsync_command = config.get_rsync_command(latest_backup_path, backup_method=backup_method)
 
     exit_code = run_rsync(rsync_command)
 
@@ -61,21 +63,22 @@ def backup(config: _BaseConfig) -> str:
         logging.info(f"Symlink created from {latest_backup_path} to {config.link_dir}")
         return latest_backup_path
     else:
+        msg = f"Backup failed. Rsync exit code: {exit_code}"
+        logging.error(msg)
         # backup failed, we should delete the most recent backup
-        logging.error(f"Backup failed. Rsync exit code: {exit_code}")
         if config.file_exists(latest_backup_path):
             logging.info(f"Deleting failed backup at {latest_backup_path}")
             shutil.rmtree(latest_backup_path)
-        return None
+        raise Exception(msg)
 
 
 def configure_logging(filename: str):
-    filename = Path(filename)
-    if not filename.parent.exists():
-        filename.parent.mkdir(parents=True)
+    _filename = Path(filename)
+    if not _filename.parent.exists():
+        _filename.parent.mkdir(parents=True)
 
     logging.basicConfig(
-        filename=str(filename.resolve()),
+        filename=str(_filename.resolve()),
         filemode="a",
         format="%(levelname)s\t%(asctime)s\t%(message)s",
         datefmt="%m/%d/%Y %I:%M:%S %p",

src/pisync/config/__init__.py

@@ -1,3 +1,5 @@
-from pisync.config.base_config import InvalidPath
+from pisync.config.base_config import BackupType, InvalidPathError
 from pisync.config.local_config import LocalConfig
 from pisync.config.remote_config import RemoteConfig
+
+__all__ = ("InvalidPathError", "BackupType", "LocalConfig", "RemoteConfig")

src/pisync/config/base_config.py

@@ -1,15 +1,21 @@
 from abc import ABC, abstractmethod
-from typing import List
+from enum import Enum
+from typing import List, Optional
 
 
-class InvalidPath(Exception):
+class InvalidPathError(Exception):
     pass
 
 
-class _BaseConfig(ABC):
+class BackupType(Enum):
+    Complete = 1
+    Incremental = 2
+
+
+class BaseConfig(ABC):
     source_dir: str
     destination_dir: str
-    exclude_file_patterns: List[str]
+    exclude_file_patterns: Optional[List[str]]
     log_file: str
     link_dir: str
 
@@ -44,11 +50,11 @@ def resolve(self, path: str) -> str:
         pass
 
     @abstractmethod
-    def ensure_dir_exists(self, path: str):
+    def ensure_dir_exists(self, path: str) -> None:
         """
         :returns: The Path object from the input path str
         :raises:
-            InvalidPath: If path does not exist or is not a directory
+            InvalidPathError: If path does not exist or is not a directory
         """
         pass
 
@@ -58,10 +64,10 @@ def generate_new_backup_dir_path(self) -> str:
         :returns: The Path string of the directory where the new backup will be
         written.
         :raises:
-            InvalidPath: If the destination directory already exists
+            InvalidPathError: If the destination directory already exists
         """
         pass
 
     @abstractmethod
-    def get_rsync_command(self, new_backup_dir: str, previous_backup_exists: bool = False):
+    def get_rsync_command(self, new_backup_dir: str, backup_method: BackupType):
         pass

src/pisync/config/local_config.py

@@ -1,24 +1,27 @@
 from pathlib import Path
 from typing import List, Optional
 
-from pisync.config.base_config import InvalidPath, _BaseConfig
+from pisync.config.base_config import BackupType, BaseConfig, InvalidPathError
 from pisync.util import get_time_stamp
 
 
-class LocalConfig(_BaseConfig):
+class LocalConfig(BaseConfig):
     def __init__(
         self,
         source_dir: str,
         destination_dir: str,
         exclude_file_patterns: Optional[List[str]] = None,
-        log_file: str = str(Path.home() / ".local/share/backup/rsync-backups.log"),
+        log_file: Optional[str] = None,
     ):
         self.ensure_dir_exists(source_dir)
         self.ensure_dir_exists(destination_dir)
         self.source_dir = source_dir
         self.destination_dir = destination_dir
         self.exclude_file_patterns = exclude_file_patterns
-        self.log_file = log_file
+        if log_file is None:
+            self.log_file = str(Path.home() / ".local/share/backup/rsync-backups.log")
+        else:
+            self.log_file = log_file
         self.link_dir = str(Path(self.destination_dir) / "latest")
         self._optionless_rsync_arguments = [
             "--delete",  # delete extraneous files from dest dirs
@@ -46,30 +49,30 @@ def is_empty_directory(self, path: str) -> bool:
         return not any(Path(path).iterdir())
 
     def ensure_dir_exists(self, path: str):
-        path = Path(path)
-        if not path.exists():
-            msg = f"{path} does not exist"
-            raise InvalidPath(msg)
-        if not path.is_dir():
-            msg = f"{path} is not a directory"
-            raise InvalidPath(msg)
+        _path = Path(path)
+        if not _path.exists():
+            msg = f"{_path} does not exist"
+            raise InvalidPathError(msg)
+        if not _path.is_dir():
+            msg = f"{_path} is not a directory"
+            raise InvalidPathError(msg)
 
     def generate_new_backup_dir_path(self) -> str:
         time_stamp = get_time_stamp()
         new_backup_dir = Path(self.destination_dir) / time_stamp
         if new_backup_dir.exists():
             msg = f"{new_backup_dir} already exists and will get overwritten"
-            raise InvalidPath(msg)
+            raise InvalidPathError(msg)
         else:
             return str(new_backup_dir)
 
-    def get_rsync_command(self, new_backup_dir: str, previous_backup_exists: bool = False) -> List[str]:
+    def get_rsync_command(self, new_backup_dir: str, backup_method: BackupType) -> List[str]:
         destination = new_backup_dir
         source = self.source_dir
         link_dest = self.link_dir
         option_arguments = []
 
-        if previous_backup_exists:
+        if backup_method == BackupType.Incremental:
             option_arguments.append(f"--link-dest={link_dest}")
 
         if self.exclude_file_patterns is not None:

src/pisync/config/remote_config.py

@@ -3,18 +3,18 @@
 
 from fabric import Connection
 
-from pisync.config.base_config import InvalidPath, _BaseConfig
+from pisync.config.base_config import BackupType, BaseConfig, InvalidPathError
 from pisync.util import get_time_stamp
 
 
-class RemoteConfig(_BaseConfig):
+class RemoteConfig(BaseConfig):
     def __init__(
         self,
         user_at_hostname: str,
         source_dir: str,
         destination_dir: str,
         exclude_file_patterns: Optional[List[str]] = None,
-        log_file: str = Path.home() / ".local/share/backup/rsync-backups.log",
+        log_file: Optional[str] = None,
     ):
         self.user_at_hostname = user_at_hostname
         self.connection: Connection = Connection(user_at_hostname)
@@ -23,7 +23,10 @@ def __init__(
         self.source_dir = source_dir
         self.destination_dir = destination_dir
         self.exclude_file_patterns = exclude_file_patterns
-        self.log_file = log_file
+        if log_file is None:
+            self.log_file = str(Path.home() / ".local/share/backup/rsync-backups.log")
+        else:
+            self.log_file = log_file
         self.link_dir = f"{self.destination_dir}/latest"
         self._optionless_rsync_arguments = [
             "--delete",  # delete extraneous files from dest dirs
@@ -36,13 +39,13 @@ def __init__(
         ]
 
     def _ensure_dir_exists_locally(self, path: str):
-        path = Path(path)
-        if not path.exists():
-            msg = f"{path} does not exist"
-            raise InvalidPath(msg)
-        if not path.is_dir():
-            msg = f"{path} is not a directory"
-            raise InvalidPath(msg)
+        _path = Path(path)
+        if not _path.exists():
+            msg = f"{_path} does not exist"
+            raise InvalidPathError(msg)
+        if not _path.is_dir():
+            msg = f"{_path} is not a directory"
+            raise InvalidPathError(msg)
 
     def is_symlink(self, path: str) -> bool:
         """returns true if path is a symbolic link"""
@@ -77,13 +80,13 @@ def resolve(self, path: str) -> str:
         result = self.connection.run(f"realpath {path}", warn=True)
         return result.stdout.strip()
 
-    def ensure_dir_exists(self, path: str) -> str:
+    def ensure_dir_exists(self, path: str) -> None:
         result = self.connection.run(f"test -d {path}", warn=True)
         if not result.ok:
             msg = f"{path} is not a directory"
-            raise InvalidPath(msg)
+            raise InvalidPathError(msg)
 
-    def _is_directory(self, path) -> bool:
+    def _is_directory(self, path) -> BackupType:
         result = self.connection.run(f"test -d {path}", warn=True)
         return result.ok
 
@@ -92,24 +95,24 @@ def generate_new_backup_dir_path(self) -> str:
         :returns: The Path string of the directory where the new backup will be
         written.
         :raises:
-            InvalidPath: If the destination directory already exists
+            InvalidPathError: If the destination directory already exists
         """
         time_stamp = get_time_stamp()
         new_backup_dir = f"{self.destination_dir}/{time_stamp}"
         exists = self._is_directory(new_backup_dir) or self.file_exists(new_backup_dir)
         if exists:
             msg = f"{new_backup_dir} already exists and will get overwritten"
-            raise InvalidPath(msg)
+            raise InvalidPathError(msg)
         else:
             return str(new_backup_dir)
 
-    def get_rsync_command(self, new_backup_dir: str, previous_backup_exists: bool = False) -> List[str]:
+    def get_rsync_command(self, new_backup_dir: str, backup_method: BackupType) -> List[str]:
         destination = f"{self.user_at_hostname}:{new_backup_dir}"
         source = self.source_dir
         link_dest = self.link_dir
         option_arguments = []
 
-        if previous_backup_exists:
+        if backup_method == BackupType.Incremental:
             option_arguments.append(f"--link-dest={link_dest}")
 
         if self.exclude_file_patterns is not None:

src/pisync/util.py

@@ -2,6 +2,6 @@
 
 
 def get_time_stamp() -> str:
-    now = datetime.now()
+    now = datetime.now().astimezone()
     stamp = now.strftime("%Y-%m-%d-%H-%M-%S")
     return str(stamp)