Add type annotations

end-to-end-tests/tests.py

@@ -11,15 +11,15 @@
 
 
 class TestE2E(unittest.TestCase):
-    def test_send_file(self):
+    def test_send_file(self) -> None:
         sysrsync.run(source="end-to-end-tests/test-cases/test_file",
                      destination="/tmp/target_test_file",
                      destination_ssh="test@openssh-server",
                      private_key="end-to-end-tests/keys/test-key",
                      rsh_port=2222,
                      strict_host_key_checking=False)
 
-    def test_send_file_with_spaces(self):
+    def test_send_file_with_spaces(self) -> None:
         sysrsync.run(source="end-to-end-tests/test-cases/file with spaces",
                      destination="/tmp/target_test_file",
                      destination_ssh="test@openssh-server",

sysrsync/command_maker.py

@@ -7,16 +7,18 @@
 from sysrsync.helpers.rsync import get_exclusions, get_rsh_command
 
 
-def get_rsync_command(source: str,
-                      destination: str,
-                      source_ssh: Optional[str] = None,
-                      destination_ssh: Optional[str] = None,
-                      exclusions: Optional[Iterable[str]] = None,
-                      sync_source_contents: bool = True,
-                      options: Optional[Iterable[str]] = None,
-                      private_key: Optional[str] = None,
-                      rsh_port: Optional[int] = None,
-                      strict_host_key_checking: Optional[bool] = None) -> List[str]:
+def get_rsync_command(
+    source: str,
+    destination: str,
+    source_ssh: Optional[str] = None,
+    destination_ssh: Optional[str] = None,
+    exclusions: Optional[Iterable[str]] = None,
+    sync_source_contents: Optional[bool] = True,
+    options: Optional[Iterable[str]] = None,
+    private_key: Optional[str] = None,
+    rsh_port: Optional[int] = None,
+    strict_host_key_checking: Optional[bool] = None,
+) -> List[str]:
     if source_ssh is not None and destination_ssh is not None:
         raise RemotesError()
 

sysrsync/exceptions.py

@@ -1,5 +1,5 @@
 class RemotesError(Exception):
-    def __init__(self):
+    def __init__(self) -> None:
         message = 'source and destination cannot both be remote'
         super().__init__(message)
 
@@ -9,6 +9,6 @@ class RsyncError(Exception):
 
 
 class PrivateKeyError(Exception):
-    def __init__(self, key_file):
+    def __init__(self, key_file: str) -> None:
         message = f'Private Key File "{key_file}" does not exist'
         super().__init__(message)

sysrsync/helpers/directories.py

@@ -8,8 +8,11 @@ def get_directory_with_ssh(directory: str, ssh: Optional[str]) -> str:
     return f'{ssh}:{directory}'
 
 
-def sanitize_trailing_slash(source_dir, target_dir, sync_sourcedir_contents=True):
-    # type: (str, str, bool) -> Tuple[str, str]
+def sanitize_trailing_slash(
+    source_dir: str,
+    target_dir: str,
+    sync_sourcedir_contents: Optional[bool] = True,
+) -> Tuple[str, str]:
     target_dir = strip_trailing_slash(target_dir)
 
     if sync_sourcedir_contents is True:

sysrsync/helpers/rsync.py

@@ -11,9 +11,11 @@ def get_exclusions(exclusions: Iterable[str]) -> Iterable[str]:
                     for exclusion in exclusions
                     if exclusion != '--exclude'))
 
-
-def get_rsh_command(private_key: Optional[str] = None, port: Optional[int] = None, strict_host_key_checking: Optional[bool] = None):
-
+def get_rsh_command(
+    private_key: Optional[str] = None,
+    port: Optional[int] = None,
+    strict_host_key_checking: Optional[bool] = None,
+) -> List[str]:
     args: List[str] = []
 
     if private_key is not None:

sysrsync/runner.py

@@ -4,8 +4,15 @@
 from sysrsync.command_maker import get_rsync_command
 from sysrsync.exceptions import RsyncError
 
+from typing import Optional
 
-def run(cwd=os.getcwd(), strict=True, verbose=False, **kwargs):
+
+def run(
+    cwd: Optional[str] = os.getcwd(),
+    strict: Optional[bool] = True,
+    verbose: Optional[bool] = False,
+    **kwargs,
+) -> subprocess.CompletedProcess:
     rsync_command = get_rsync_command(**kwargs)
 
     rsync_string = ' '.join(rsync_command)
@@ -22,6 +29,9 @@ def run(cwd=os.getcwd(), strict=True, verbose=False, **kwargs):
     return process
 
 
-def _check_return_code(return_code: int, action: str):
+def _check_return_code(
+    return_code: int,
+    action: str,
+) -> None:
     if return_code != 0:
         raise RsyncError(f"[sysrsync runner] {action} exited with code {return_code}")

test/test_directories_helper.py

@@ -4,37 +4,37 @@
 
 
 class TestDirectoriesHelper(unittest.TestCase):
-    def test_strip_trailing_slash(self):
+    def test_strip_trailing_slash(self) -> None:
         """test strip trailing slash"""
         test_dir = '/a/'
         expect = '/a'
         result = directories.strip_trailing_slash(test_dir)
 
         self.assertEqual(expect, result)
 
-    def test_skip_strip_trailing_slash(self):
+    def test_skip_strip_trailing_slash(self) -> None:
         """test skip strip trailing slash when not necessary"""
         test_dir = '/a'
         result = directories.strip_trailing_slash(test_dir)
 
         self.assertEqual(result, test_dir)
 
-    def test_add_trailing_slash(self):
+    def test_add_trailing_slash(self) -> None:
         """test add trailing slash"""
         test_dir = '/a'
         expect = '/a/'
         result = directories.add_trailing_slash(test_dir)
 
         self.assertEqual(expect, result)
 
-    def test_skip_add_trailing_slash(self):
+    def test_skip_add_trailing_slash(self) -> None:
         """test skip add trailing slash when not necessary"""
         test_dir = '/a/'
         result = directories.add_trailing_slash(test_dir)
 
         self.assertEqual(result, test_dir)
 
-    def test_sanitize_trailing_slash(self):
+    def test_sanitize_trailing_slash(self) -> None:
         """test sanitize trailing slash when syncing source contents"""
         source, target = '/a', '/b/'
         expect_source, expect_target = '/a/', '/b'
@@ -44,7 +44,7 @@ def test_sanitize_trailing_slash(self):
         self.assertEqual(expect_source, result_source)
         self.assertEqual(expect_target, result_target)
 
-    def test_sanitize_trailing_slash_no_action_needed(self):
+    def test_sanitize_trailing_slash_no_action_needed(self) -> None:
         """test sanitize trailing slash when syncing source contents when already sanitized"""
         source, target = '/a/', '/b'
         expect_source, expect_target = '/a/', '/b'
@@ -54,7 +54,7 @@ def test_sanitize_trailing_slash_no_action_needed(self):
         self.assertEqual(expect_source, result_source)
         self.assertEqual(expect_target, result_target)
 
-    def test_sanitize_trailing_slash_whole_source(self):
+    def test_sanitize_trailing_slash_whole_source(self) -> None:
         """test sanitize trailing slash when syncing whole source"""
         source, target = '/a/', '/b/'
         expect_source, expect_target = '/a', '/b'
@@ -64,7 +64,7 @@ def test_sanitize_trailing_slash_whole_source(self):
         self.assertEqual(expect_source, result_source)
         self.assertEqual(expect_target, result_target)
 
-    def test_sanitize_trailing_slash_whole_source_no_action_needed(self):
+    def test_sanitize_trailing_slash_whole_source_no_action_needed(self) -> None:
         """test sanitize trailing slash when syncing whole source when already sanitized"""
         source, target = '/a', '/b/'
         expect_source, expect_target = '/a', '/b'
@@ -74,7 +74,7 @@ def test_sanitize_trailing_slash_whole_source_no_action_needed(self):
         self.assertEqual(expect_source, result_source)
         self.assertEqual(expect_target, result_target)
 
-    def test_dir_with_ssh(self):
+    def test_dir_with_ssh(self) -> None:
         """should compose string with ssh for rsync connection"""
         directory = '/a'
         ssh = 'host'
@@ -83,7 +83,7 @@ def test_dir_with_ssh(self):
 
         self.assertEqual(result, expect)
 
-    def test_dir_without_ssh(self):
+    def test_dir_without_ssh(self) -> None:
         """should return directory when ssh is None"""
         directory = '/a'
         ssh = None

test/test_iterators_helper.py

@@ -4,21 +4,21 @@
 
 
 class TestIteratorsHelper(unittest.TestCase):
-    def test_list_flatten(self):
+    def test_list_flatten(self) -> None:
         list_input = [1, [2, 3], [4]]
         expect = [1, 2, 3, 4]
         result = iterators.flatten(list_input)
 
         self.assertEqual(expect, result)
 
-    def test_tuple_flatten(self):
+    def test_tuple_flatten(self) -> None:
         tuple_input = (1, [2, 3], [4])
         expect = [1, 2, 3, 4]
         result = iterators.flatten(tuple_input)
 
         self.assertEqual(expect, result)
 
-    def test_tuples_and_lists_list_flatten(self):
+    def test_tuples_and_lists_list_flatten(self) -> None:
         tuple_input = (1, (2, 3), [4])
         expect = [1, 2, 3, 4]
         result = iterators.flatten(tuple_input)

test/test_sysrsync.py

@@ -7,31 +7,31 @@
 
 
 class TestPackage(unittest.TestCase):
-    def test_get_exclusions(self):
+    def test_get_exclusions(self) -> None:
         """should map list of exclusions to a list with each element following a --exclude statement"""
         exclusions = ['a', 'b']
         expect = ['--exclude', 'a', '--exclude', 'b']
         result = get_exclusions(exclusions)
 
         self.assertEqual(expect, result)
 
-    def test_get_exclusions_already_in_rsync_format(self):
+    def test_get_exclusions_already_in_rsync_format(self) -> None:
         """should ignore --exclude in exclusions"""
         exclusions = ['--exclude', 'a', '--exclude', 'b']
         expect = ['--exclude', 'a', '--exclude', 'b']
         result = get_exclusions(exclusions)
 
         self.assertEqual(expect, result)
 
-    def test_simple_rsync_command(self):
+    def test_simple_rsync_command(self) -> None:
         source = '/a'
         target = '/b'
         expect = 'rsync /a/ /b'.split()
         result = get_rsync_command(source, target)
 
         self.assertEqual(expect, result)
 
-    def test_rsync_options(self):
+    def test_rsync_options(self) -> None:
         source = '/a'
         target = '/b'
         options = ['-a', '--verbose']
@@ -40,15 +40,15 @@ def test_rsync_options(self):
 
         self.assertEqual(expect, result)
 
-    def test_simple_rsync_command_content_false(self):
+    def test_simple_rsync_command_content_false(self) -> None:
         source = '/a'
         target = '/b'
         expect = 'rsync /a /b'.split()
         result = get_rsync_command(source, target, sync_source_contents=False)
 
         self.assertEqual(expect, result)
 
-    def test_rsync_exclusions(self):
+    def test_rsync_exclusions(self) -> None:
         source = '/a'
         target = '/b'
         exclusions = ['file1', 'file2']
@@ -57,7 +57,7 @@ def test_rsync_exclusions(self):
 
         self.assertEqual(expect, result)
 
-    def test_rsync_exclusions_source_ssh(self):
+    def test_rsync_exclusions_source_ssh(self) -> None:
         source = '/a'
         source_ssh = 'host1'
         target = '/b'
@@ -67,7 +67,7 @@ def test_rsync_exclusions_source_ssh(self):
 
         self.assertEqual(expect, result)
 
-    def test_rsync_exclusions_target_ssh(self):
+    def test_rsync_exclusions_target_ssh(self) -> None:
         source = '/a'
         target_ssh = 'host1'
         target = '/b'
@@ -77,7 +77,7 @@ def test_rsync_exclusions_target_ssh(self):
 
         self.assertEqual(expect, result)
 
-    def test_rsync_throws_both_remotes(self):
+    def test_rsync_throws_both_remotes(self) -> None:
         """raises RemotesError when both source and destination are remotes"""
         source_ssh = 'host1'
         source = '/a'
@@ -86,7 +86,7 @@ def test_rsync_throws_both_remotes(self):
         with self.assertRaises(RemotesError):
             get_rsync_command(source, target, source_ssh=source_ssh, destination_ssh=target_ssh)
 
-    def test_rsync_private_key(self):
+    def test_rsync_private_key(self) -> None:
         """test if correctly creates rsh option when passing a private key"""
         with NamedTemporaryFile() as temp_file:
             source_dir = '/home/user/files/'
@@ -108,7 +108,7 @@ def test_rsync_private_key(self):
                                        strict_host_key_checking=strict_host_key_checking)
             self.assertEqual(expect, actual)
 
-    def test_rsync_private_key_missing(self):
+    def test_rsync_private_key_missing(self) -> None:
         """test if get_rsync_command raises PrivateKeyError when key missing"""
         source_dir = '/home/user/files/'
         target_dir = '/home/server/files'