synchroniser : misc pylint suggested fixes

- Do not use mutables as default arguments
- Remove trailing whitespaces
- Initialise using [] and {} instead of list() and dict()
- Use is not None instead of != None
- Some other small stuff

src/unisync/defaults.py

@@ -1,4 +1,4 @@
-# copyright (c) 2026 paul retourné
+# Copyright (c) 2026 paul retourné
 # spdx-license-identifier: gpl-3.0-or-later
 
 from pathlib import Path

src/unisync/errors.py

@@ -1,8 +1,21 @@
 # Copyright (C) 2025-2026 Paul Retourné
 # SPDX-License-Identifier: GPL-3.0-or-later
 
-class RemoteMountedError(BaseException):
+from typing import NoReturn
+import sys
+
+class RemoteMountedError(Exception):
     pass
 
-class InvalidMountError(BaseException):
+class InvalidMountError(Exception):
     pass
+
+class UnknownSSHError(Exception):
+    pass
+
+class FatalSyncError(Exception):
+    pass
+
+def unisync_exit_fatal(reason:str) -> NoReturn:
+    print(reason)
+    sys.exit(1)

src/unisync/main.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 
 from unisync.argparser import create_argparser
+from unisync.errors import UnknownSSHError, unisync_exit_fatal
 from unisync.runners import unisync_sync, unisync_add, unisync_mount
 from unisync.config import load_config
 from unisync.synchroniser import Synchroniser
@@ -38,7 +39,10 @@ def main():
 
     paths_manager = PathsManager(Path(config.roots.local), config.other.cache_dir_path)
 
-    cli_args.func(synchroniser, paths_manager, config)
+    try:
+        cli_args.func(synchroniser, paths_manager, config)
+    except UnknownSSHError:
+        unisync_exit_fatal("Connection failed quitting")
 
 
 if __name__ == "__main__":

src/unisync/runners.py

@@ -5,11 +5,11 @@ from unisync.synchroniser import Synchroniser
 from unisync.paths import PathsManager
 from unisync.config import Config
 
+
 def unisync_sync(synchroniser:Synchroniser, paths_manager:PathsManager, config: Config):
     del config # The function signature must be the same for all runners
-    if synchroniser.create_ssh_master_connection() != 0:
+
-        print("Connection failed quitting")
+    synchroniser.create_ssh_master_connection()
-        return 1
     print("Connected to the remote.")
 
     synchroniser.sync_files(paths_manager.get_paths_to_sync())
@@ -24,9 +24,8 @@ def unisync_sync(synchroniser:Synchroniser, paths_manager:PathsManager, config:
 
 def unisync_add(synchroniser:Synchroniser, paths_manager:PathsManager, config: Config):
     del config # The function signature must be the same for all runners
-    if synchroniser.create_ssh_master_connection() != 0:
+
-        print("Connection failed quitting")
+    synchroniser.create_ssh_master_connection()
-        return 1
     print("Connected to the remote.")
 
     # TODO config or cli to skip this first sync

src/unisync/synchroniser.py

@@ -10,13 +10,12 @@ the remote.
 import subprocess
 import os
 import sys
-import time
 import logging
 
 from pathlib import Path
 from typing import cast
 
-from unisync.errors import RemoteMountedError, InvalidMountError
+from unisync.errors import RemoteMountedError, InvalidMountError, UnknownSSHError, FatalSyncError
 from unisync.config import BackupConfig
 
 logger = logging.getLogger(__name__)
@@ -50,26 +49,27 @@ class Synchroniser:
     """
 
     def __init__(self, remote:str, local:str, user:str, ip:str, port:int=22,
-                 args_bool:list=[], args_value:dict={}, ssh_settings:dict={},
+                 args_bool:list[str] | None = None, args_value:dict[str, str] | None = None,
-                 backup:BackupConfig | None = None
+                 ssh_settings:dict[str, str] | None = None, backup:BackupConfig | None = None
                  ):
         """Initialises an instance of Synchroniser.
         """
         self.remote_dir:str = remote
         self.local:str = local
-        self.args_bool:list[str] = args_bool 
+        self.args_bool:list[str] = args_bool if args_bool is not None else []
-        self.args_value:dict[str, str] = args_value 
+        self.args_value:dict[str, str] = args_value if args_value is not None else {}
-        self.ssh_settings:dict[str, str] = dict()
+        self.ssh_settings:dict[str, str] = ssh_settings if ssh_settings is not None else {}
         self.remote_user:str = user
         self.remote_ip:str = ip
         self.remote_port:int = port
-        self.files_extra:list = list()
+        self.files_extra:list = []
-        self.links_extra:list = list()
+        self.links_extra:list = []
+        self.control_path:str = ""
 
-        if(backup != None and backup.enabled):
+        if backup is not None and backup.enabled:
             backup = cast(BackupConfig, backup)
             self.files_extra.append("-backup")
-            if(backup.selection != ""):
+            if backup.selection != "":
                 self.files_extra.append(backup.selection)
             else:
                 self.files_extra.append("Name *")
@@ -92,7 +92,9 @@ class Synchroniser:
                 f"Name {backup.backupprefix[:-1]}"
                 ])
 
-    def create_ssh_master_connection(self, control_path:str="~/.ssh/control_%C", connection_timeout:int=60) -> int:
+    def create_ssh_master_connection(self, control_path:str="~/.ssh/control_%C",
+                                     connection_timeout:int=60
+                                     ) -> None:
         """Creates an ssh master connection.
 
         It is used so the user only has to authenticate once to the remote server.
@@ -105,8 +107,14 @@ class Synchroniser:
             connection_timeout:
                 Time given to the user to authenticate to the remote server.
                 On slow connections one might want to increase this.
-        Returns:
+
-            An error code (0 success, 1 TimeoutExpired, 2 KeyboardInterrupt).
+        Raises:
+            subprocess.TimeoutExpired:
+                The user didn't finish loging in in time.
+            KeyboardInterrupt:
+                The user interrupted the process.
+            UnknownSSHError:
+                An error occured during the connection.
         """
         self.control_path = os.path.expanduser(control_path)
         command = [
@@ -117,20 +125,19 @@ class Synchroniser:
                 f"{self.remote_user}@{self.remote_ip}",
                 "-p", str(self.remote_port)
                 ]
-        master_ssh = subprocess.Popen(command)
+        with subprocess.Popen(command) as master_ssh:
-        # TODO: Raise an exception instead of changing the return value
+            # TODO: Raise an exception instead of changing the return value
-        try:
+            try:
-            ret_code = master_ssh.wait(timeout=connection_timeout)
+                ret_code = master_ssh.wait(timeout=connection_timeout)
-        except subprocess.TimeoutExpired:
+            except subprocess.TimeoutExpired as e:
-            print("Time to login expired", file=sys.stderr)
+                print("Time to login expired", file=sys.stderr)
-            return 1
+                raise e
-        except KeyboardInterrupt:
+            except KeyboardInterrupt as e:
-            return 2
+                raise e
 
         if ret_code != 0:
             print("Login to remote failed", file=sys.stderr)
-            return ret_code
+            raise UnknownSSHError
-        return 0
 
 
     def close_ssh_master_connection(self) -> int:
@@ -146,21 +153,23 @@ class Synchroniser:
                 f"{self.remote_user}@{self.remote_ip}",
                 "-p", str(self.remote_port)
                 ]
-        close = subprocess.Popen(command)
+        with subprocess.Popen(command) as close:
-        return close.wait()
+            retval = close.wait()
 
-    def sync_files(self, paths:list, force:bool=False) -> int:
+        return retval
+
+    def sync_files(self, paths:list, force:bool=False) -> None:
         """Synchronises the files.
 
         Args:
             paths: List of paths to synchronise.
             force: Force the changes from remote to local.
 
-        Returns:
+        Raises:
-            The return code of sync.
+            FatalSyncError: A fatal error occured during the synchronisation.
         """
 
-        return self.sync(
+        self.sync(
                 f"ssh://{self.remote_user}@{self.remote_ip}/{self.remote_dir}/.data",
                 self.local,
                 paths=paths,
@@ -168,16 +177,16 @@ class Synchroniser:
                 other=self.files_extra
                 )
 
-    def sync_links(self, ignore:list) -> int:
+    def sync_links(self, ignore:list) -> None:
         """Synchronises the links, they must exist already.
 
         Args:
             ignore: List of paths to ignore.
 
-        Returns:
+        Raises:
-            The return code of sync.
+            FatalSyncError: A fatal error occured during the synchronisation.
         """
-        return self.sync(
+        self.sync(
                 f"ssh://{self.remote_user}@{self.remote_ip}/{self.remote_dir}/links",
                 self.local,
                 ignore=ignore,
@@ -185,9 +194,9 @@ class Synchroniser:
                 )
 
     def sync(self, remote_root:str, local_root:str,
-             paths:list=[], ignore:list=[], force:bool=False,
+             paths:list | None=None, ignore:list | None = None, force:bool=False,
-             other:list=[]
+             other:list | None = None
-             ) -> int:
+             ) -> None:
         """Performs the synchronisation by calling unison.
 
         Args:
@@ -206,9 +215,19 @@ class Synchroniser:
                 They will be added to the command as is no - in front.
                 For exemple backups are implemented using this argument.
 
-        Returns:
+        Raises:
-            the unison return code see section 6.11 of the documentation
+            FatalSyncError:
+                If unison returns 3 it means either a fatal error occured or the synchronisation
+                was interrupted.
+                If this happens propagate the error to unisync.
         """
+        if paths is None:
+            paths = []
+        if ignore is None:
+            ignore = []
+        if other is None:
+            other = []
+
         command = [ "/usr/bin/unison", "-root", remote_root, "-root", local_root ]
         for arg in self.args_bool:
             command.append(f"-{arg}")
@@ -239,9 +258,12 @@ class Synchroniser:
         for arg in other:
             command.append(arg)
 
-        proc = subprocess.Popen(command)
+        with subprocess.Popen(command) as proc:
-        ret_code = proc.wait()
+            ret_code = proc.wait()
-        return ret_code
+
+        # See unison manual section 6.11
+        if ret_code == 3:
+            raise FatalSyncError("Synchronisation could not be completed")
 
     def update_links(self, background:bool=True):
         """Updates the links on the remote.
@@ -270,12 +292,12 @@ class Synchroniser:
                 link_background_wrapper
                 ]
 
-        link_update_process = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        print("Updating links")
-
+        subprocess.run(command,
-        if not background:
+                       stdout=subprocess.DEVNULL,
-            print("Starting links update.")
+                       stderr=subprocess.DEVNULL,
-            link_update_process.wait()
+                       check=False
-            print("Done")
+                       )
 
     def mount_remote_dir(self):
         """Mounts the remote directory to make the local links work.
@@ -302,5 +324,5 @@ class Synchroniser:
                 f"{self.remote_user}@{self.remote_ip}:{self.remote_dir}/.data",
                 str(path_to_mount)
                 ]
-        completed_process = subprocess.run(command)
+        completed_process = subprocess.run(command, check=True)
         completed_process.check_returncode()