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

setup.cfg

@@ -8,3 +8,5 @@ extend-ignore = E203, E501
 
 [tool:pytest]
 asyncio_mode = strict
+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")