add tests for get_package_version

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

vulnerabilities/importers/github.py

@@ -20,7 +20,6 @@
 #  VulnerableCode is a free software  from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
-import json
 import logging
 import os
 from datetime import datetime
@@ -48,13 +47,13 @@
 from vulnerabilities.improver import Improver
 from vulnerabilities.improver import Inference
 from vulnerabilities.models import Advisory
-from vulnerabilities.package_managers import Version as PackageVersion
 from vulnerabilities.package_managers_2 import ComposerVersionAPI
 from vulnerabilities.package_managers_2 import GoproxyVersionAPI
 from vulnerabilities.package_managers_2 import MavenVersionAPI
 from vulnerabilities.package_managers_2 import NugetVersionAPI
 from vulnerabilities.package_managers_2 import PypiVersionAPI
 from vulnerabilities.package_managers_2 import RubyVersionAPI
+from vulnerabilities.package_managers_2 import Version as PackageVersion
 from vulnerabilities.package_managers_2 import VersionAPI
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
 
@@ -434,24 +433,23 @@ def interesting_advisories(self) -> QuerySet:
         return Advisory.objects.filter(created_by=GitHubAPIImporter.qualified_name)
 
     def get_package_versions(
-        self, package_url: PackageURL, until: Optional[datetime]
+        self, package_url: PackageURL, until: Optional[datetime] = None
     ) -> List[PackageVersion]:
         """
         Return a list of `PackageVersion` for the `package_url`
         """
-        purl = str(package_url)
-        version_api = self.version_api_by_purl_type.get(purl.type)
+        api_name = get_api_package_name(package_url)
+        if not api_name:
+            LOGGER.error(f"Could not get versions for {package_url!r}")
+            return []
+        version_api = self.version_api_by_purl_type.get(package_url.type)
         if not version_api:
-            version_api: VersionAPI = VERSION_API_CLASSES_BY_PACKAGE_TYPE(package_url.type)
-            api_name = get_api_package_name(package_url)
-            if not api_name:
-                LOGGER.error(f"Could not get versions for {package_url!r}")
-                return []
-
-            version_api.load_api([api_name])
-            self.version_api_by_purl_type[purl.type] = version_api()
-
-        return version_api.get(package_name=api_name, until=until).valid_versions
+            version_api: VersionAPI = VERSION_API_CLASSES_BY_PACKAGE_TYPE[package_url.type]
+            self.version_api_by_purl_type[package_url.type] = version_api()
+        api_object = self.version_api_by_purl_type[package_url.type]
+        api_object.load_api([api_name])
+        self.version_api_by_purl_type[package_url.type] = api_object
+        return api_object.get(package_name=api_name, until=until).valid_versions
 
     def get_inferences(self, advisory_data: AdvisoryData) -> Iterable[Inference]:
         """

vulnerabilities/package_managers_2.py

@@ -1,20 +1,43 @@
+import dataclasses
+import logging
 import traceback
 import xml.etree.ElementTree as ET
+from datetime import datetime
 from json import JSONDecodeError
 from typing import List
 from typing import MutableMapping
 from typing import Optional
 from typing import Set
 
-from aiohttp import ClientResponseError
-from aiohttp import ClientSession
-from aiohttp.web_exceptions import HTTPGone
+import requests
 from dateutil import parser as dateparser
 from django.utils.dateparse import parse_datetime
 
 from vulnerabilities.package_managers import Version as LegacyVersion
 from vulnerabilities.package_managers import VersionResponse
 
+LOGGER = logging.getLogger(__name__)
+
+
+@dataclasses.dataclass(frozen=True)
+class Version:
+    value: str
+    release_date: Optional[datetime] = None
+
+
+def get_response(url, type="json"):
+    resp = requests.get(url=url)
+    type_of_response = {
+        "json": resp.json(),
+        "text": resp.text,
+        "read": resp.content,
+    }
+    if resp.status_code == 200:
+        return type_of_response[type]
+    else:
+        LOGGER.error(f"Error while fetching {url}: {resp.status_code}")
+        return None
+
 
 class VersionAPI:
     def __init__(self, cache: MutableMapping[str, Set[LegacyVersion]] = None):
@@ -35,13 +58,12 @@ def load_api(self, pkg_set):
         """
         Populate the cache with the versions of the packages in pkg_set
         """
-        with ClientSession() as session:
-            for pkg in pkg_set:
-                if pkg in self.cache:
-                    continue
-                self.fetch(pkg, session)
+        for pkg in pkg_set:
+            if pkg in self.cache:
+                continue
+            self.fetch(pkg)
 
-    def fetch(self, pkg, session):
+    def fetch(self, pkg):
         """
         Override this method to fetch the pkg's version in the cache
         """
@@ -52,54 +74,50 @@ class PypiVersionAPI(VersionAPI):
 
     package_type = "pypi"
 
-    def fetch(self, pkg, session):
+    def fetch(self, pkg):
         url = f"https://pypi.org/pypi/{pkg}/json"
         versions = set()
-        try:
-            response = session.request(method="GET", url=url)
-            response = response.json()
-            for version, download_items in response["releases"].items():
-                if download_items:
-                    latest_download_item = max(
-                        download_items,
-                        key=lambda download_item: dateparser.parse(
-                            download_item["upload_time_iso_8601"]
-                        ),
-                    )
-                    versions.add(
-                        LegacyVersion(
-                            value=version,
-                            release_date=dateparser.parse(
-                                latest_download_item["upload_time_iso_8601"]
-                            ),
-                        )
+        response = get_response(url=url)
+
+        if not response:
+            self.cache[pkg] = versions
+            return
+
+        for version, download_items in response["releases"].items():
+            if download_items:
+                latest_download_item = max(
+                    download_items,
+                    key=lambda download_item: dateparser.parse(
+                        download_item["upload_time_iso_8601"]
+                    ),
+                )
+                versions.add(
+                    LegacyVersion(
+                        value=version,
+                        release_date=dateparser.parse(latest_download_item["upload_time_iso_8601"]),
                     )
-        except ClientResponseError:
-            # PYPI removed this package.
-            # https://www.zdnet.com/article/twelve-malicious-python-libraries-found-and-removed-from-pypi/  # nopep8
-            pass
+                )
         self.cache[pkg] = versions
 
 
 class RubyVersionAPI(VersionAPI):
 
     package_type = "gem"
 
-    def fetch(self, pkg, session):
+    def fetch(self, pkg):
         url = f"https://rubygems.org/api/v1/versions/{pkg}.json"
         versions = set()
-        try:
-            response = session.request(method="GET", url=url)
-            response = response.json()
-            for release in response:
-                versions.add(
-                    LegacyVersion(
-                        value=release["number"],
-                        release_date=dateparser.parse(release["created_at"]),
-                    )
+        response = get_response(url=url)
+        if not response:
+            self.cache[pkg] = versions
+            return
+        for release in response:
+            versions.add(
+                LegacyVersion(
+                    value=release["number"],
+                    release_date=dateparser.parse(release["created_at"]),
                 )
-        except (ClientResponseError, JSONDecodeError):
-            pass
+            )
 
         self.cache[pkg] = versions
 
@@ -108,22 +126,20 @@ class NpmVersionAPI(VersionAPI):
 
     package_type = "npm"
 
-    def fetch(self, pkg, session):
+    def fetch(self, pkg):
         url = f"https://registry.npmjs.org/{pkg}"
         versions = set()
-        try:
-            response = session.request(method="GET", url=url)
-            response = response.json()
-            for version in response.get("versions", []):
-                release_date = response.get("time", {}).get(version)
-                if release_date:
-                    release_date = dateparser.parse(release_date)
-                    versions.add(LegacyVersion(value=version, release_date=release_date))
-                else:
-                    versions.add(LegacyVersion(value=version, release_date=None))
-
-        except ClientResponseError:
-            pass
+        response = get_response(url=url)
+        if not response:
+            self.cache[pkg] = versions
+            return
+        for version in response.get("versions", []):
+            release_date = response.get("time", {}).get(version)
+            if release_date:
+                release_date = dateparser.parse(release_date)
+                versions.add(LegacyVersion(value=version, release_date=release_date))
+            else:
+                versions.add(LegacyVersion(value=version, release_date=None))
 
         self.cache[pkg] = versions
 
@@ -132,14 +148,13 @@ class MavenVersionAPI(VersionAPI):
 
     package_type = "maven"
 
-    def fetch(self, pkg, session) -> None:
+    def fetch(self, pkg) -> None:
         artifact_comps = pkg.split(":")
         endpoint = self.artifact_url(artifact_comps)
-        try:
-            resp = session.request(method="GET", url=endpoint)
-            resp = resp.read()
 
-        except ClientResponseError:
+        resp = get_response(url=endpoint, type="read")
+
+        if not resp:
             self.cache[pkg] = set()
             return
 
@@ -182,10 +197,12 @@ class NugetVersionAPI(VersionAPI):
 
     package_type = "nuget"
 
-    def fetch(self, pkg, session) -> None:
+    def fetch(self, pkg) -> None:
         endpoint = self.nuget_url(pkg)
-        resp = session.request(method="GET", url=endpoint)
-        resp = resp.json()
+        resp = get_response(url=endpoint)
+        if not resp:
+            self.cache[pkg] = set()
+            return
         self.cache[pkg] = self.extract_versions(resp)
 
     @staticmethod
@@ -271,9 +288,7 @@ def escape_path(path: str) -> str:
         return escaped_path
 
     @staticmethod
-    def parse_version_info(
-        version_info: str, escaped_pkg: str, session: ClientSession
-    ) -> Optional[LegacyVersion]:
+    def parse_version_info(version_info: str, escaped_pkg: str) -> Optional[LegacyVersion]:
         v = version_info.split()
         if not v:
             return None
@@ -283,33 +298,28 @@ def parse_version_info(
             release_date = parse_datetime(v[1])
         else:
             escaped_ver = GoproxyVersionAPI.escape_path(value)
-            try:
-                response = session.request(
-                    method="GET",
-                    url=f"https://proxy.golang.org/{escaped_pkg}/@v/{escaped_ver}.info",
-                )
-                resp_json = response.json()
-                release_date = parse_datetime(resp_json.get("Time", ""))
-            except:
+            resp_json = get_response(
+                url=f"https://proxy.golang.org/{escaped_pkg}/@v/{escaped_ver}.info"
+            )
+            if not resp_json:
                 traceback.print_exc()
                 print(
                     f"error while fetching version info for {escaped_pkg}/{escaped_ver} from goproxy"
                 )
-                release_date = None
+            release_date = parse_datetime(resp_json.get("Time", "")) if resp_json else None
+
         return LegacyVersion(value=value, release_date=release_date)
 
-    def fetch(self, pkg: str, session: ClientSession):
+    def fetch(self, pkg: str):
         # escape uppercase in module path
         escaped_pkg = GoproxyVersionAPI.escape_path(pkg)
         trimmed_pkg = pkg
         resp_text = None
         # resolve module name from package name, see https://go.dev/ref/mod#resolve-pkg-mod
         while escaped_pkg is not None:
             url = f"https://proxy.golang.org/{escaped_pkg}/@v/list"
-            try:
-                response = session.request(method="GET", url=url)
-                resp_text = response.text()
-            except HTTPGone:
+            resp_text = get_response(url=url, type="text")
+            if not resp_text:
                 escaped_pkg = GoproxyVersionAPI.trim_url_path(escaped_pkg)
                 trimmed_pkg = GoproxyVersionAPI.trim_url_path(trimmed_pkg) or ""
                 continue
@@ -320,7 +330,7 @@ def fetch(self, pkg: str, session: ClientSession):
         self.module_name_by_package_name[pkg] = trimmed_pkg
         versions = set()
         for version_info in resp_text.split("\n"):
-            version = GoproxyVersionAPI.parse_version_info(version_info, escaped_pkg, session)
+            version = GoproxyVersionAPI.parse_version_info(version_info, escaped_pkg)
             if version is not None:
                 versions.add(version)
         self.cache[pkg] = versions
@@ -330,11 +340,13 @@ class ComposerVersionAPI(VersionAPI):
 
     package_type = "composer"
 
-    def fetch(self, pkg, session) -> None:
+    def fetch(self, pkg) -> None:
         endpoint = self.composer_url(pkg)
         if endpoint:
-            resp = session.request(method="GET", url=endpoint)
-            resp = resp.json()
+            resp = get_response(url=endpoint)
+            if not resp:
+                self.cache[pkg] = set()
+                return
             self.cache[pkg] = self.extract_versions(resp, pkg)
 
     @staticmethod