Raphaël G. Git Repositories - youtubedl/blob - test/test_http.py

   1 #!/usr/bin/env python
   2 # coding: utf-8
   3 from __future__ import unicode_literals
   4
   5 # Allow direct execution
   6 import os
   7 import sys
   8 import unittest
   9 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
  10
  11 from youtube_dl import YoutubeDL
  12 from youtube_dl.compat import compat_http_server, compat_urllib_request
  13 import ssl
  14 import threading
  15
  16 TEST_DIR = os.path.dirname(os.path.abspath(__file__))
  17
  18
  19 def http_server_port(httpd):
  20     if os.name == 'java' and isinstance(httpd.socket, ssl.SSLSocket):
  21         # In Jython SSLSocket is not a subclass of socket.socket
  22         sock = httpd.socket.sock
  23     else:
  24         sock = httpd.socket
  25     return sock.getsockname()[1]
  26
  27
  28 class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
  29     def log_message(self, format, *args):
  30         pass
  31
  32     def do_GET(self):
  33         if self.path == '/video.html':
  34             self.send_response(200)
  35             self.send_header('Content-Type', 'text/html; charset=utf-8')
  36             self.end_headers()
  37             self.wfile.write(b'<html><video src="/vid.mp4" /></html>')
  38         elif self.path == '/vid.mp4':
  39             self.send_response(200)
  40             self.send_header('Content-Type', 'video/mp4')
  41             self.end_headers()
  42             self.wfile.write(b'\x00\x00\x00\x00\x20\x66\x74[video]')
  43         elif self.path == '/302':
  44             if sys.version_info[0] == 3:
  45                 # XXX: Python 3 http server does not allow non-ASCII header values
  46                 self.send_response(404)
  47                 self.end_headers()
  48                 return
  49
  50             new_url = 'http://localhost:%d/中文.html' % http_server_port(self.server)
  51             self.send_response(302)
  52             self.send_header(b'Location', new_url.encode('utf-8'))
  53             self.end_headers()
  54         elif self.path == '/%E4%B8%AD%E6%96%87.html':
  55             self.send_response(200)
  56             self.send_header('Content-Type', 'text/html; charset=utf-8')
  57             self.end_headers()
  58             self.wfile.write(b'<html><video src="/vid.mp4" /></html>')
  59         else:
  60             assert False
  61
  62
  63 class FakeLogger(object):
  64     def debug(self, msg):
  65         pass
  66
  67     def warning(self, msg):
  68         pass
  69
  70     def error(self, msg):
  71         pass
  72
  73
  74 class TestHTTP(unittest.TestCase):
  75     def setUp(self):
  76         self.httpd = compat_http_server.HTTPServer(
  77             ('localhost', 0), HTTPTestRequestHandler)
  78         self.port = http_server_port(self.httpd)
  79         self.server_thread = threading.Thread(target=self.httpd.serve_forever)
  80         self.server_thread.daemon = True
  81         self.server_thread.start()
  82
  83     def test_unicode_path_redirection(self):
  84         # XXX: Python 3 http server does not allow non-ASCII header values
  85         if sys.version_info[0] == 3:
  86             return
  87
  88         ydl = YoutubeDL({'logger': FakeLogger()})
  89         r = ydl.extract_info('http://localhost:%d/302' % self.port)
  90         self.assertEqual(r['entries'][0]['url'], 'http://localhost:%d/vid.mp4' % self.port)
  91
  92
  93 class TestHTTPS(unittest.TestCase):
  94     def setUp(self):
  95         certfn = os.path.join(TEST_DIR, 'testcert.pem')
  96         self.httpd = compat_http_server.HTTPServer(
  97             ('localhost', 0), HTTPTestRequestHandler)
  98         self.httpd.socket = ssl.wrap_socket(
  99             self.httpd.socket, certfile=certfn, server_side=True)
 100         self.port = http_server_port(self.httpd)
 101         self.server_thread = threading.Thread(target=self.httpd.serve_forever)
 102         self.server_thread.daemon = True
 103         self.server_thread.start()
 104
 105     def test_nocheckcertificate(self):
 106         if sys.version_info >= (2, 7, 9):  # No certificate checking anyways
 107             ydl = YoutubeDL({'logger': FakeLogger()})
 108             self.assertRaises(
 109                 Exception,
 110                 ydl.extract_info, 'https://localhost:%d/video.html' % self.port)
 111
 112         ydl = YoutubeDL({'logger': FakeLogger(), 'nocheckcertificate': True})
 113         r = ydl.extract_info('https://localhost:%d/video.html' % self.port)
 114         self.assertEqual(r['entries'][0]['url'], 'https://localhost:%d/vid.mp4' % self.port)
 115
 116
 117 def _build_proxy_handler(name):
 118     class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
 119         proxy_name = name
 120
 121         def log_message(self, format, *args):
 122             pass
 123
 124         def do_GET(self):
 125             self.send_response(200)
 126             self.send_header('Content-Type', 'text/plain; charset=utf-8')
 127             self.end_headers()
 128             self.wfile.write('{self.proxy_name}: {self.path}'.format(self=self).encode('utf-8'))
 129     return HTTPTestRequestHandler
 130
 131
 132 class TestProxy(unittest.TestCase):
 133     def setUp(self):
 134         self.proxy = compat_http_server.HTTPServer(
 135             ('localhost', 0), _build_proxy_handler('normal'))
 136         self.port = http_server_port(self.proxy)
 137         self.proxy_thread = threading.Thread(target=self.proxy.serve_forever)
 138         self.proxy_thread.daemon = True
 139         self.proxy_thread.start()
 140
 141         self.geo_proxy = compat_http_server.HTTPServer(
 142             ('localhost', 0), _build_proxy_handler('geo'))
 143         self.geo_port = http_server_port(self.geo_proxy)
 144         self.geo_proxy_thread = threading.Thread(target=self.geo_proxy.serve_forever)
 145         self.geo_proxy_thread.daemon = True
 146         self.geo_proxy_thread.start()
 147
 148     def test_proxy(self):
 149         geo_proxy = 'localhost:{0}'.format(self.geo_port)
 150         ydl = YoutubeDL({
 151             'proxy': 'localhost:{0}'.format(self.port),
 152             'geo_verification_proxy': geo_proxy,
 153         })
 154         url = 'http://foo.com/bar'
 155         response = ydl.urlopen(url).read().decode('utf-8')
 156         self.assertEqual(response, 'normal: {0}'.format(url))
 157
 158         req = compat_urllib_request.Request(url)
 159         req.add_header('Ytdl-request-proxy', geo_proxy)
 160         response = ydl.urlopen(req).read().decode('utf-8')
 161         self.assertEqual(response, 'geo: {0}'.format(url))
 162
 163     def test_proxy_with_idn(self):
 164         ydl = YoutubeDL({
 165             'proxy': 'localhost:{0}'.format(self.port),
 166         })
 167         url = 'http://中文.tw/'
 168         response = ydl.urlopen(url).read().decode('utf-8')
 169         # b'xn--fiq228c' is '中文'.encode('idna')
 170         self.assertEqual(response, 'normal: http://xn--fiq228c.tw/')
 171
 172
 173 if __name__ == '__main__':
 174     unittest.main()