diff --git a/safety/scan/ecosystems/python/main.py b/safety/scan/ecosystems/python/main.py index 12d2af45..5ce3d1c9 100644 --- a/safety/scan/ecosystems/python/main.py +++ b/safety/scan/ecosystems/python/main.py @@ -277,6 +277,22 @@ def __init__(self, file_type: FileType, file: FileTextWrite) -> None: self.ecosystem = file_type.ecosystem self.file_type = file_type + def _load_full_db(self) -> dict | None: + """ + Lazily loads the full vulnerability database from the local cache. + + Returns: + dict | None: The full database, or None if unavailable or invalid. + """ + db_full = get_from_cache( + db_name="insecure_full.json", skip_time_verification=True + ) + if not db_full: + LOG.debug( + "Cache data for insecure_full.json is not available or is invalid." + ) + return db_full + def __find_dependency_vulnerabilities__( self, dependencies: List[PythonDependency], config: ConfigModel ) -> None: @@ -325,13 +341,8 @@ def __find_dependency_vulnerabilities__( if not dependency.version: if not db_full: - db_full = get_from_cache( - db_name="insecure_full.json", skip_time_verification=True - ) + db_full = self._load_full_db() if not db_full: - LOG.debug( - "Cache data for insecure_full.json is not available or is invalid." - ) return dependency.refresh_from(db_full) @@ -342,14 +353,8 @@ def __find_dependency_vulnerabilities__( if spec.is_vulnerable(spec_set, dependency.insecure_versions): if not db_full: - db_full = get_from_cache( - db_name="insecure_full.json", - skip_time_verification=True, - ) + db_full = self._load_full_db() if not db_full: - LOG.debug( - "Cache data for insecure_full.json is not available or is invalid." - ) return if not dependency.latest_version: dependency.refresh_from(db_full) @@ -497,9 +502,7 @@ def remediate(self) -> None: """ Remediates the vulnerabilities in the file. """ - db_full = get_from_cache( - db_name="insecure_full.json", skip_time_verification=True - ) + db_full = self._load_full_db() if not db_full: return diff --git a/tests/scan/ecosystems/python/test_main.py b/tests/scan/ecosystems/python/test_main.py index 8960af63..f7d786be 100644 --- a/tests/scan/ecosystems/python/test_main.py +++ b/tests/scan/ecosystems/python/test_main.py @@ -1,6 +1,10 @@ import unittest -from unittest.mock import MagicMock -from safety.scan.ecosystems.python.main import should_fail, VulnerabilitySeverityLabels +from unittest.mock import MagicMock, patch +from safety.scan.ecosystems.python.main import ( + PythonFile, + should_fail, + VulnerabilitySeverityLabels, +) class TestMain(unittest.TestCase): @@ -80,3 +84,30 @@ def test_unexpected_severity_with_warning(self): result = should_fail(self.config, self.vulnerability) self.assertIn("Unexpected base severity value", log.output[0]) self.assertFalse(result) + + +class TestLoadFullDb(unittest.TestCase): + def setUp(self): + self.python_file = PythonFile(file_type=MagicMock(), file=MagicMock()) + + @patch("safety.scan.ecosystems.python.main.get_from_cache") + def test_returns_db_when_cache_is_available(self, mock_get_from_cache): + db_full = {"vulnerable_packages": {}} + mock_get_from_cache.return_value = db_full + + result = self.python_file._load_full_db() + self.assertIs(result, db_full) + + mock_get_from_cache.assert_called_once_with( + db_name="insecure_full.json", skip_time_verification=True + ) + + @patch("safety.scan.ecosystems.python.main.get_from_cache") + def test_returns_none_and_logs_when_cache_is_missing(self, mock_get_from_cache): + mock_get_from_cache.return_value = None + + with self.assertLogs(level="DEBUG") as log: + result = self.python_file._load_full_db() + + self.assertIsNone(result) + self.assertIn("insecure_full.json", log.output[0])