test: Fix tests to reduce chance of 'database is locked' error. #4651

cve_bin_tool/cvedb.py

@@ -21,6 +21,7 @@
 
 import gnupg
 from rich.progress import track
+from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
 
 from cve_bin_tool.async_utils import run_coroutine
 from cve_bin_tool.data_sources import (
@@ -765,17 +766,66 @@ def update_vendors(self, cve_data):
 
     def db_open_and_get_cursor(self) -> sqlite3.Cursor:
         """Opens connection to sqlite database, returns cursor object."""
-
         if not self.connection:
-            self.connection = sqlite3.connect(self.dbpath)
+            self.connection = sqlite3.connect(self.dbpath, timeout=30)
         if self.connection is not None:
             cursor = self.connection.cursor()
         if cursor is None:
-            # if this happens somsething has gone horribly wrong
             LOGGER.error("Database cursor does not exist")
             raise CVEDBError
         return cursor
 
+    @retry(  # ADDED RETRY DECORATOR
+        retry=retry_if_exception_type(sqlite3.OperationalError),
+        stop=stop_after_attempt(5),
+        wait=wait_exponential(multiplier=1, min=1, max=10),
+    )
+    def refresh_cache_and_update_db(self) -> None:
+        """Refresh cached NVD and update CVE database with latest data."""
+        self.LOGGER.debug("Updating CVE data. This will take a few minutes.")
+        run_coroutine(self.refresh())
+        self.init_database()
+        self.populate_db()
+        self.LOGGER.debug("Updating exploits data.")
+        self.create_exploit_db()
+        self.update_exploits()
+
+    def get_cvelist_if_stale(self) -> None:
+        """Update if the local db is more than one day old."""
+        if not self.dbpath.is_file() or (
+            datetime.datetime.today()
+            - datetime.datetime.fromtimestamp(self.dbpath.stat().st_mtime)
+        ) > datetime.timedelta(hours=24):
+            try:
+                self.refresh_cache_and_update_db()
+            except sqlite3.OperationalError as e:
+                # ADDED ERROR HANDLING
+                raise CVEDBError(
+                    "Database lock failed after retries. Use '-u never' for parallel jobs."
+                ) from e
+            self.time_of_last_update = datetime.datetime.today()
+        else:
+            _ = self.get_db_update_date()
+            self.LOGGER.info(
+                "Using cached CVE data (<24h old). Use -u now to update immediately."
+            )
+            if (
+                not self.latest_schema(
+                    "cve_severity", self.TABLE_SCHEMAS["cve_severity"]
+                )
+                or not self.latest_schema("cve_range", self.TABLE_SCHEMAS["cve_range"])
+                or not self.latest_schema(
+                    "cve_exploited", self.TABLE_SCHEMAS["cve_exploited"]
+                )
+            ):
+                try:
+                    self.refresh_cache_and_update_db()
+                except sqlite3.OperationalError as e:
+                    raise CVEDBError(
+                        "Database lock failed after retries. Use '-u never' for parallel jobs."
+                    ) from e
+                self.time_of_last_update = datetime.datetime.today()
+
     def db_close(self) -> None:
         """Closes connection to sqlite database."""
         if self.connection:

setup.cfg

@@ -8,4 +8,7 @@ extend-ignore = E203, E501
 
 [tool:pytest]
 asyncio_mode = strict
-asyncio_default_fixture_loop_scope = function
+asyncio_default_fixture_loop_scope = function
+markers =
+    synchronous: Marks tests that are synchronous
+

test/test_cvedb.py

@@ -27,7 +27,7 @@ def teardown_class(cls):
         shutil.rmtree(cls.nvd.cachedir)
         shutil.rmtree(cls.exported_data)
 
-    @pytest.mark.asyncio
+    @pytest.mark.synchronous
     @pytest.mark.skipif(
         not EXTERNAL_SYSTEM(), reason="Skipping NVD calls due to rate limits"
     )
@@ -37,12 +37,13 @@ async def test_refresh_nvd_json(self):
         for year in range(2002, datetime.datetime.now().year):
             assert year in years, f"Missing NVD data for {year}"
 
+    @pytest.mark.synchronous
     @pytest.mark.skipif(not LONG_TESTS(), reason="Skipping long tests")
     def test_import_export_json(self):
         main(["cve-bin-tool", "-u", "never", "--export", self.nvd.cachedir])
         cve_entries_check = "SELECT data_source, COUNT(*) as number FROM cve_severity GROUP BY data_source ORDER BY number DESC"
         cursor = self.cvedb.db_open_and_get_cursor()
-        cursor.execute(cve_entries_check)
+        self.execute_with_retry(cursor, cve_entries_check)
         cve_entries_before = 0
         rows = cursor.fetchall()
         for row in rows:
@@ -57,7 +58,7 @@ def test_import_export_json(self):
             log_signature_error=False,
         )
         cursor = self.cvedb.db_open_and_get_cursor()
-        cursor.execute(cve_entries_check)
+        self.execute_with_retry(cursor, cve_entries_check)
         cve_entries_after = 0
         rows = cursor.fetchall()
         for row in rows:
@@ -91,3 +92,16 @@ def test_new_database_schema(self):
             assert all(column in column_names for column in required_columns[table])
 
         self.cvedb.db_close()
+
+    def execute_with_retry(self, cursor, query, retries=3, delay=1):  # Added helper function for retry logic
+        """Helper function to handle sqlite database lock errors."""
+        for attempt in range(retries):
+            try:
+                cursor.execute(query)
+                return
+            except sqlite3.OperationalError as e:
+                if "database is locked" in str(e):
+                    time.sleep(delay)
+                else:
+                    raise
+        raise sqlite3.OperationalError("Retries exhausted: database is still locked")