Fix code formatting with black and isort

Signed-off-by: Sampurna Pyne <sampurnapyne1710@gmail.com>

Author: Samk1710

vulnerabilities/pipelines/v2_importers/euvd_importer.py

@@ -9,12 +9,12 @@
 
 import json
 import logging
-import requests
 import time
 from datetime import datetime
 from http import HTTPStatus
 from typing import Iterable
 
+import requests
 from dateutil import parser as dateparser
 
 from vulnerabilities.importer import AdvisoryData
@@ -51,7 +51,7 @@ def fetch_data(self):
         if self._cached_data is not None:
             logger.info(f"Using cached data: {len(self._cached_data)} items")
             return self._cached_data
-            
+
         headers = {"User-Agent": "VulnerableCode"}
         all_items = []
         page = 0
@@ -61,7 +61,7 @@ def fetch_data(self):
         logger.info(f"Fetching data from EUVD API: {self.url}")
 
         while True:
-            
+
             retry_count = 0
             success = False
 
@@ -75,7 +75,9 @@ def fetch_data(self):
                         retry_count += 1
                         if retry_count < max_retries:
                             sleep_time = min(10 * (2 ** min(retry_count - 1, 5)), 60)
-                            logger.info(f"Retrying page {page} in {sleep_time}s (attempt {retry_count}/{max_retries})")
+                            logger.info(
+                                f"Retrying page {page} in {sleep_time}s (attempt {retry_count}/{max_retries})"
+                            )
                             time.sleep(sleep_time)
                             continue
                         else:
@@ -87,23 +89,29 @@ def fetch_data(self):
 
                     if not items:
                         logger.info(f"No items in response for page {page}; stopping fetch.")
-                        logger.info(f"Fetch completed successfully. Total items collected: {len(all_items)}")
-                        
+                        logger.info(
+                            f"Fetch completed successfully. Total items collected: {len(all_items)}"
+                        )
+
                         # Cache the fetched data for reuse
                         self._cached_data = all_items
                         logger.info(f"Cached {len(all_items)} items for reuse")
-                        
+
                         return all_items
 
                     all_items.extend(items)
-                    logger.info(f"Fetched page {page}: {len(items)} items (total: {len(all_items)})")
+                    logger.info(
+                        f"Fetched page {page}: {len(items)} items (total: {len(all_items)})"
+                    )
                     success = True
                     page += 1
 
                 except requests.exceptions.Timeout as e:
                     retry_count += 1
                     if retry_count < max_retries:
-                        logger.warning(f"Timeout on page {page}: {e}. Retrying in 10s (attempt {retry_count}/{max_retries})")
+                        logger.warning(
+                            f"Timeout on page {page}: {e}. Retrying in 10s (attempt {retry_count}/{max_retries})"
+                        )
                         time.sleep(10)
                     else:
                         logger.error(f"Max retries reached for page {page} after timeout")
@@ -112,7 +120,9 @@ def fetch_data(self):
                 except Exception as e:
                     retry_count += 1
                     if retry_count < max_retries:
-                        logger.error(f"Error fetching page {page}: {e}. Retrying in 10s (attempt {retry_count}/{max_retries})")
+                        logger.error(
+                            f"Error fetching page {page}: {e}. Retrying in 10s (attempt {retry_count}/{max_retries})"
+                        )
                         time.sleep(10)
                     else:
                         logger.error(f"Max retries reached for page {page}")
@@ -134,41 +144,43 @@ def collect_advisories(self) -> Iterable[AdvisoryData]:
 
     def parse_advisory(self, raw_data: dict) -> AdvisoryData:
         advisory_id = raw_data.get("id", "")
-        
+
         aliases = [advisory_id] if advisory_id else []
         aliases_str = raw_data.get("aliases", "")
         if aliases_str:
             cve_aliases = [alias.strip() for alias in aliases_str.split("\n") if alias.strip()]
             aliases.extend(cve_aliases)
-        
+
         summary = raw_data.get("description", "")
-        
+
         date_published = None
         date_str = raw_data.get("datePublished", "")
         if date_str:
             try:
                 date_published = dateparser.parse(date_str)
                 if date_published and date_published.tzinfo is None:
-                    date_published = date_published.replace(tzinfo=datetime.now().astimezone().tzinfo)
+                    date_published = date_published.replace(
+                        tzinfo=datetime.now().astimezone().tzinfo
+                    )
             except Exception as e:
                 logger.warning(f"Failed to parse date '{date_str}': {e}")
-        
+
         references = []
         references_str = raw_data.get("references", "")
         if references_str:
             urls = [url.strip() for url in references_str.split("\n") if url.strip()]
             for url in urls:
                 references.append(ReferenceV2(url=url))
-        
+
         if advisory_id:
             advisory_url = f"https://euvd.enisa.europa.eu/vulnerability/{advisory_id}"
             references.append(ReferenceV2(url=advisory_url))
-        
+
         severities = []
         base_score = raw_data.get("baseScore")
         base_score_version = raw_data.get("baseScoreVersion")
         base_score_vector = raw_data.get("baseScoreVector")
-        
+
         if base_score and base_score_version:
             scoring_system = self.get_scoring_system(base_score_version)
             if scoring_system:
@@ -178,7 +190,7 @@ def parse_advisory(self, raw_data: dict) -> AdvisoryData:
                     scoring_elements=base_score_vector or "",
                 )
                 severities.append(severity)
-        
+
         return AdvisoryData(
             advisory_id=advisory_id,
             aliases=aliases,

vulnerabilities/tests/pipelines/v2_importers/test_euvd_importer_v2.py

@@ -46,15 +46,15 @@ def test_collect_advisories(self, mock_get):
         assert first.advisory_id == "EUVD-2025-197757"
         assert "EUVD-2025-197757" in first.aliases
         assert "CVE-2025-13284" in first.aliases
-        assert (
-            first.summary == "ThinPLUS vulnerability that allows remote code execution"
-        )
+        assert first.summary == "ThinPLUS vulnerability that allows remote code execution"
         assert first.date_published is not None
         assert len(first.severities) == 1
         assert first.severities[0].system.identifier == "cvssv3.1"
         assert first.severities[0].value == "9.8"
-        assert first.severities[0].scoring_elements == "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"
-        
+        assert (
+            first.severities[0].scoring_elements == "CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H"
+        )
+
         urls = [ref.url for ref in first.references_v2]
         assert "https://nvd.nist.gov/vuln/detail/CVE-2025-13284" in urls
         assert "https://euvd.enisa.europa.eu/vulnerability/EUVD-2025-197757" in urls
@@ -88,23 +88,23 @@ def test_collect_advisories(self, mock_get):
     def test_get_scoring_system(self):
         """Test CVSS version to scoring system mapping"""
         pipeline = EUVDImporterPipeline()
-        
+
         system_v4 = pipeline.get_scoring_system("4.0")
         assert system_v4 is not None
         assert system_v4.identifier == "cvssv4"
-        
+
         system_v31 = pipeline.get_scoring_system("3.1")
         assert system_v31 is not None
         assert system_v31.identifier == "cvssv3.1"
-        
+
         system_v3 = pipeline.get_scoring_system("3.0")
         assert system_v3 is not None
         assert system_v3.identifier == "cvssv3"
-        
+
         system_v2 = pipeline.get_scoring_system("2.0")
         assert system_v2 is not None
         assert system_v2.identifier == "cvssv2"
-        
+
         system_unknown = pipeline.get_scoring_system("unknown")
         assert system_unknown is None
 
@@ -122,4 +122,3 @@ def test_advisories_count(self, mock_get):
         count = pipeline.advisories_count()
 
         assert count == 3
-