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

#!/usr/bin/env python3
# Allow direct execution
import os
import sys
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import ssl
import threading
from test.helper import http_server_port

from yt_dlp import YoutubeDL
from yt_dlp.compat import compat_http_server, compat_urllib_request

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]')
        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


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

    def warning(self, msg):
        pass

    def error(self, msg):
        pass


class TestHTTP(unittest.TestCase):
    def setUp(self):
        self.httpd = compat_http_server.HTTPServer(
            ('127.0.0.1', 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()


class TestHTTPS(unittest.TestCase):
    def setUp(self):
        certfn = os.path.join(TEST_DIR, 'testcert.pem')
        self.httpd = compat_http_server.HTTPServer(
            ('127.0.0.1', 0), HTTPTestRequestHandler)
        sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        sslctx.load_cert_chain(certfn, None)
        self.httpd.socket = sslctx.wrap_socket(self.httpd.socket, server_side=True)
        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_nocheckcertificate(self):
        ydl = YoutubeDL({'logger': FakeLogger()})
        self.assertRaises(
            Exception,
            ydl.extract_info, 'https://127.0.0.1:%d/video.html' % self.port)

        ydl = YoutubeDL({'logger': FakeLogger(), 'nocheckcertificate': True})
        r = ydl.extract_info('https://127.0.0.1:%d/video.html' % self.port)
        self.assertEqual(r['entries'][0]['url'], 'https://127.0.0.1:%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(
            ('127.0.0.1', 0), _build_proxy_handler('normal'))
        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.geo_proxy = compat_http_server.HTTPServer(
            ('127.0.0.1', 0), _build_proxy_handler('geo'))
        self.geo_port = http_server_port(self.geo_proxy)
        self.geo_proxy_thread = threading.Thread(target=self.geo_proxy.serve_forever)
        self.geo_proxy_thread.daemon = True
        self.geo_proxy_thread.start()

    def test_proxy(self):
        geo_proxy = f'127.0.0.1:{self.geo_port}'
        ydl = YoutubeDL({
            'proxy': f'127.0.0.1:{self.port}',
            'geo_verification_proxy': geo_proxy,
        })
        url = 'http://foo.com/bar'
        response = ydl.urlopen(url).read().decode('utf-8')
        self.assertEqual(response, f'normal: {url}')

        req = compat_urllib_request.Request(url)
        req.add_header('Ytdl-request-proxy', geo_proxy)
        response = ydl.urlopen(req).read().decode('utf-8')
        self.assertEqual(response, f'geo: {url}')

    def test_proxy_with_idn(self):
        ydl = YoutubeDL({
            'proxy': f'127.0.0.1:{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
cc52de43	1	#!/usr/bin/env python3
83fda3c0 PH	2	# Allow direct execution
	3	import os
	4	import sys
	5	import unittest
f8271158	6
83fda3c0 PH	7	sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
83fda3c0 PH	8
f8271158	9	import ssl
f8271158	10	import threading
95e42d73	11	from test.helper import http_server_port
f8271158	12
7a5c1cfe P	13	from yt_dlp import YoutubeDL
7a5c1cfe P	14	from yt_dlp.compat import compat_http_server, compat_urllib_request
83fda3c0 PH	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]')
8c32e5dc YCH	34	elif self.path == '/%E4%B8%AD%E6%96%87.html':
	35	self.send_response(200)
	36	self.send_header('Content-Type', 'text/html; charset=utf-8')
	37	self.end_headers()
	38	self.wfile.write(b'<html><video src="/vid.mp4" /></html>')
83fda3c0 PH	39	else:
	40	assert False
	41
	42
86e5f3ed	43	class FakeLogger:
83fda3c0 PH	44	def debug(self, msg):
	45	pass
	46
	47	def warning(self, msg):
	48	pass
	49
	50	def error(self, msg):
	51	pass
	52
	53
	54	class TestHTTP(unittest.TestCase):
8c32e5dc YCH	55	def setUp(self):
8c32e5dc YCH	56	self.httpd = compat_http_server.HTTPServer(
f19eae42	57	('127.0.0.1', 0), HTTPTestRequestHandler)
8c32e5dc YCH	58	self.port = http_server_port(self.httpd)
	59	self.server_thread = threading.Thread(target=self.httpd.serve_forever)
	60	self.server_thread.daemon = True
	61	self.server_thread.start()
	62
8c32e5dc YCH	63
8c32e5dc YCH	64	class TestHTTPS(unittest.TestCase):
83fda3c0 PH	65	def setUp(self):
	66	certfn = os.path.join(TEST_DIR, 'testcert.pem')
	67	self.httpd = compat_http_server.HTTPServer(
f19eae42	68	('127.0.0.1', 0), HTTPTestRequestHandler)
b6dc37fe	69	sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
	70	sslctx.load_cert_chain(certfn, None)
	71	self.httpd.socket = sslctx.wrap_socket(self.httpd.socket, server_side=True)
8c32e5dc	72	self.port = http_server_port(self.httpd)
83fda3c0 PH	73	self.server_thread = threading.Thread(target=self.httpd.serve_forever)
	74	self.server_thread.daemon = True
	75	self.server_thread.start()
	76
	77	def test_nocheckcertificate(self):
cfb0511d	78	ydl = YoutubeDL({'logger': FakeLogger()})
	79	self.assertRaises(
	80	Exception,
	81	ydl.extract_info, 'https://127.0.0.1:%d/video.html' % self.port)
83fda3c0 PH	82
83fda3c0 PH	83	ydl = YoutubeDL({'logger': FakeLogger(), 'nocheckcertificate': True})
f19eae42 S	84	r = ydl.extract_info('https://127.0.0.1:%d/video.html' % self.port)
f19eae42 S	85	self.assertEqual(r['entries'][0]['url'], 'https://127.0.0.1:%d/vid.mp4' % self.port)
83fda3c0	86
01218f91 JMF	87
	88	def _build_proxy_handler(name):
	89	class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
	90	proxy_name = name
	91
	92	def log_message(self, format, *args):
	93	pass
	94
	95	def do_GET(self):
	96	self.send_response(200)
	97	self.send_header('Content-Type', 'text/plain; charset=utf-8')
	98	self.end_headers()
	99	self.wfile.write('{self.proxy_name}: {self.path}'.format(self=self).encode('utf-8'))
	100	return HTTPTestRequestHandler
	101
	102
	103	class TestProxy(unittest.TestCase):
	104	def setUp(self):
	105	self.proxy = compat_http_server.HTTPServer(
f19eae42	106	('127.0.0.1', 0), _build_proxy_handler('normal'))
8c32e5dc	107	self.port = http_server_port(self.proxy)
01218f91 JMF	108	self.proxy_thread = threading.Thread(target=self.proxy.serve_forever)
	109	self.proxy_thread.daemon = True
	110	self.proxy_thread.start()
	111
40f3666f	112	self.geo_proxy = compat_http_server.HTTPServer(
f19eae42	113	('127.0.0.1', 0), _build_proxy_handler('geo'))
40f3666f YCH	114	self.geo_port = http_server_port(self.geo_proxy)
	115	self.geo_proxy_thread = threading.Thread(target=self.geo_proxy.serve_forever)
	116	self.geo_proxy_thread.daemon = True
	117	self.geo_proxy_thread.start()
01218f91 JMF	118
01218f91 JMF	119	def test_proxy(self):
86e5f3ed	120	geo_proxy = f'127.0.0.1:{self.geo_port}'
01218f91	121	ydl = YoutubeDL({
86e5f3ed	122	'proxy': f'127.0.0.1:{self.port}',
40f3666f	123	'geo_verification_proxy': geo_proxy,
01218f91 JMF	124	})
	125	url = 'http://foo.com/bar'
	126	response = ydl.urlopen(url).read().decode('utf-8')
86e5f3ed	127	self.assertEqual(response, f'normal: {url}')
01218f91 JMF	128
01218f91 JMF	129	req = compat_urllib_request.Request(url)
40f3666f	130	req.add_header('Ytdl-request-proxy', geo_proxy)
01218f91	131	response = ydl.urlopen(req).read().decode('utf-8')
86e5f3ed	132	self.assertEqual(response, f'geo: {url}')
01218f91	133
efbed08d YCH	134	def test_proxy_with_idn(self):
efbed08d YCH	135	ydl = YoutubeDL({
86e5f3ed	136	'proxy': f'127.0.0.1:{self.port}',
efbed08d YCH	137	})
	138	url = 'http://中文.tw/'
	139	response = ydl.urlopen(url).read().decode('utf-8')
	140	# b'xn--fiq228c' is '中文'.encode('idna')
	141	self.assertEqual(response, 'normal: http://xn--fiq228c.tw/')
	142
582be358	143
83fda3c0 PH	144	if __name__ == '__main__':
83fda3c0 PH	145	unittest.main()