tests: remotes: use TmpDir-like fixtures (#4140)

tests/utils/httpd.py

@@ -1,14 +1,48 @@
 import hashlib
 import os
+import sys
 import threading
 from http import HTTPStatus
-from http.server import HTTPServer, SimpleHTTPRequestHandler
+from http.server import HTTPServer
 
 from RangeHTTPServer import RangeRequestHandler
 
 
 class TestRequestHandler(RangeRequestHandler):
-    checksum_header = None
+    def __init__(self, *args, **kwargs):
+        # NOTE: `directory` was introduced in 3.7
+        if sys.version_info < (3, 7):
+            self.directory = kwargs.pop("directory", None) or os.getcwd()
+        super().__init__(*args, **kwargs)
+
+    def translate_path(self, path):
+        import urllib
+        import posixpath
+
+        # NOTE: `directory` was introduced in 3.7
+        if sys.version_info < (3, 7):
+            return super().translate_path(path)
+
+        path = path.split("?", 1)[0]
+        path = path.split("#", 1)[0]
+        # Don't forget explicit trailing slash when normalizing. Issue17324
+        trailing_slash = path.rstrip().endswith("/")
+        try:
+            path = urllib.parse.unquote(path, errors="surrogatepass")
+        except UnicodeDecodeError:
+            path = urllib.parse.unquote(path)
+        path = posixpath.normpath(path)
+        words = path.split("/")
+        words = filter(None, words)
+        path = self.directory
+        for word in words:
+            if os.path.dirname(word) or word in (os.curdir, os.pardir):
+                # Ignore components that are not a simple file/directory name
+                continue
+            path = os.path.join(path, word)
+        if trailing_slash:
+            path += "/"
+        return path
 
     def end_headers(self):
         # RangeRequestHandler only sends Accept-Ranges header if Range header
@@ -17,27 +51,16 @@ class TestRequestHandler(RangeRequestHandler):
             self.send_header("Accept-Ranges", "bytes")
 
         # Add a checksum header
-        if self.checksum_header:
-            file = self.translate_path(self.path)
+        file = self.translate_path(self.path)
 
-            if not os.path.isdir(file) and os.path.exists(file):
-                with open(file) as fd:
-                    encoded_text = fd.read().encode("utf8")
-                    checksum = hashlib.md5(encoded_text).hexdigest()
-                    self.send_header(self.checksum_header, checksum)
+        if not os.path.isdir(file) and os.path.exists(file):
+            with open(file) as fd:
+                encoded_text = fd.read().encode("utf8")
+                checksum = hashlib.md5(encoded_text).hexdigest()
+                self.send_header("Content-MD5", checksum)
 
         RangeRequestHandler.end_headers(self)
 
-
-class ETagHandler(TestRequestHandler):
-    checksum_header = "ETag"
-
-
-class ContentMD5Handler(TestRequestHandler):
-    checksum_header = "Content-MD5"
-
-
-class PushRequestHandler(SimpleHTTPRequestHandler):
     def _chunks(self):
         while True:
             data = self.rfile.readline(65537)
@@ -69,9 +92,13 @@ class PushRequestHandler(SimpleHTTPRequestHandler):
 class StaticFileServer:
     _lock = threading.Lock()
 
-    def __init__(self, handler_class=ETagHandler):
+    def __init__(self, directory):
+        from functools import partial
+
         self._lock.acquire()
-        self._httpd = HTTPServer(("localhost", 0), handler_class)
+        self._httpd = HTTPServer(
+            ("localhost", 0), partial(TestRequestHandler, directory=directory),
+        )
         self._thread = None
 
     def __enter__(self):