]> git.bitcoin.ninja Git - youtube-dl/blob - test/test_http.py
[vk:wallpost] Add extractor
[youtube-dl] / 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['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['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 if __name__ == '__main__':
173     unittest.main()