diff --git a/setup.cfg b/setup.cfg index 674003e1ba..f152945c8d 100644 --- a/setup.cfg +++ b/setup.cfg @@ -8,3 +8,5 @@ extend-ignore = E203, E501 [tool:pytest] asyncio_mode = strict +markers = + synchronous: Marks tests that are synchronous diff --git a/test/test_cvedb.py b/test/test_cvedb.py index 80742ebbb4..cd9821950b 100644 --- a/test/test_cvedb.py +++ b/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")