diff --git a/end-to-end-tests/tests.py b/end-to-end-tests/tests.py index 7f50b07..67adc83 100644 --- a/end-to-end-tests/tests.py +++ b/end-to-end-tests/tests.py @@ -11,7 +11,7 @@ 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", @@ -19,7 +19,7 @@ def test_send_file(self): 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", diff --git a/sysrsync/command_maker.py b/sysrsync/command_maker.py index 353672a..7bd6f0f 100644 --- a/sysrsync/command_maker.py +++ b/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() diff --git a/sysrsync/exceptions.py b/sysrsync/exceptions.py index cfb7488..c4e0c71 100644 --- a/sysrsync/exceptions.py +++ b/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) diff --git a/sysrsync/helpers/directories.py b/sysrsync/helpers/directories.py index 55c363e..046128c 100644 --- a/sysrsync/helpers/directories.py +++ b/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: diff --git a/sysrsync/helpers/rsync.py b/sysrsync/helpers/rsync.py index a194737..511e02a 100644 --- a/sysrsync/helpers/rsync.py +++ b/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: diff --git a/sysrsync/runner.py b/sysrsync/runner.py index c0b40f4..3125d8b 100644 --- a/sysrsync/runner.py +++ b/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}") diff --git a/test/test_directories_helper.py b/test/test_directories_helper.py index dc64c08..35cb2b0 100644 --- a/test/test_directories_helper.py +++ b/test/test_directories_helper.py @@ -4,7 +4,7 @@ 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' @@ -12,14 +12,14 @@ def test_strip_trailing_slash(self): 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/' @@ -27,14 +27,14 @@ def test_add_trailing_slash(self): 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 diff --git a/test/test_iterators_helper.py b/test/test_iterators_helper.py index e50a20a..0790e2a 100644 --- a/test/test_iterators_helper.py +++ b/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) diff --git a/test/test_sysrsync.py b/test/test_sysrsync.py index 0f6ce8c..1a706bb 100644 --- a/test/test_sysrsync.py +++ b/test/test_sysrsync.py @@ -7,7 +7,7 @@ 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'] @@ -15,7 +15,7 @@ def test_get_exclusions(self): 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'] @@ -23,7 +23,7 @@ def test_get_exclusions_already_in_rsync_format(self): 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() @@ -31,7 +31,7 @@ def test_simple_rsync_command(self): self.assertEqual(expect, result) - def test_rsync_options(self): + def test_rsync_options(self) -> None: source = '/a' target = '/b' options = ['-a', '--verbose'] @@ -40,7 +40,7 @@ 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() @@ -48,7 +48,7 @@ def test_simple_rsync_command_content_false(self): 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'