refactor: use http wrapper in amazon provider

src/vunnel/providers/amazon/parser.py

@@ -7,10 +7,8 @@
 from html.parser import HTMLParser
 
 import defusedxml.ElementTree as ET
-import requests
 
-from vunnel import utils
-from vunnel.utils import rpm
+from vunnel.utils import http, rpm
 
 namespace = "amzn"
 
@@ -48,17 +46,13 @@ def __init__(self, workspace, download_timeout=125, security_advisories=None, lo
             logger = logging.getLogger(self.__class__.__name__)
         self.logger = logger
 
-    @utils.retry_with_backoff()
     def _download_rss(self, rss_url, rss_file):
         try:
             self.logger.info(f"downloading amazon security advisory from {rss_url}")
             self.urls.append(rss_url)
-            r = requests.get(rss_url, timeout=self.download_timeout)
-            if r.status_code == 200:
-                with open(rss_file, "w", encoding="utf-8") as fp:
-                    fp.write(r.text)
-            else:
-                raise Exception(f"GET {rss_url} failed with HTTP error {r.status_code}")
+            r = http.get(rss_url, self.logger, timeout=self.download_timeout)
+            with open(rss_file, "w", encoding="utf-8") as fp:
+                fp.write(r.text)
         except Exception:
             self.logger.exception("error downloading amazon linux vulnerability feeds")
             raise
@@ -91,23 +85,18 @@ def _parse_rss(self, file_path):
 
         return sorted(alas_summaries)
 
-    @utils.retry_with_backoff()
     def _get_alas_html(self, alas_url, alas_file, skip_if_exists=True):
         if skip_if_exists and os.path.exists(alas_file):  # read alas from disk if its available
             self.logger.debug(f"loading existing ALAS from {alas_file}")
             with open(alas_file, encoding="utf-8") as fp:
                 content = fp.read()
             return content  # noqa: RET504
-
         try:
-            self.logger.debug(f"downloading ALAS from {alas_url}")
-            r = requests.get(alas_url, timeout=self.download_timeout)
-            if r.status_code == 200:
-                content = r.text
-                with open(alas_file, "w", encoding="utf-8") as fp:
-                    fp.write(content)
-                return content
-            raise Exception(f"GET {alas_url} failed with HTTP error {r.status_code}")
+            r = http.get(alas_url, self.logger, timeout=self.download_timeout)
+            content = r.text
+            with open(alas_file, "w", encoding="utf-8") as fp:
+                fp.write(content)
+            return content
         except Exception:
             self.logger.exception(f"error downloading data from {alas_url}")
             raise

tests/unit/providers/amazon/test_amazon.py

@@ -4,6 +4,7 @@
 
 import pytest
 from vunnel import result, workspace
+from vunnel.utils import http
 from vunnel.providers.amazon import Config, Provider, parser
 
 
@@ -76,14 +77,6 @@ def test_get_pkg_name_version(self):
         assert a == b
 
 
-@pytest.fixture()
-def disable_get_requests(monkeypatch):
-    def disabled(*args, **kwargs):
-        raise RuntimeError("requests disabled but HTTP GET attempted")
-
-    monkeypatch.setattr(parser.requests, "get", disabled)
-
-
 def test_provider_schema(helpers, disable_get_requests, monkeypatch):
     workspace = helpers.provider_workspace_helper(name=Provider.name())