Merge pull request #60 from greenbone/cpe-match-fixes-and-tests-2

Small fixes and tests for CPE matches

greenbone/scap/cpe_match/producer/nvd_api.py

@@ -51,7 +51,8 @@ def from_args(
         request_filter_opts = {}
 
         since = NvdApiProducer.since_from_args(args, error_console)
-        request_filter_opts["last_modified_start_date"] = since
+        if since is not None:
+            request_filter_opts["last_modified_start_date"] = since
 
         return CpeMatchNvdApiProducer(
             console,

greenbone/scap/data_utils/json.py

@@ -48,8 +48,9 @@ def convert_keys_to_camel(obj: Any) -> Any:
             # Exclude None values
             if v is not None:
                 new_key = _snake_to_camel(old_key)
-                obj[new_key] = v
-            del obj[old_key]
+                if new_key != old_key:
+                    obj[new_key] = v
+                    del obj[old_key]
     elif isinstance(obj, list):
         for item in obj:
             convert_keys_to_camel(item)

greenbone/scap/generic_cli/processor.py

@@ -63,7 +63,7 @@ def add_args_to_parser(
             "Default: %(default)s.",
             type=int,
             metavar="N",
-            default=cls._arg_defaults["chunk_size"],
+            default=cls._arg_defaults["queue_size"],
         )
         parser.add_argument(
             "-v",

tests/cpe_match/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later

tests/cpe_match/cli/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later

tests/cpe_match/cli/test_json_download.py

@@ -0,0 +1,39 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import unittest
+from datetime import datetime
+from pathlib import Path
+
+from greenbone.scap.cli import DEFAULT_RETRIES, DEFAULT_VERBOSITY
+from greenbone.scap.cpe_match.cli.json_download import parse_args
+from greenbone.scap.cpe_match.cli.processor import CPE_MATCH_DEFAULT_CHUNK_SIZE
+from greenbone.scap.generic_cli.queue import DEFAULT_QUEUE_SIZE
+
+
+class ParseArgsTestCase(unittest.TestCase):
+    def test_defaults(self):
+        args = parse_args([])
+
+        self.assertIsNone(args.since)
+        self.assertIsNone(args.since_from_file)
+        self.assertIsNone(args.number)
+        self.assertIsNone(args.start)
+        self.assertEqual(DEFAULT_RETRIES, args.retry_attempts)
+        self.assertIsNone(args.nvd_api_key)
+
+        self.assertEqual(Path("."), args.storage_path)
+        self.assertFalse(args.compress)
+
+        self.assertEqual(CPE_MATCH_DEFAULT_CHUNK_SIZE, args.chunk_size)
+        self.assertEqual(DEFAULT_QUEUE_SIZE, args.queue_size)
+        self.assertEqual(DEFAULT_VERBOSITY, args.verbose)
+
+    def test_since(self):
+        args = parse_args(["--since", "2024-12-09"])
+        self.assertEqual(datetime(2024, 12, 9), args.since)
+
+    def test_since_from_file(self):
+        args = parse_args(["--since-from-file", "/tmp/path"])
+        self.assertEqual(Path("/tmp/path"), args.since_from_file)

tests/cpe_match/producer/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later

tests/cpe_match/producer/test_nvd_api.py

@@ -0,0 +1,269 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import argparse
+import unittest
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+
+from pontos.testing import temp_file
+from rich.console import Console
+from rich.progress import Progress
+
+from greenbone.scap.cli import DEFAULT_RETRIES, DEFAULT_VERBOSITY
+from greenbone.scap.cpe_match.cli.processor import CpeMatchProcessor
+from greenbone.scap.cpe_match.producer.nvd_api import CpeMatchNvdApiProducer
+
+
+def parse_producer_args(raw_args):
+    parser = argparse.ArgumentParser()
+    CpeMatchProcessor.add_args_to_parser(
+        parser
+    )  # for common args like "verbose"
+    CpeMatchNvdApiProducer.add_args_to_parser(parser)
+    return parser.parse_args(raw_args)
+
+
+class ParseArgsTestCase(unittest.TestCase):
+    @patch(
+        "greenbone.scap.cpe_match.producer.nvd_api.CpeMatchNvdApiProducer",
+        autospec=True,
+    )
+    def test_defaults(self, mock_producer: MagicMock):
+        console = Console(quiet=True)
+        error_console = Console(quiet=True)
+        progress = Progress(disable=True)
+
+        args = parse_producer_args([])
+
+        CpeMatchNvdApiProducer.from_args(args, console, error_console, progress)
+
+        mock_producer.assert_called_once_with(
+            console=console,
+            error_console=error_console,
+            progress=progress,
+            nvd_api_key=None,
+            retry_attempts=DEFAULT_RETRIES,
+            request_results=None,
+            request_filter_opts={},
+            start_index=None,
+            verbose=DEFAULT_VERBOSITY,
+        )
+
+    @patch(
+        "greenbone.scap.cpe_match.producer.nvd_api.CpeMatchNvdApiProducer",
+        autospec=True,
+    )
+    def test_since(self, mock_producer: MagicMock):
+        console = Console(quiet=True)
+        error_console = Console(quiet=True)
+        progress = Progress(disable=True)
+
+        args = parse_producer_args(
+            [
+                "--since",
+                "2024-12-09",
+            ]
+        )
+
+        CpeMatchNvdApiProducer.from_args(args, console, error_console, progress)
+
+        mock_producer.assert_called_once_with(
+            console=console,
+            error_console=error_console,
+            progress=progress,
+            nvd_api_key=None,
+            retry_attempts=DEFAULT_RETRIES,
+            request_results=None,
+            request_filter_opts={
+                "last_modified_start_date": datetime(2024, 12, 9)
+            },
+            start_index=None,
+            verbose=DEFAULT_VERBOSITY,
+        )
+
+    @patch(
+        "greenbone.scap.cpe_match.producer.nvd_api.CpeMatchNvdApiProducer",
+        autospec=True,
+    )
+    def test_since_from_file(self, mock_producer: MagicMock):
+        console = Console(quiet=True)
+        error_console = Console(quiet=True)
+        progress = Progress(disable=True)
+
+        with temp_file("2024-12-05\n", name="since.txt") as temp_file_path:
+            args = parse_producer_args(
+                [
+                    "--since-from-file",
+                    str(temp_file_path),
+                ]
+            )
+
+            CpeMatchNvdApiProducer.from_args(
+                args, console, error_console, progress
+            )
+
+            mock_producer.assert_called_once_with(
+                console=console,
+                error_console=error_console,
+                progress=progress,
+                nvd_api_key=None,
+                retry_attempts=DEFAULT_RETRIES,
+                request_results=None,
+                request_filter_opts={
+                    "last_modified_start_date": datetime(2024, 12, 5)
+                },
+                start_index=None,
+                verbose=DEFAULT_VERBOSITY,
+            )
+
+    @patch(
+        "greenbone.scap.cpe_match.producer.nvd_api.CpeMatchNvdApiProducer",
+        autospec=True,
+    )
+    def test_since_number(self, mock_producer: MagicMock):
+        console = Console(quiet=True)
+        error_console = Console(quiet=True)
+        progress = Progress(disable=True)
+
+        args = parse_producer_args(
+            [
+                "--number",
+                "123",
+            ]
+        )
+
+        CpeMatchNvdApiProducer.from_args(args, console, error_console, progress)
+
+        mock_producer.assert_called_once_with(
+            console=console,
+            error_console=error_console,
+            progress=progress,
+            nvd_api_key=None,
+            retry_attempts=DEFAULT_RETRIES,
+            request_results=123,
+            request_filter_opts={},
+            start_index=None,
+            verbose=DEFAULT_VERBOSITY,
+        )
+
+    @patch(
+        "greenbone.scap.cpe_match.producer.nvd_api.CpeMatchNvdApiProducer",
+        autospec=True,
+    )
+    def test_number(self, mock_producer: MagicMock):
+        console = Console(quiet=True)
+        error_console = Console(quiet=True)
+        progress = Progress(disable=True)
+
+        args = parse_producer_args(
+            [
+                "--number",
+                "123",
+            ]
+        )
+
+        CpeMatchNvdApiProducer.from_args(args, console, error_console, progress)
+
+        mock_producer.assert_called_once_with(
+            console=console,
+            error_console=error_console,
+            progress=progress,
+            nvd_api_key=None,
+            retry_attempts=DEFAULT_RETRIES,
+            request_results=123,
+            request_filter_opts={},
+            start_index=None,
+            verbose=DEFAULT_VERBOSITY,
+        )
+
+    @patch(
+        "greenbone.scap.cpe_match.producer.nvd_api.CpeMatchNvdApiProducer",
+        autospec=True,
+    )
+    def test_start(self, mock_producer: MagicMock):
+        console = Console(quiet=True)
+        error_console = Console(quiet=True)
+        progress = Progress(disable=True)
+
+        args = parse_producer_args(
+            [
+                "--start",
+                "321",
+            ]
+        )
+
+        CpeMatchNvdApiProducer.from_args(args, console, error_console, progress)
+
+        mock_producer.assert_called_once_with(
+            console=console,
+            error_console=error_console,
+            progress=progress,
+            nvd_api_key=None,
+            retry_attempts=DEFAULT_RETRIES,
+            request_results=None,
+            request_filter_opts={},
+            start_index=321,
+            verbose=DEFAULT_VERBOSITY,
+        )
+
+    @patch(
+        "greenbone.scap.cpe_match.producer.nvd_api.CpeMatchNvdApiProducer",
+        autospec=True,
+    )
+    def test_retry_attempts(self, mock_producer: MagicMock):
+        console = Console(quiet=True)
+        error_console = Console(quiet=True)
+        progress = Progress(disable=True)
+
+        args = parse_producer_args(
+            [
+                "--retry-attempts",
+                "7",
+            ]
+        )
+
+        CpeMatchNvdApiProducer.from_args(args, console, error_console, progress)
+
+        mock_producer.assert_called_once_with(
+            console=console,
+            error_console=error_console,
+            progress=progress,
+            nvd_api_key=None,
+            retry_attempts=7,
+            request_results=None,
+            request_filter_opts={},
+            start_index=None,
+            verbose=DEFAULT_VERBOSITY,
+        )
+
+    @patch(
+        "greenbone.scap.cpe_match.producer.nvd_api.CpeMatchNvdApiProducer",
+        autospec=True,
+    )
+    def test_nvd_api_key(self, mock_producer: MagicMock):
+        console = Console(quiet=True)
+        error_console = Console(quiet=True)
+        progress = Progress(disable=True)
+
+        args = parse_producer_args(
+            [
+                "--nvd-api-key",
+                "token123",
+            ]
+        )
+
+        CpeMatchNvdApiProducer.from_args(args, console, error_console, progress)
+
+        mock_producer.assert_called_once_with(
+            console=console,
+            error_console=error_console,
+            progress=progress,
+            nvd_api_key="token123",
+            retry_attempts=DEFAULT_RETRIES,
+            request_results=None,
+            request_filter_opts={},
+            start_index=None,
+            verbose=DEFAULT_VERBOSITY,
+        )

tests/cpe_match/worker/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later