diff --git a/greenbone/feed/sync/config.py b/greenbone/feed/sync/config.py index c153ad6..f8ebb97 100644 --- a/greenbone/feed/sync/config.py +++ b/greenbone/feed/sync/config.py @@ -31,7 +31,7 @@ try: import tomllib except ImportError: - import tomli as tomllib + import tomli as tomllib # type: ignore[no-redef] def maybe_int(value: Optional[str]) -> Union[int, str, None]: @@ -299,11 +299,9 @@ def feed_url(self) -> str: class ConfigDict(Protocol): - def items(self) -> Iterable[tuple[str, Any]]: - ... + def items(self) -> Iterable[tuple[str, Any]]: ... - def __getitem__(self, key: str) -> Any: - ... + def __getitem__(self, key: str) -> Any: ... class Config: diff --git a/greenbone/feed/sync/helper.py b/greenbone/feed/sync/helper.py index cef75f9..168c4dd 100644 --- a/greenbone/feed/sync/helper.py +++ b/greenbone/feed/sync/helper.py @@ -46,7 +46,7 @@ async def flock_wait( path: Union[str, Path], *, console: Optional[Console] = None, - wait_interval: Optional[int] = DEFAULT_FLOCK_WAIT_INTERVAL, + wait_interval: Optional[Union[int, float]] = DEFAULT_FLOCK_WAIT_INTERVAL, ) -> AsyncGenerator[None, None]: """ Try to lock a file and wait if it is already locked diff --git a/greenbone/feed/sync/parser.py b/greenbone/feed/sync/parser.py index 49840db..4b8a0c5 100644 --- a/greenbone/feed/sync/parser.py +++ b/greenbone/feed/sync/parser.py @@ -347,9 +347,9 @@ def parse_arguments( # set greenbone enterprise feed key in config if user passed one to load # desired key for determining the feed url if known_args.greenbone_enterprise_feed_key: - config[ - "greenbone-enterprise-feed-key" - ] = known_args.greenbone_enterprise_feed_key + config["greenbone-enterprise-feed-key"] = ( + known_args.greenbone_enterprise_feed_key + ) if self.parser.prog == "greenbone-nvt-sync": config["type"] = "nvt" diff --git a/greenbone/feed/sync/rsync.py b/greenbone/feed/sync/rsync.py index 638a65b..de4b774 100644 --- a/greenbone/feed/sync/rsync.py +++ b/greenbone/feed/sync/rsync.py @@ -42,14 +42,16 @@ async def exec_rsync(*args: str) -> None: DEFAULT_RSYNC_URL = "rsync://feed.community.greenbone.net/community" DEFAULT_RSYNC_COMPRESSION_LEVEL = 9 -DEFAULT_RSYNC_TIMEOUT: Optional[ - int -] = None # in seconds. 0 means no timeout and None use rsync default +DEFAULT_RSYNC_TIMEOUT: Optional[int] = ( + None # in seconds. 0 means no timeout and None use rsync default +) DEFAULT_RSYNC_SSH_PORT = 24 DEFAULT_RSYNC_SSH_OPTS = ( "-o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no" ) +PathLike = Union[os.PathLike, str] + class Rsync: """ @@ -71,11 +73,11 @@ def __init__( self, *, verbose: bool = False, - private_subdir: Optional[Path] = None, + private_subdir: Optional[PathLike] = None, compression_level: Optional[int] = DEFAULT_RSYNC_COMPRESSION_LEVEL, timeout: Optional[int] = DEFAULT_RSYNC_TIMEOUT, - ssh_key: Optional[Path] = None, - exclude: Optional[Iterable[os.PathLike]] = None, + ssh_key: Optional[PathLike] = None, + exclude: Optional[Iterable[PathLike]] = None, ) -> None: self.verbose = verbose self.private_subdir = private_subdir @@ -84,7 +86,7 @@ def __init__( self.ssh_key = ssh_key self.exclude = exclude - async def sync(self, url: str, destination: Union[str, Path]) -> None: + async def sync(self, url: str, destination: PathLike) -> None: """ Sync data from a remote URL to a destination path @@ -147,7 +149,7 @@ async def sync(self, url: str, destination: Union[str, Path]) -> None: ] if self.private_subdir: - rsync_delete.extend(["--exclude", str(self.private_subdir)]) + rsync_delete.extend(["--exclude", os.fspath(self.private_subdir)]) if self.exclude: for exclude in self.exclude: diff --git a/tests/test_helper.py b/tests/test_helper.py index f2d180b..a0e355e 100644 --- a/tests/test_helper.py +++ b/tests/test_helper.py @@ -86,7 +86,9 @@ async def test_retry(self, flock_mock: MagicMock): console = MagicMock(spec=Console) async with flock_wait( - lock_file, console=console, wait_interval=0.5 + lock_file, + console=console, + wait_interval=0.5, ): pass diff --git a/tests/test_main.py b/tests/test_main.py index 3c50de0..a97a9cc 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -83,17 +83,21 @@ async def test_do_not_run_as_root( console = MagicMock() rsync_mock_instance = rsync_mock.return_value - with temp_directory() as temp_dir, patch.dict( - "os.environ", - {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, - ), patch.object( - sys, - "argv", - [ - "greenbone-feed-sync", - "--type", - "nvt", - ], + with ( + temp_directory() as temp_dir, + patch.dict( + "os.environ", + {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, + ), + patch.object( + sys, + "argv", + [ + "greenbone-feed-sync", + "--type", + "nvt", + ], + ), ): ret = await feed_sync(console=console, error_console=console) self.assertEqual(ret, 0) @@ -137,17 +141,21 @@ async def test_sync_nvts(self, rsync_mock: MagicMock): console = MagicMock() rsync_mock_instance = rsync_mock.return_value - with temp_directory() as temp_dir, patch.dict( - "os.environ", - {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, - ), patch.object( - sys, - "argv", - [ - "greenbone-feed-sync", - "--type", - "nvt", - ], + with ( + temp_directory() as temp_dir, + patch.dict( + "os.environ", + {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, + ), + patch.object( + sys, + "argv", + [ + "greenbone-feed-sync", + "--type", + "nvt", + ], + ), ): ret = await feed_sync(console=console, error_console=console) self.assertEqual(ret, 0) @@ -194,13 +202,17 @@ async def test_sync_nvts_verbose(self, rsync_mock: MagicMock): console = MagicMock() rsync_mock_instance = rsync_mock.return_value - with temp_directory() as temp_dir, patch.dict( - "os.environ", - {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, - ), patch.object( - sys, - "argv", - ["greenbone-feed-sync", "--type", "nvt", "-vvv"], + with ( + temp_directory() as temp_dir, + patch.dict( + "os.environ", + {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, + ), + patch.object( + sys, + "argv", + ["greenbone-feed-sync", "--type", "nvt", "-vvv"], + ), ): ret = await feed_sync(console=console, error_console=console) self.assertEqual(ret, 0) @@ -261,13 +273,17 @@ async def test_sync_nvts_quiet(self, rsync_mock: MagicMock): console = MagicMock() rsync_mock_instance = rsync_mock.return_value - with temp_directory() as temp_dir, patch.dict( - "os.environ", - {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, - ), patch.object( - sys, - "argv", - ["greenbone-feed-sync", "--type", "nvt", "--quiet"], + with ( + temp_directory() as temp_dir, + patch.dict( + "os.environ", + {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, + ), + patch.object( + sys, + "argv", + ["greenbone-feed-sync", "--type", "nvt", "--quiet"], + ), ): ret = await feed_sync(console=console, error_console=console) self.assertEqual(ret, 0) @@ -303,13 +319,17 @@ async def test_sync_nvts_rsync_error(self, rsync_mock: MagicMock): 2, [], b"An rsync error" ) - with temp_directory() as temp_dir, patch.dict( - "os.environ", - {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, - ), patch.object( - sys, - "argv", - ["greenbone-feed-sync", "--type", "nvt", "--fail-fast"], + with ( + temp_directory() as temp_dir, + patch.dict( + "os.environ", + {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, + ), + patch.object( + sys, + "argv", + ["greenbone-feed-sync", "--type", "nvt", "--fail-fast"], + ), ): ret = await feed_sync(console=console, error_console=console) self.assertEqual(ret, 1) @@ -354,17 +374,21 @@ def test_sync_nvts(self, rsync_mock: MagicMock, console_mock: MagicMock): rsync_mock_instance = rsync_mock.return_value console_mock_instance = console_mock.return_value - with temp_directory() as temp_dir, patch.dict( - "os.environ", - {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, - ), patch.object( - sys, - "argv", - [ - "greenbone-feed-sync", - "--type", - "nvt", - ], + with ( + temp_directory() as temp_dir, + patch.dict( + "os.environ", + {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, + ), + patch.object( + sys, + "argv", + [ + "greenbone-feed-sync", + "--type", + "nvt", + ], + ), ): with self.assertRaises(SystemExit) as cm: main() @@ -419,13 +443,17 @@ def test_sync_nvts_error( "An error" ) - with temp_directory() as temp_dir, patch.dict( - "os.environ", - {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, - ), patch.object( - sys, - "argv", - ["greenbone-feed-sync", "--type", "nvt", "--fail-fast"], + with ( + temp_directory() as temp_dir, + patch.dict( + "os.environ", + {"GREENBONE_FEED_SYNC_DESTINATION_PREFIX": str(temp_dir)}, + ), + patch.object( + sys, + "argv", + ["greenbone-feed-sync", "--type", "nvt", "--fail-fast"], + ), ): with self.assertRaises(SystemExit) as cm: main() diff --git a/tests/test_parser.py b/tests/test_parser.py index 23839fe..27a7046 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -223,9 +223,10 @@ def test_defaults(self): def test_help(self): parser = CliParser() - with redirect_stdout(io.StringIO()) as f, self.assertRaises( - SystemExit - ) as cm: + with ( + redirect_stdout(io.StringIO()) as f, + self.assertRaises(SystemExit) as cm, + ): parser.parse_arguments(["--help"]) self.assertEqual(cm.exception.code, 0) @@ -260,7 +261,7 @@ def test_selftest(self): self.assertTrue(args.selftest) @patch("greenbone.feed.sync.parser.Path") - def test_use_default_config_files(self, path_mock: Path): + def test_use_default_config_files(self, path_mock): path_mock_instance = path_mock.return_value path_mock_instance.expanduser.return_value = path_mock_instance path_mock_instance.resolve.return_value = path_mock_instance @@ -461,9 +462,10 @@ def test_greenbone_enterprise_feed_key(self): def test_other(self): parser = CliParser() - with redirect_stderr(io.StringIO()) as f, self.assertRaises( - SystemExit - ) as cm: + with ( + redirect_stderr(io.StringIO()) as f, + self.assertRaises(SystemExit) as cm, + ): parser.parse_arguments(["--foo-bar", "10"]) self.assertIn("error: unrecognized arguments: --foo-bar", f.getvalue())