refactor: Abstract general behaviour from TopPypiReference (#299)

Author: sdn4z

src/twyn/base/exceptions.py

@@ -30,9 +30,3 @@ def __init__(self, message: str = "") -> None:
     def show(self, file: Optional[IO[Any]] = None) -> None:
         logger.debug(self.format_message(), exc_info=True)
         logger.error(self.format_message(), exc_info=False)
-
-
-class PackageNormalizingError(TwynError):
-    """Exception for when it is not possible to normalize a package name."""
-
-    message = "Failed to normalize pacakges."

src/twyn/base/utils.py

@@ -7,7 +7,6 @@
     SELECTOR_METHOD_MAPPING,
     SelectorMethod,
 )
-from twyn.base.utils import normalize_packages
 from twyn.config.config_handler import ConfigHandler
 from twyn.dependency_parser.dependency_selector import DependencySelector
 from twyn.file_handler.file_handler import FileHandler
@@ -68,9 +67,9 @@ def check_dependencies(
         selector=_get_candidate_selector(config.selector_method),
         threshold_class=SimilarityThreshold,
     )
-    normalized_allowlist_packages = normalize_packages(config.allowlist)
+    normalized_allowlist_packages = TopPyPiReference.normalize_packages(config.allowlist)
     dependencies = dependencies if dependencies else _get_parsed_dependencies_from_file(config.dependency_file)
-    normalized_dependencies = normalize_packages(dependencies)
+    normalized_dependencies = TopPyPiReference.normalize_packages(dependencies)
 
     typos_list = TyposquatCheckResultList()
     dependencies_list = (

src/twyn/main.py

@@ -8,8 +8,6 @@
 
 from pydantic import BaseModel, ValidationError, field_validator
 
-from twyn.base.exceptions import PackageNormalizingError
-from twyn.base.utils import normalize_packages
 from twyn.file_handler.exceptions import PathIsNotFileError, PathNotFoundError
 from twyn.file_handler.file_handler import FileHandler
 from twyn.trusted_packages.constants import CACHE_DIR, TRUSTED_PACKAGES_MAX_RETENTION_DAYS
@@ -31,14 +29,6 @@ def validate_saved_date(cls, v: str) -> str:
         else:
             return v
 
-    @field_validator("packages")
-    @classmethod
-    def validate_packages(cls, v: set[str]) -> set[str]:
-        try:
-            return normalize_packages(v)
-        except PackageNormalizingError as e:
-            raise ValueError(f"Failed to normalize packages: {e}") from e
-
 
 class CacheHandler:
     """Cache class that provides basic read/write/delete operation for individual source cache files."""

src/twyn/trusted_packages/cache_handler.py

@@ -29,3 +29,9 @@ class InvalidCacheError(TwynError):
     """Error for when the cache content is not valid."""
 
     message = "Invalid cache content"
+
+
+class PackageNormalizingError(TwynError):
+    """Exception for when it is not possible to normalize a package name."""
+
+    message = "Failed to normalize pacakges."

src/twyn/trusted_packages/exceptions.py

@@ -1,52 +1,56 @@
 import logging
-from abc import ABC, abstractmethod
+import re
+from abc import abstractmethod
 from datetime import datetime
 from typing import Any, Union
 
 import requests
+from typing_extensions import override
 
-from twyn.base.utils import normalize_packages
 from twyn.trusted_packages.cache_handler import CacheEntry, CacheHandler
 from twyn.trusted_packages.exceptions import (
     EmptyPackagesListError,
     InvalidJSONError,
     InvalidPyPiFormatError,
+    PackageNormalizingError,
 )
 
 logger = logging.getLogger("twyn")
 
 
-class AbstractPackageReference(ABC):
-    """Represents a reference from where to retrieve trusted packages."""
+class AbstractPackageReference:
+    """Represents a reference from where to retrieve trusted packages.
+
+    It abstracts all the package-retrieval and caching logic.
+
+    It defines the `_parse` abstract method, so each subclass defines how to handle the feched data.
+    It defines the `normalize_package` abstract method, so each subclass validates that the packages names are correct.
+    """
 
     def __init__(self, source: str, cache_handler: Union[CacheHandler, None] = None) -> None:
         self.source = source
         self.cache_handler = cache_handler
 
+    @staticmethod
     @abstractmethod
-    def get_packages(self) -> set[str]:
-        """Return the names of the trusted packages available in the reference."""
-
-
-class TopPyPiReference(AbstractPackageReference):
-    """Top PyPi packages retrieved from an online source."""
-
-    def get_packages(self) -> set[str]:
-        """Download and parse online source of top Python Package Index packages."""
-        packages_to_use = set()
-        packages_to_use = self._get_packages_from_cache_if_enabled()
-        # we don't save the cache here, we keep it as it is so the date remains the original one.
-
-        if not packages_to_use:
-            # no cache usage, no cache hit (non-existent or outdated) or cache was empty.
-            logger.info("Fetching trusted packages from PyPI reference...")
-            packages_to_use = self._parse(self._download())
+    def _parse(packages_json: dict[str, Any]) -> set[str]:
+        """Parse and retrieve the packages within the given json structure."""
 
-            # New packages were downloaded, we create a new entry updating all values.
-            self._save_trusted_packages_to_cache_if_enabled(packages_to_use)
+    @staticmethod
+    @abstractmethod
+    def normalize_packages(packages: set[str]) -> set[str]:
+        """Normalize package names to make sure they're valid within the package manager context."""
 
-        normalized_packages = normalize_packages(packages_to_use)
-        return normalized_packages
+    def _download(self) -> dict[str, Any]:
+        packages = requests.get(self.source)
+        packages.raise_for_status()
+        try:
+            packages_json: dict[str, Any] = packages.json()
+        except requests.exceptions.JSONDecodeError as err:
+            raise InvalidJSONError from err
+        else:
+            logger.debug("Successfully downloaded trusted packages list from %s", self.source)
+            return packages_json
 
     def _save_trusted_packages_to_cache_if_enabled(self, packages: set[str]) -> None:
         """Save trusted packages using CacheHandler."""
@@ -67,18 +71,28 @@ def _get_packages_from_cache_if_enabled(self) -> set[str]:
 
         return cache_entry.packages
 
-    def _download(self) -> dict[str, Any]:
-        packages = requests.get(self.source)
-        packages.raise_for_status()
-        try:
-            packages_json: dict[str, Any] = packages.json()
-        except requests.exceptions.JSONDecodeError as err:
-            raise InvalidJSONError from err
+    def get_packages(self) -> set[str]:
+        """Download and parse online source of top Python Package Index packages."""
+        packages_to_use = set()
+        packages_to_use = self._get_packages_from_cache_if_enabled()
+        # we don't save the cache here, we keep it as it is so the date remains the original one.
+
+        if not packages_to_use:
+            # no cache usage, no cache hit (non-existent or outdated) or cache was empty.
+            logger.info("Fetching trusted packages from PyPI reference...")
+            packages_to_use = self._parse(self._download())
+
+            # New packages were downloaded, we create a new entry updating all values.
+            self._save_trusted_packages_to_cache_if_enabled(packages_to_use)
 
-        logger.debug("Successfully downloaded trusted packages list from %s", self.source)
+        normalized_packages = self.normalize_packages(packages_to_use)
+        return normalized_packages
 
-        return packages_json
 
+class TopPyPiReference(AbstractPackageReference):
+    """Top PyPi packages retrieved from an online source."""
+
+    @override
     @staticmethod
     def _parse(packages_info: dict[str, Any]) -> set[str]:
         try:
@@ -90,5 +104,17 @@ def _parse(packages_info: dict[str, Any]) -> set[str]:
             raise EmptyPackagesListError
 
         logger.debug("Successfully parsed trusted packages list")
-
         return names
+
+    @override
+    @staticmethod
+    def normalize_packages(packages: set[str]) -> set[str]:
+        """Normalize dependency names according to PyPi https://packaging.python.org/en/latest/specifications/name-normalization/."""
+        renamed_packages = {re.sub(r"[-_.]+", "-", name).lower() for name in packages}
+
+        pattern = re.compile(r"^([a-z0-9]|[a-z0-9][a-z0-9._-]*[a-z0-9])\Z")  # noqa: F821
+        for package in renamed_packages:
+            if not pattern.match(package):
+                raise PackageNormalizingError(f"Package name '{package}' does not match required pattern")
+
+        return renamed_packages

src/twyn/trusted_packages/references.py

@@ -172,12 +172,12 @@ def test_check_dependencies_with_input_from_cli_accepts_multiple_dependencies(
         assert TyposquatCheckResult(dependency="reqests", similars=["requests"]) in error.errors
         assert TyposquatCheckResult(dependency="my-package", similars=["mypackage"]) in error.errors
 
-    @patch("twyn.main.TopPyPiReference")
+    @patch("twyn.main.TopPyPiReference.get_packages")
     @patch("twyn.main._get_parsed_dependencies_from_file")
     def test_check_dependencies_ignores_package_in_allowlist(
-        self, mock_get_parsed_dependencies_from_file: Mock, mock_top_pypi_reference: Mock
+        self, mock_get_parsed_dependencies_from_file: Mock, mock_get_packages: Mock
     ) -> None:
-        mock_top_pypi_reference.return_value.get_packages.return_value = {"mypackage"}
+        mock_get_packages.return_value = {"mypackage"}
         mock_get_parsed_dependencies_from_file.return_value = {"my-package"}
 
         # Verify that before the whitelist configuration the package is classified as an error.

tests/main/test_main.py

@@ -121,11 +121,6 @@ def test_is_entry_outdated_with_invalid_date_format(self) -> None:
         with pytest.raises(ValidationError):
             CacheEntry(saved_date="invalid-date-format", packages={"package1"})
 
-    def test_cache_entry_invalid_package_normalization(self) -> None:
-        """Test CacheEntry raises ValidationError when package normalization fails."""
-        with pytest.raises(ValidationError):
-            CacheEntry(saved_date="2024-01-01", packages={"../invalid-package"})
-
     def test_clear_all_removes_all_cache_files(self, tmp_path: Path) -> None:
         """Test clear_all removes all cache files and directory."""
         cache_handler = CacheHandler(str(tmp_path))