[yt-dlp.git] / test / test_http.py

#!/usr/bin/env python
# coding: utf-8
from __future__ import unicode_literals

# Allow direct execution
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from youtube_dl import YoutubeDL
from youtube_dl.compat import compat_http_server, compat_urllib_request
import ssl
import threading

TEST_DIR = os.path.dirname(os.path.abspath(__file__))


class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass

    def do_GET(self):
        if self.path == '/video.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>')
        elif self.path == '/vid.mp4':
            self.send_response(200)
            self.send_header('Content-Type', 'video/mp4')
            self.end_headers()
            self.wfile.write(b'\x00\x00\x00\x00\x20\x66\x74[video]')
        else:
            assert False


class FakeLogger(object):
    def debug(self, msg):
        pass

    def warning(self, msg):
        pass

    def error(self, msg):
        pass


class TestHTTP(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.server_thread = threading.Thread(target=self.httpd.serve_forever)
        self.server_thread.daemon = True
        self.server_thread.start()

    def test_nocheckcertificate(self):
        if sys.version_info >= (2, 7, 9):  # No certificate checking anyways
            ydl = YoutubeDL({'logger': FakeLogger()})
            self.assertRaises(
                Exception,
                ydl.extract_info, 'https://localhost:%d/video.html' % self.port)

        ydl = YoutubeDL({'logger': FakeLogger(), 'nocheckcertificate': True})
        r = ydl.extract_info('https://localhost:%d/video.html' % self.port)
        self.assertEqual(r['url'], 'https://localhost:%d/vid.mp4' % self.port)


def _build_proxy_handler(name):
    class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
        proxy_name = name

        def log_message(self, format, *args):
            pass

        def do_GET(self):
            self.send_response(200)
            self.send_header('Content-Type', 'text/plain; charset=utf-8')
            self.end_headers()
            self.wfile.write('{self.proxy_name}: {self.path}'.format(self=self).encode('utf-8'))
    return HTTPTestRequestHandler


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.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_proxy_thread = threading.Thread(target=self.cn_proxy.serve_forever)
        self.cn_proxy_thread.daemon = True
        self.cn_proxy_thread.start()

    def test_proxy(self):
        cn_proxy = 'localhost:{0}'.format(self.cn_port)
        ydl = YoutubeDL({
            'proxy': 'localhost:{0}'.format(self.port),
            'cn_verification_proxy': cn_proxy,
        })
        url = 'http://foo.com/bar'
        response = ydl.urlopen(url).read().decode('utf-8')
        self.assertEqual(response, 'normal: {0}'.format(url))

        req = compat_urllib_request.Request(url)
        req.add_header('Ytdl-request-proxy', cn_proxy)
        response = ydl.urlopen(req).read().decode('utf-8')
        self.assertEqual(response, 'cn: {0}'.format(url))

    def test_proxy_with_idn(self):
        ydl = YoutubeDL({
            'proxy': 'localhost:{0}'.format(self.port),
        })
        url = 'http://中文.tw/'
        response = ydl.urlopen(url).read().decode('utf-8')
        # b'xn--fiq228c' is '中文'.encode('idna')
        self.assertEqual(response, 'normal: http://xn--fiq228c.tw/')

if __name__ == '__main__':
    unittest.main()
Commit	Line	Data
83fda3c0	1	#!/usr/bin/env python
efbed08d	2	# coding: utf-8
83fda3c0 PH	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
01218f91	12	from youtube_dl.compat import compat_http_server, compat_urllib_request
83fda3c0 PH	13	import ssl
	14	import threading
	15
	16	TEST_DIR = os.path.dirname(os.path.abspath(__file__))
	17
03d8d4df	18
83fda3c0 PH	19	class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
	20	def log_message(self, format, *args):
	21	pass
	22
	23	def do_GET(self):
	24	if self.path == '/video.html':
	25	self.send_response(200)
	26	self.send_header('Content-Type', 'text/html; charset=utf-8')
	27	self.end_headers()
	28	self.wfile.write(b'<html><video src="/vid.mp4" /></html>')
	29	elif self.path == '/vid.mp4':
	30	self.send_response(200)
	31	self.send_header('Content-Type', 'video/mp4')
	32	self.end_headers()
	33	self.wfile.write(b'\x00\x00\x00\x00\x20\x66\x74[video]')
	34	else:
	35	assert False
	36
	37
	38	class FakeLogger(object):
	39	def debug(self, msg):
	40	pass
	41
	42	def warning(self, msg):
	43	pass
	44
	45	def error(self, msg):
	46	pass
	47
	48
	49	class TestHTTP(unittest.TestCase):
	50	def setUp(self):
	51	certfn = os.path.join(TEST_DIR, 'testcert.pem')
	52	self.httpd = compat_http_server.HTTPServer(
	53	('localhost', 0), HTTPTestRequestHandler)
	54	self.httpd.socket = ssl.wrap_socket(
	55	self.httpd.socket, certfile=certfn, server_side=True)
429491f5 YCH	56	if os.name == 'java':
	57	# In Jython SSLSocket is not a subclass of socket.socket
	58	sock = self.httpd.socket.sock
	59	else:
	60	sock = self.httpd.socket
	61	self.port = sock.getsockname()[1]
83fda3c0 PH	62	self.server_thread = threading.Thread(target=self.httpd.serve_forever)
	63	self.server_thread.daemon = True
	64	self.server_thread.start()
	65
	66	def test_nocheckcertificate(self):
	67	if sys.version_info >= (2, 7, 9): # No certificate checking anyways
	68	ydl = YoutubeDL({'logger': FakeLogger()})
	69	self.assertRaises(
	70	Exception,
	71	ydl.extract_info, 'https://localhost:%d/video.html' % self.port)
	72
	73	ydl = YoutubeDL({'logger': FakeLogger(), 'nocheckcertificate': True})
	74	r = ydl.extract_info('https://localhost:%d/video.html' % self.port)
	75	self.assertEqual(r['url'], 'https://localhost:%d/vid.mp4' % self.port)
	76
01218f91 JMF	77
	78	def _build_proxy_handler(name):
	79	class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
	80	proxy_name = name
	81
	82	def log_message(self, format, *args):
	83	pass
	84
	85	def do_GET(self):
	86	self.send_response(200)
	87	self.send_header('Content-Type', 'text/plain; charset=utf-8')
	88	self.end_headers()
	89	self.wfile.write('{self.proxy_name}: {self.path}'.format(self=self).encode('utf-8'))
	90	return HTTPTestRequestHandler
	91
	92
	93	class TestProxy(unittest.TestCase):
	94	def setUp(self):
	95	self.proxy = compat_http_server.HTTPServer(
	96	('localhost', 0), _build_proxy_handler('normal'))
	97	self.port = self.proxy.socket.getsockname()[1]
	98	self.proxy_thread = threading.Thread(target=self.proxy.serve_forever)
	99	self.proxy_thread.daemon = True
	100	self.proxy_thread.start()
	101
	102	self.cn_proxy = compat_http_server.HTTPServer(
	103	('localhost', 0), _build_proxy_handler('cn'))
	104	self.cn_port = self.cn_proxy.socket.getsockname()[1]
	105	self.cn_proxy_thread = threading.Thread(target=self.cn_proxy.serve_forever)
	106	self.cn_proxy_thread.daemon = True
	107	self.cn_proxy_thread.start()
	108
	109	def test_proxy(self):
	110	cn_proxy = 'localhost:{0}'.format(self.cn_port)
	111	ydl = YoutubeDL({
	112	'proxy': 'localhost:{0}'.format(self.port),
	113	'cn_verification_proxy': cn_proxy,
	114	})
	115	url = 'http://foo.com/bar'
	116	response = ydl.urlopen(url).read().decode('utf-8')
	117	self.assertEqual(response, 'normal: {0}'.format(url))
	118
	119	req = compat_urllib_request.Request(url)
	120	req.add_header('Ytdl-request-proxy', cn_proxy)
	121	response = ydl.urlopen(req).read().decode('utf-8')
	122	self.assertEqual(response, 'cn: {0}'.format(url))
	123
efbed08d YCH	124	def test_proxy_with_idn(self):
	125	ydl = YoutubeDL({
	126	'proxy': 'localhost:{0}'.format(self.port),
	127	})
	128	url = 'http://中文.tw/'
	129	response = ydl.urlopen(url).read().decode('utf-8')
	130	# b'xn--fiq228c' is '中文'.encode('idna')
	131	self.assertEqual(response, 'normal: http://xn--fiq228c.tw/')
	132
83fda3c0 PH	133	if __name__ == '__main__':
83fda3c0 PH	134	unittest.main()