diff options
Diffstat (limited to 'test/test_http.py')
| -rw-r--r-- | test/test_http.py | 126 |
1 files changed, 76 insertions, 50 deletions
diff --git a/test/test_http.py b/test/test_http.py index a7656b0..71d6f1b 100644 --- a/test/test_http.py +++ b/test/test_http.py @@ -1,23 +1,25 @@ #!/usr/bin/env python3 -# 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 test.helper import http_server_port -from hypervideo_dl import YoutubeDL -from hypervideo_dl.compat import compat_http_server, compat_urllib_request + +import http.server import ssl import threading +import urllib.request + +from test.helper import http_server_port +from hypervideo_dl import YoutubeDL TEST_DIR = os.path.dirname(os.path.abspath(__file__)) -class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler): +class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler): def log_message(self, format, *args): pass @@ -32,17 +34,6 @@ 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://127.0.0.1:%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') @@ -52,7 +43,7 @@ class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler): assert False -class FakeLogger(object): +class FakeLogger: def debug(self, msg): pass @@ -65,49 +56,84 @@ class FakeLogger(object): class TestHTTP(unittest.TestCase): def setUp(self): - self.httpd = compat_http_server.HTTPServer( + self.httpd = 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() - 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://127.0.0.1:%d/302' % self.port) - self.assertEqual(r['entries'][0]['url'], 'http://127.0.0.1:%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( + self.httpd = http.server.HTTPServer( ('127.0.0.1', 0), HTTPTestRequestHandler) - self.httpd.socket = ssl.wrap_socket( - self.httpd.socket, certfile=certfn, server_side=True) + 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): - if sys.version_info >= (2, 7, 9): # No certificate checking anyways - ydl = YoutubeDL({'logger': FakeLogger()}) - self.assertRaises( - Exception, - ydl.extract_info, 'https://127.0.0.1:%d/video.html' % self.port) + 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) + self.assertEqual(r['url'], 'https://127.0.0.1:%d/vid.mp4' % self.port) + + +class TestClientCert(unittest.TestCase): + def setUp(self): + certfn = os.path.join(TEST_DIR, 'testcert.pem') + self.certdir = os.path.join(TEST_DIR, 'testdata', 'certificate') + cacertfn = os.path.join(self.certdir, 'ca.crt') + self.httpd = http.server.HTTPServer(('127.0.0.1', 0), HTTPTestRequestHandler) + sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + sslctx.verify_mode = ssl.CERT_REQUIRED + sslctx.load_verify_locations(cafile=cacertfn) + 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 _run_test(self, **params): + ydl = YoutubeDL({ + 'logger': FakeLogger(), + # Disable client-side validation of unacceptable self-signed testcert.pem + # The test is of a check on the server side, so unaffected + 'nocheckcertificate': True, + **params, + }) + r = ydl.extract_info('https://127.0.0.1:%d/video.html' % self.port) + self.assertEqual(r['url'], 'https://127.0.0.1:%d/vid.mp4' % self.port) + + def test_certificate_combined_nopass(self): + self._run_test(client_certificate=os.path.join(self.certdir, 'clientwithkey.crt')) + + def test_certificate_nocombined_nopass(self): + self._run_test(client_certificate=os.path.join(self.certdir, 'client.crt'), + client_certificate_key=os.path.join(self.certdir, 'client.key')) + + def test_certificate_combined_pass(self): + self._run_test(client_certificate=os.path.join(self.certdir, 'clientwithencryptedkey.crt'), + client_certificate_password='foobar') + + def test_certificate_nocombined_pass(self): + self._run_test(client_certificate=os.path.join(self.certdir, 'client.crt'), + client_certificate_key=os.path.join(self.certdir, 'clientencrypted.key'), + client_certificate_password='foobar') def _build_proxy_handler(name): - class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler): + class HTTPTestRequestHandler(http.server.BaseHTTPRequestHandler): proxy_name = name def log_message(self, format, *args): @@ -117,20 +143,20 @@ def _build_proxy_handler(name): 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')) + self.wfile.write(f'{self.proxy_name}: {self.path}'.encode()) return HTTPTestRequestHandler class TestProxy(unittest.TestCase): def setUp(self): - self.proxy = compat_http_server.HTTPServer( + self.proxy = 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( + self.geo_proxy = 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) @@ -138,26 +164,26 @@ class TestProxy(unittest.TestCase): self.geo_proxy_thread.start() def test_proxy(self): - geo_proxy = '127.0.0.1:{0}'.format(self.geo_port) + geo_proxy = f'127.0.0.1:{self.geo_port}' ydl = YoutubeDL({ - 'proxy': '127.0.0.1:{0}'.format(self.port), + '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, 'normal: {0}'.format(url)) + response = ydl.urlopen(url).read().decode() + self.assertEqual(response, f'normal: {url}') - req = compat_urllib_request.Request(url) + req = urllib.request.Request(url) req.add_header('Ytdl-request-proxy', geo_proxy) - response = ydl.urlopen(req).read().decode('utf-8') - self.assertEqual(response, 'geo: {0}'.format(url)) + response = ydl.urlopen(req).read().decode() + self.assertEqual(response, f'geo: {url}') def test_proxy_with_idn(self): ydl = YoutubeDL({ - 'proxy': '127.0.0.1:{0}'.format(self.port), + 'proxy': f'127.0.0.1:{self.port}', }) url = 'http://中文.tw/' - response = ydl.urlopen(url).read().decode('utf-8') + response = ydl.urlopen(url).read().decode() # b'xn--fiq228c' is '中文'.encode('idna') self.assertEqual(response, 'normal: http://xn--fiq228c.tw/') |