refactor(scan): deduplicate insecure_full.json loading

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
 

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])