]> gitweb @ CieloNegro.org - youtube-dl.git/commitdiff
[test/test_utils] Add test for #9588
authorYen Chi Hsuan <yan12125@gmail.com>
Thu, 26 May 2016 09:24:40 +0000 (17:24 +0800)
committerYen Chi Hsuan <yan12125@gmail.com>
Thu, 2 Jun 2016 07:00:49 +0000 (15:00 +0800)
test/test_http.py

index 15e0ad369d57966bef222bf35c422ad9bdb4e755..6b8493e5e6b9522ebe594872067f41d8eb365fc2 100644 (file)
@@ -16,6 +16,15 @@ import threading
 TEST_DIR = os.path.dirname(os.path.abspath(__file__))
 
 
+def http_server_port(httpd):
+    if os.name == 'java':
+        # In Jython SSLSocket is not a subclass of socket.socket
+        sock = httpd.socket.sock
+    else:
+        sock = httpd.socket
+    return sock.getsockname()[1]
+
+
 class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
     def log_message(self, format, *args):
         pass
@@ -31,6 +40,22 @@ class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
             self.send_header('Content-Type', 'video/mp4')
             self.end_headers()
             self.wfile.write(b'\x00\x00\x00\x00\x20\x66\x74[video]')
+        elif self.path == '/302':
+            if sys.version_info[0] == 3:
+                # XXX: Python 3 http server does not allow non-ASCII header values
+                self.send_response(404)
+                self.end_headers()
+                return
+
+            new_url = 'http://localhost:%d/中文.html' % http_server_port(self.server)
+            self.send_response(302)
+            self.send_header(b'Location', new_url.encode('utf-8'))
+            self.end_headers()
+        elif self.path == '/%E4%B8%AD%E6%96%87.html':
+            self.send_response(200)
+            self.send_header('Content-Type', 'text/html; charset=utf-8')
+            self.end_headers()
+            self.wfile.write(b'<html><video src="/vid.mp4" /></html>')
         else:
             assert False
 
@@ -47,18 +72,32 @@ class FakeLogger(object):
 
 
 class TestHTTP(unittest.TestCase):
+    def setUp(self):
+        self.httpd = compat_http_server.HTTPServer(
+            ('localhost', 0), HTTPTestRequestHandler)
+        self.port = http_server_port(self.httpd)
+        self.server_thread = threading.Thread(target=self.httpd.serve_forever)
+        self.server_thread.daemon = True
+        self.server_thread.start()
+
+    def test_unicode_path_redirection(self):
+        # XXX: Python 3 http server does not allow non-ASCII header values
+        if sys.version_info[0] == 3:
+            return
+
+        ydl = YoutubeDL({'logger': FakeLogger()})
+        r = ydl.extract_info('http://localhost:%d/302' % self.port)
+        self.assertEqual(r['url'], 'http://localhost:%d/vid.mp4' % self.port)
+
+
+class TestHTTPS(unittest.TestCase):
     def setUp(self):
         certfn = os.path.join(TEST_DIR, 'testcert.pem')
         self.httpd = compat_http_server.HTTPServer(
             ('localhost', 0), HTTPTestRequestHandler)
         self.httpd.socket = ssl.wrap_socket(
             self.httpd.socket, certfile=certfn, server_side=True)
-        if os.name == 'java':
-            # In Jython SSLSocket is not a subclass of socket.socket
-            sock = self.httpd.socket.sock
-        else:
-            sock = self.httpd.socket
-        self.port = sock.getsockname()[1]
+        self.port = http_server_port(self.httpd)
         self.server_thread = threading.Thread(target=self.httpd.serve_forever)
         self.server_thread.daemon = True
         self.server_thread.start()
@@ -94,14 +133,14 @@ class TestProxy(unittest.TestCase):
     def setUp(self):
         self.proxy = compat_http_server.HTTPServer(
             ('localhost', 0), _build_proxy_handler('normal'))
-        self.port = self.proxy.socket.getsockname()[1]
+        self.port = http_server_port(self.proxy)
         self.proxy_thread = threading.Thread(target=self.proxy.serve_forever)
         self.proxy_thread.daemon = True
         self.proxy_thread.start()
 
         self.cn_proxy = compat_http_server.HTTPServer(
             ('localhost', 0), _build_proxy_handler('cn'))
-        self.cn_port = self.cn_proxy.socket.getsockname()[1]
+        self.cn_port = http_server_port(self.cn_proxy)
         self.cn_proxy_thread = threading.Thread(target=self.cn_proxy.serve_forever)
         self.cn_proxy_thread.daemon = True
         self.cn_proxy_thread.start()