+ self.https_httpd = http.server.ThreadingHTTPServer(
+ ('127.0.0.1', 0), HTTPTestRequestHandler)
+ sslctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+ sslctx.load_cert_chain(certfn, None)
+ self.https_httpd.socket = sslctx.wrap_socket(self.https_httpd.socket, server_side=True)
+ self.https_port = http_server_port(self.https_httpd)
+ self.https_server_thread = threading.Thread(target=self.https_httpd.serve_forever)
+ self.https_server_thread.daemon = True
+ self.https_server_thread.start()
+
+ def test_nocheckcertificate(self):
+ with FakeYDL({'logger': FakeLogger()}) as ydl:
+ with self.assertRaises(urllib.error.URLError):
+ ydl.urlopen(sanitized_Request(f'https://127.0.0.1:{self.https_port}/headers'))
+
+ with FakeYDL({'logger': FakeLogger(), 'nocheckcertificate': True}) as ydl:
+ r = ydl.urlopen(sanitized_Request(f'https://127.0.0.1:{self.https_port}/headers'))
+ self.assertEqual(r.status, 200)
+ r.close()
+
+ def test_percent_encode(self):
+ with FakeYDL() as ydl:
+ # Unicode characters should be encoded with uppercase percent-encoding
+ res = ydl.urlopen(sanitized_Request(f'http://127.0.0.1:{self.http_port}/中文.html'))
+ self.assertEqual(res.status, 200)
+ res.close()
+ # don't normalize existing percent encodings
+ res = ydl.urlopen(sanitized_Request(f'http://127.0.0.1:{self.http_port}/%c7%9f'))
+ self.assertEqual(res.status, 200)
+ res.close()
+
+ def test_unicode_path_redirection(self):
+ with FakeYDL() as ydl:
+ r = ydl.urlopen(sanitized_Request(f'http://127.0.0.1:{self.http_port}/302-non-ascii-redirect'))
+ self.assertEqual(r.url, f'http://127.0.0.1:{self.http_port}/%E4%B8%AD%E6%96%87.html')
+ r.close()
+
+ def test_redirect(self):
+ with FakeYDL() as ydl:
+ def do_req(redirect_status, method):
+ data = b'testdata' if method in ('POST', 'PUT') else None
+ res = ydl.urlopen(sanitized_Request(
+ f'http://127.0.0.1:{self.http_port}/redirect_{redirect_status}', method=method, data=data))
+ return res.read().decode('utf-8'), res.headers.get('method', '')
+
+ # A 303 must either use GET or HEAD for subsequent request
+ self.assertEqual(do_req(303, 'POST'), ('', 'GET'))
+ self.assertEqual(do_req(303, 'HEAD'), ('', 'HEAD'))
+
+ self.assertEqual(do_req(303, 'PUT'), ('', 'GET'))
+
+ # 301 and 302 turn POST only into a GET
+ self.assertEqual(do_req(301, 'POST'), ('', 'GET'))
+ self.assertEqual(do_req(301, 'HEAD'), ('', 'HEAD'))
+ self.assertEqual(do_req(302, 'POST'), ('', 'GET'))
+ self.assertEqual(do_req(302, 'HEAD'), ('', 'HEAD'))
+
+ self.assertEqual(do_req(301, 'PUT'), ('testdata', 'PUT'))
+ self.assertEqual(do_req(302, 'PUT'), ('testdata', 'PUT'))
+
+ # 307 and 308 should not change method
+ for m in ('POST', 'PUT'):
+ self.assertEqual(do_req(307, m), ('testdata', m))
+ self.assertEqual(do_req(308, m), ('testdata', m))
+
+ self.assertEqual(do_req(307, 'HEAD'), ('', 'HEAD'))
+ self.assertEqual(do_req(308, 'HEAD'), ('', 'HEAD'))
+
+ # These should not redirect and instead raise an HTTPError
+ for code in (300, 304, 305, 306):
+ with self.assertRaises(urllib.error.HTTPError):
+ do_req(code, 'GET')
+
+ def test_content_type(self):
+ # https://github.com/yt-dlp/yt-dlp/commit/379a4f161d4ad3e40932dcf5aca6e6fb9715ab28
+ with FakeYDL({'nocheckcertificate': True}) as ydl:
+ # method should be auto-detected as POST
+ r = sanitized_Request(f'https://localhost:{self.https_port}/headers', data=urlencode_postdata({'test': 'test'}))
+
+ headers = ydl.urlopen(r).read().decode('utf-8')
+ self.assertIn('Content-Type: application/x-www-form-urlencoded', headers)
+
+ # test http
+ r = sanitized_Request(f'http://localhost:{self.http_port}/headers', data=urlencode_postdata({'test': 'test'}))
+ headers = ydl.urlopen(r).read().decode('utf-8')
+ self.assertIn('Content-Type: application/x-www-form-urlencoded', headers)
+
+ def test_cookiejar(self):
+ with FakeYDL() as ydl:
+ ydl.cookiejar.set_cookie(http.cookiejar.Cookie(
+ 0, 'test', 'ytdlp', None, False, '127.0.0.1', True,
+ False, '/headers', True, False, None, False, None, None, {}))
+ data = ydl.urlopen(sanitized_Request(f'http://127.0.0.1:{self.http_port}/headers')).read()
+ self.assertIn(b'Cookie: test=ytdlp', data)
+
+ def test_no_compression_compat_header(self):
+ with FakeYDL() as ydl:
+ data = ydl.urlopen(
+ sanitized_Request(
+ f'http://127.0.0.1:{self.http_port}/headers',
+ headers={'Youtubedl-no-compression': True})).read()
+ self.assertIn(b'Accept-Encoding: identity', data)
+ self.assertNotIn(b'youtubedl-no-compression', data.lower())
+
+ def test_gzip_trailing_garbage(self):
+ # https://github.com/ytdl-org/youtube-dl/commit/aa3e950764337ef9800c936f4de89b31c00dfcf5
+ # https://github.com/ytdl-org/youtube-dl/commit/6f2ec15cee79d35dba065677cad9da7491ec6e6f
+ with FakeYDL() as ydl:
+ data = ydl.urlopen(sanitized_Request(f'http://localhost:{self.http_port}/trailing_garbage')).read().decode('utf-8')
+ self.assertEqual(data, '<html><video src="/vid.mp4" /></html>')
+
+ @unittest.skipUnless(brotli, 'brotli support is not installed')
+ def test_brotli(self):
+ with FakeYDL() as ydl:
+ res = ydl.urlopen(
+ sanitized_Request(
+ f'http://127.0.0.1:{self.http_port}/content-encoding',
+ headers={'ytdl-encoding': 'br'}))
+ self.assertEqual(res.headers.get('Content-Encoding'), 'br')
+ self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
+
+ def test_deflate(self):
+ with FakeYDL() as ydl:
+ res = ydl.urlopen(
+ sanitized_Request(
+ f'http://127.0.0.1:{self.http_port}/content-encoding',
+ headers={'ytdl-encoding': 'deflate'}))
+ self.assertEqual(res.headers.get('Content-Encoding'), 'deflate')
+ self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
+
+ def test_gzip(self):
+ with FakeYDL() as ydl:
+ res = ydl.urlopen(
+ sanitized_Request(
+ f'http://127.0.0.1:{self.http_port}/content-encoding',
+ headers={'ytdl-encoding': 'gzip'}))
+ self.assertEqual(res.headers.get('Content-Encoding'), 'gzip')
+ self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
+
+ def test_multiple_encodings(self):
+ # https://www.rfc-editor.org/rfc/rfc9110.html#section-8.4
+ with FakeYDL() as ydl:
+ for pair in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
+ res = ydl.urlopen(
+ sanitized_Request(
+ f'http://127.0.0.1:{self.http_port}/content-encoding',
+ headers={'ytdl-encoding': pair}))
+ self.assertEqual(res.headers.get('Content-Encoding'), pair)
+ self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
+
+ def test_unsupported_encoding(self):
+ # it should return the raw content
+ with FakeYDL() as ydl:
+ res = ydl.urlopen(
+ sanitized_Request(
+ f'http://127.0.0.1:{self.http_port}/content-encoding',
+ headers={'ytdl-encoding': 'unsupported'}))
+ self.assertEqual(res.headers.get('Content-Encoding'), 'unsupported')
+ self.assertEqual(res.read(), b'raw')
+
+
+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)