Merge pull request #592 from adanaja/features

Add kwargs parameter to HTTPSession.download_file

Author: adanaja

src/e3/net/http.py

@@ -240,6 +240,7 @@ def download_file(
         filename: str | None = None,
         validate: Callable[[str], bool] | None = None,
         exception_on_error: bool = False,
+        **kwargs: Any,
     ) -> str | None:
         """Download a file.
 
@@ -253,6 +254,7 @@ def download_file(
             parameter and returns a boolean.
         :param exception_on_error: if True raises an exception in case download
             fails instead of returning None.
+        :param kwargs: additional parameters for the request
         :return: the name of the file or None if there is an error
         """
         # When using stream=True, Requests cannot release the connection back
@@ -262,7 +264,7 @@ def download_file(
         path = None
         try:
             with contextlib.closing(
-                self.request(method="GET", url=url, stream=True)
+                self.request(method="GET", url=url, stream=True, **kwargs)
             ) as response:
                 content_length = int(response.headers.get("content-length", 0))
                 e3.log.debug(response.headers)

tests/tests_e3/net/http_test.py

@@ -93,6 +93,17 @@ def do_POST(self):
         logging.debug("POST finish")
 
 
+class AuthorizationHeaderHandler(ContentDispoHandler):
+    def do_GET(self):
+        if self.headers.get("Authorization") != "Bearer toto":
+            self.send_response(403)
+            self.end_headers()
+            self.wfile.close()
+            return
+
+        super(AuthorizationHeaderHandler, self).do_GET()
+
+
 def run_server(handler, func):
     server = HTTPServer(("localhost", 0), handler)
     try:
@@ -220,3 +231,28 @@ def func(server, url):
             run_server(MultiPartPostHandler, func)
 
         run_server(ServerErrorHandler, outter_func)
+
+    def test_authorization_header(self, socket_enabled):
+        def func(server, base_url):
+            with HTTPSession() as session:
+                # first test with no authorization header
+                try:
+                    result = session.download_file(
+                        base_url + "dummy", dest=".", exception_on_error=True
+                    )
+                    raise AssertionError("exception not raised")
+                except HTTPError as e:
+                    assert e.status == 403
+                # second test with authorization header
+                result = session.download_file(
+                    base_url + "dummy",
+                    dest=".",
+                    exception_on_error=True,
+                    headers={"Authorization": "Bearer toto"},
+                )
+                with open(result, "rb") as fd:
+                    content = fd.read()
+                assert content == b"Dummy!"
+                assert os.path.basename(result) == "dummy.txt"
+
+        run_server(AuthorizationHeaderHandler, func)