logo

youtube-dl

[mirror] Download/Watch videos from video hostersgit clone https://hacktivis.me/git/mirror/youtube-dl.git

test_http.py (23920B)

    

  1. #!/usr/bin/env python
  2. # coding: utf-8
  3. from __future__ import unicode_literals
  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__))))
  11. import contextlib
  12. import gzip
  13. import io
  14. import ssl
  15. import tempfile
  16. import threading
  17. import zlib
  19. # avoid deprecated alias assertRaisesRegexp
  20. if hasattr(unittest.TestCase, 'assertRaisesRegex'):
  21.     unittest.TestCase.assertRaisesRegexp = unittest.TestCase.assertRaisesRegex
  23. try:
  24.     import brotli
  25. except ImportError:
  26.     brotli = None
  27. try:
  28.     from urllib.request import pathname2url
  29. except ImportError:
  30.     from urllib import pathname2url
  32. from youtube_dl.compat import (
  33.     compat_http_cookiejar_Cookie,
  34.     compat_http_server,
  35.     compat_str as str,
  36.     compat_urllib_error,
  37.     compat_urllib_HTTPError,
  38.     compat_urllib_parse,
  39.     compat_urllib_request,
  40. )
  42. from youtube_dl.utils import (
  43.     sanitized_Request,
  44.     update_Request,
  45.     urlencode_postdata,
  46. )
  48. from test.helper import (
  49.     expectedFailureIf,
  50.     FakeYDL,
  51.     FakeLogger,
  52.     http_server_port,
  53. )
  54. from youtube_dl import YoutubeDL
  56. TEST_DIR = os.path.dirname(os.path.abspath(__file__))
  59. class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
  60.     protocol_version = 'HTTP/1.1'
  62.     # work-around old/new -style class inheritance
  63.     def super(self, meth_name, *args, **kwargs):
  64.         from types import MethodType
  65.         try:
  66.             super()
  67.             fn = lambda s, m, *a, **k: getattr(super(), m)(*a, **k)
  68.         except TypeError:
  69.             fn = lambda s, m, *a, **k: getattr(compat_http_server.BaseHTTPRequestHandler, m)(s, *a, **k)
  70.         self.super = MethodType(fn, self)
  71.         return self.super(meth_name, *args, **kwargs)
  73.     def log_message(self, format, *args):
  74.         pass
  76.     def _headers(self):
  77.         payload = str(self.headers).encode('utf-8')
  78.         self.send_response(200)
  79.         self.send_header('Content-Type', 'application/json')
  80.         self.send_header('Content-Length', str(len(payload)))
  81.         self.end_headers()
  82.         self.wfile.write(payload)
  84.     def _redirect(self):
  85.         self.send_response(int(self.path[len('/redirect_'):]))
  86.         self.send_header('Location', '/method')
  87.         self.send_header('Content-Length', '0')
  88.         self.end_headers()
  90.     def _method(self, method, payload=None):
  91.         self.send_response(200)
  92.         self.send_header('Content-Length', str(len(payload or '')))
  93.         self.send_header('Method', method)
  94.         self.end_headers()
  95.         if payload:
  96.             self.wfile.write(payload)
  98.     def _status(self, status):
  99.         payload = '<html>{0} NOT FOUND</html>'.format(status).encode('utf-8')
  100.         self.send_response(int(status))
  101.         self.send_header('Content-Type', 'text/html; charset=utf-8')
  102.         self.send_header('Content-Length', str(len(payload)))
  103.         self.end_headers()
  104.         self.wfile.write(payload)
  106.     def _read_data(self):
  107.         if 'Content-Length' in self.headers:
  108.             return self.rfile.read(int(self.headers['Content-Length']))
  110.     def _test_url(self, path, host='127.0.0.1', scheme='http', port=None):
  111.         return '{0}://{1}:{2}/{3}'.format(
  112.             scheme, host,
  113.             port if port is not None
  114.             else http_server_port(self.server), path)
  116.     def do_POST(self):
  117.         data = self._read_data()
  118.         if self.path.startswith('/redirect_'):
  119.             self._redirect()
  120.         elif self.path.startswith('/method'):
  121.             self._method('POST', data)
  122.         elif self.path.startswith('/headers'):
  123.             self._headers()
  124.         else:
  125.             self._status(404)
  127.     def do_HEAD(self):
  128.         if self.path.startswith('/redirect_'):
  129.             self._redirect()
  130.         elif self.path.startswith('/method'):
  131.             self._method('HEAD')
  132.         else:
  133.             self._status(404)
  135.     def do_PUT(self):
  136.         data = self._read_data()
  137.         if self.path.startswith('/redirect_'):
  138.             self._redirect()
  139.         elif self.path.startswith('/method'):
  140.             self._method('PUT', data)
  141.         else:
  142.             self._status(404)
  144.     def do_GET(self):
  146.         def respond(payload=b'<html><video src="/vid.mp4" /></html>',
  147.                     payload_type='text/html; charset=utf-8',
  148.                     payload_encoding=None,
  149.                     resp_code=200):
  150.             self.send_response(resp_code)
  151.             self.send_header('Content-Type', payload_type)
  152.             if payload_encoding:
  153.                 self.send_header('Content-Encoding', payload_encoding)
  154.             self.send_header('Content-Length', str(len(payload)))  # required for persistent connections
  155.             self.end_headers()
  156.             self.wfile.write(payload)
  158.         def gzip_compress(p):
  159.             buf = io.BytesIO()
  160.             with contextlib.closing(gzip.GzipFile(fileobj=buf, mode='wb')) as f:
  161.                 f.write(p)
  162.             return buf.getvalue()
  164.         if self.path == '/video.html':
  165.             respond()
  166.         elif self.path == '/vid.mp4':
  167.             respond(b'\x00\x00\x00\x00\x20\x66\x74[video]', 'video/mp4')
  168.         elif self.path == '/302':
  169.             if sys.version_info[0] == 3:
  170.                 # XXX: Python 3 http server does not allow non-ASCII header values
  171.                 self.send_response(404)
  172.                 self.end_headers()
  173.                 return
  175.             new_url = self._test_url('中文.html')
  176.             self.send_response(302)
  177.             self.send_header(b'Location', new_url.encode('utf-8'))
  178.             self.end_headers()
  179.         elif self.path == '/%E4%B8%AD%E6%96%87.html':
  180.             respond()
  181.         elif self.path == '/%c7%9f':
  182.             respond()
  183.         elif self.path == '/redirect_dotsegments':
  184.             self.send_response(301)
  185.             # redirect to /headers but with dot segments before
  186.             self.send_header('Location', '/a/b/./../../headers')
  187.             self.send_header('Content-Length', '0')
  188.             self.end_headers()
  189.         elif self.path.startswith('/redirect_'):
  190.             self._redirect()
  191.         elif self.path.startswith('/method'):
  192.             self._method('GET')
  193.         elif self.path.startswith('/headers'):
  194.             self._headers()
  195.         elif self.path.startswith('/308-to-headers'):
  196.             self.send_response(308)
  197.             self.send_header('Location', '/headers')
  198.             self.send_header('Content-Length', '0')
  199.             self.end_headers()
  200.         elif self.path == '/trailing_garbage':
  201.             payload = b'<html><video src="/vid.mp4" /></html>'
  202.             compressed = gzip_compress(payload) + b'trailing garbage'
  203.             respond(compressed, payload_encoding='gzip')
  204.         elif self.path == '/302-non-ascii-redirect':
  205.             new_url = self._test_url('中文.html')
  206.             # actually respond with permanent redirect
  207.             self.send_response(301)
  208.             self.send_header('Location', new_url)
  209.             self.send_header('Content-Length', '0')
  210.             self.end_headers()
  211.         elif self.path == '/content-encoding':
  212.             encodings = self.headers.get('ytdl-encoding', '')
  213.             payload = b'<html><video src="/vid.mp4" /></html>'
  214.             for encoding in filter(None, (e.strip() for e in encodings.split(','))):
  215.                 if encoding == 'br' and brotli:
  216.                     payload = brotli.compress(payload)
  217.                 elif encoding == 'gzip':
  218.                     payload = gzip_compress(payload)
  219.                 elif encoding == 'deflate':
  220.                     payload = zlib.compress(payload)
  221.                 elif encoding == 'unsupported':
  222.                     payload = b'raw'
  223.                     break
  224.                 else:
  225.                     self._status(415)
  226.                     return
  227.             respond(payload, payload_encoding=encodings)
  229.         else:
  230.             self._status(404)
  232.     def send_header(self, keyword, value):
  233.         """
  234.         Forcibly allow HTTP server to send non percent-encoded non-ASCII characters in headers.
  235.         This is against what is defined in RFC 3986: but we need to test that we support this
  236.         since some sites incorrectly do this.
  237.         """
  238.         if keyword.lower() == 'connection':
  239.             return self.super('send_header', keyword, value)
  241.         if not hasattr(self, '_headers_buffer'):
  242.             self._headers_buffer = []
  244.         self._headers_buffer.append('{0}: {1}\r\n'.format(keyword, value).encode('utf-8'))
  246.     def end_headers(self):
  247.         if hasattr(self, '_headers_buffer'):
  248.             self.wfile.write(b''.join(self._headers_buffer))
  249.             self._headers_buffer = []
  250.         self.super('end_headers')
  253. class TestHTTP(unittest.TestCase):
  254.     # when does it make sense to check the SSL certificate?
  255.     _check_cert = (
  256.         sys.version_info >= (3, 2)
  257.         or (sys.version_info[0] == 2 and sys.version_info[1:] >= (7, 19)))
  259.     def setUp(self):
  260.         # HTTP server
  261.         self.http_httpd = compat_http_server.HTTPServer(
  262.             ('127.0.0.1', 0), HTTPTestRequestHandler)
  263.         self.http_port = http_server_port(self.http_httpd)
  265.         self.http_server_thread = threading.Thread(target=self.http_httpd.serve_forever)
  266.         self.http_server_thread.daemon = True
  267.         self.http_server_thread.start()
  269.         try:
  270.             from http.server import ThreadingHTTPServer
  271.         except ImportError:
  272.             try:
  273.                 from socketserver import ThreadingMixIn
  274.             except ImportError:
  275.                 from SocketServer import ThreadingMixIn
  277.             class ThreadingHTTPServer(ThreadingMixIn, compat_http_server.HTTPServer):
  278.                 pass
  280.         # HTTPS server
  281.         certfn = os.path.join(TEST_DIR, 'testcert.pem')
  282.         self.https_httpd = ThreadingHTTPServer(
  283.             ('127.0.0.1', 0), HTTPTestRequestHandler)
  284.         try:
  285.             sslctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
  286.             sslctx.verify_mode = ssl.CERT_NONE
  287.             sslctx.check_hostname = False
  288.             sslctx.load_cert_chain(certfn, None)
  289.             self.https_httpd.socket = sslctx.wrap_socket(
  290.                 self.https_httpd.socket, server_side=True)
  291.         except AttributeError:
  292.             self.https_httpd.socket = ssl.wrap_socket(
  293.                 self.https_httpd.socket, certfile=certfn, server_side=True)
  295.         self.https_port = http_server_port(self.https_httpd)
  296.         self.https_server_thread = threading.Thread(target=self.https_httpd.serve_forever)
  297.         self.https_server_thread.daemon = True
  298.         self.https_server_thread.start()
  300.     def tearDown(self):
  302.         def closer(svr):
  303.             def _closer():
  304.                 svr.shutdown()
  305.                 svr.server_close()
  306.             return _closer
  308.         shutdown_thread = threading.Thread(target=closer(self.http_httpd))
  309.         shutdown_thread.start()
  310.         self.http_server_thread.join(2.0)
  312.         shutdown_thread = threading.Thread(target=closer(self.https_httpd))
  313.         shutdown_thread.start()
  314.         self.https_server_thread.join(2.0)
  316.     def _test_url(self, path, host='127.0.0.1', scheme='http', port=None):
  317.         return '{0}://{1}:{2}/{3}'.format(
  318.             scheme, host,
  319.             port if port is not None
  320.             else self.https_port if scheme == 'https'
  321.             else self.http_port, path)
  323.     @unittest.skipUnless(_check_cert, 'No support for certificate check in SSL')
  324.     def test_nocheckcertificate(self):
  325.         with FakeYDL({'logger': FakeLogger()}) as ydl:
  326.             with self.assertRaises(compat_urllib_error.URLError):
  327.                 ydl.urlopen(sanitized_Request(self._test_url('headers', scheme='https')))
  329.         with FakeYDL({'logger': FakeLogger(), 'nocheckcertificate': True}) as ydl:
  330.             r = ydl.urlopen(sanitized_Request(self._test_url('headers', scheme='https')))
  331.             self.assertEqual(r.getcode(), 200)
  332.             r.close()
  334.     def test_percent_encode(self):
  335.         with FakeYDL() as ydl:
  336.             # Unicode characters should be encoded with uppercase percent-encoding
  337.             res = ydl.urlopen(sanitized_Request(self._test_url('中文.html')))
  338.             self.assertEqual(res.getcode(), 200)
  339.             res.close()
  340.             # don't normalize existing percent encodings
  341.             res = ydl.urlopen(sanitized_Request(self._test_url('%c7%9f')))
  342.             self.assertEqual(res.getcode(), 200)
  343.             res.close()
  345.     def test_unicode_path_redirection(self):
  346.         with FakeYDL() as ydl:
  347.             r = ydl.urlopen(sanitized_Request(self._test_url('302-non-ascii-redirect')))
  348.             self.assertEqual(r.url, self._test_url('%E4%B8%AD%E6%96%87.html'))
  349.             r.close()
  351.     def test_redirect(self):
  352.         with FakeYDL() as ydl:
  353.             def do_req(redirect_status, method, check_no_content=False):
  354.                 data = b'testdata' if method in ('POST', 'PUT') else None
  355.                 res = ydl.urlopen(sanitized_Request(
  356.                     self._test_url('redirect_{0}'.format(redirect_status)),
  357.                     method=method, data=data))
  358.                 if check_no_content:
  359.                     self.assertNotIn('Content-Type', res.headers)
  360.                 return res.read().decode('utf-8'), res.headers.get('method', '')
  361.             # A 303 must either use GET or HEAD for subsequent request
  362.             self.assertEqual(do_req(303, 'POST'), ('', 'GET'))
  363.             self.assertEqual(do_req(303, 'HEAD'), ('', 'HEAD'))
  365.             self.assertEqual(do_req(303, 'PUT'), ('', 'GET'))
  367.             # 301 and 302 turn POST only into a GET, with no Content-Type
  368.             self.assertEqual(do_req(301, 'POST', True), ('', 'GET'))
  369.             self.assertEqual(do_req(301, 'HEAD'), ('', 'HEAD'))
  370.             self.assertEqual(do_req(302, 'POST', True), ('', 'GET'))
  371.             self.assertEqual(do_req(302, 'HEAD'), ('', 'HEAD'))
  373.             self.assertEqual(do_req(301, 'PUT'), ('testdata', 'PUT'))
  374.             self.assertEqual(do_req(302, 'PUT'), ('testdata', 'PUT'))
  376.             # 307 and 308 should not change method
  377.             for m in ('POST', 'PUT'):
  378.                 self.assertEqual(do_req(307, m), ('testdata', m))
  379.                 self.assertEqual(do_req(308, m), ('testdata', m))
  381.             self.assertEqual(do_req(307, 'HEAD'), ('', 'HEAD'))
  382.             self.assertEqual(do_req(308, 'HEAD'), ('', 'HEAD'))
  384.             # These should not redirect and instead raise an HTTPError
  385.             for code in (300, 304, 305, 306):
  386.                 with self.assertRaises(compat_urllib_HTTPError):
  387.                     do_req(code, 'GET')
  389.     # Jython 2.7.1 times out for some reason
  390.     @expectedFailureIf(sys.platform.startswith('java') and sys.version_info < (2, 7, 2))
  391.     def test_content_type(self):
  392.         # https://github.com/yt-dlp/yt-dlp/commit/379a4f161d4ad3e40932dcf5aca6e6fb9715ab28
  393.         with FakeYDL({'nocheckcertificate': True}) as ydl:
  394.             # method should be auto-detected as POST
  395.             r = sanitized_Request(self._test_url('headers', scheme='https'), data=urlencode_postdata({'test': 'test'}))
  397.             headers = ydl.urlopen(r).read().decode('utf-8')
  398.             self.assertIn('Content-Type: application/x-www-form-urlencoded', headers)
  400.             # test http
  401.             r = sanitized_Request(self._test_url('headers'), data=urlencode_postdata({'test': 'test'}))
  402.             headers = ydl.urlopen(r).read().decode('utf-8')
  403.             self.assertIn('Content-Type: application/x-www-form-urlencoded', headers)
  405.     def test_update_req(self):
  406.         req = sanitized_Request('http://example.com')
  407.         assert req.data is None
  408.         assert req.get_method() == 'GET'
  409.         assert not req.has_header('Content-Type')
  410.         # Test that zero-byte payloads will be sent
  411.         req = update_Request(req, data=b'')
  412.         assert req.data == b''
  413.         assert req.get_method() == 'POST'
  414.         # yt-dl expects data to be encoded and Content-Type to be added by sender
  415.         # assert req.get_header('Content-Type') == 'application/x-www-form-urlencoded'
  417.     def test_cookiejar(self):
  418.         with FakeYDL() as ydl:
  419.             ydl.cookiejar.set_cookie(compat_http_cookiejar_Cookie(
  420.                 0, 'test', 'ytdl', None, False, '127.0.0.1', True,
  421.                 False, '/headers', True, False, None, False, None, None, {}))
  422.             data = ydl.urlopen(sanitized_Request(
  423.                 self._test_url('headers'))).read().decode('utf-8')
  424.             self.assertIn('Cookie: test=ytdl', data)
  426.     def test_passed_cookie_header(self):
  427.         # We should accept a Cookie header being passed as in normal headers and handle it appropriately.
  428.         with FakeYDL() as ydl:
  429.             # Specified Cookie header should be used
  430.             res = ydl.urlopen(sanitized_Request(
  431.                 self._test_url('headers'), headers={'Cookie': 'test=test'})).read().decode('utf-8')
  432.             self.assertIn('Cookie: test=test', res)
  434.             # Specified Cookie header should be removed on any redirect
  435.             res = ydl.urlopen(sanitized_Request(
  436.                 self._test_url('308-to-headers'), headers={'Cookie': 'test=test'})).read().decode('utf-8')
  437.             self.assertNotIn('Cookie: test=test', res)
  439.             # Specified Cookie header should override global cookiejar for that request
  440.             ydl.cookiejar.set_cookie(compat_http_cookiejar_Cookie(
  441.                 0, 'test', 'ytdlp', None, False, '127.0.0.1', True,
  442.                 False, '/headers', True, False, None, False, None, None, {}))
  443.             data = ydl.urlopen(sanitized_Request(
  444.                 self._test_url('headers'), headers={'Cookie': 'test=test'})).read().decode('utf-8')
  445.             self.assertNotIn('Cookie: test=ytdlp', data)
  446.             self.assertIn('Cookie: test=test', data)
  448.     def test_no_compression_compat_header(self):
  449.         with FakeYDL() as ydl:
  450.             data = ydl.urlopen(
  451.                 sanitized_Request(
  452.                     self._test_url('headers'),
  453.                     headers={'Youtubedl-no-compression': True})).read()
  454.             self.assertIn(b'Accept-Encoding: identity', data)
  455.             self.assertNotIn(b'youtubedl-no-compression', data.lower())
  457.     def test_gzip_trailing_garbage(self):
  458.         # https://github.com/ytdl-org/youtube-dl/commit/aa3e950764337ef9800c936f4de89b31c00dfcf5
  459.         # https://github.com/ytdl-org/youtube-dl/commit/6f2ec15cee79d35dba065677cad9da7491ec6e6f
  460.         with FakeYDL() as ydl:
  461.             data = ydl.urlopen(sanitized_Request(self._test_url('trailing_garbage'))).read().decode('utf-8')
  462.             self.assertEqual(data, '<html><video src="/vid.mp4" /></html>')
  464.     def __test_compression(self, encoding):
  465.         with FakeYDL() as ydl:
  466.             res = ydl.urlopen(
  467.                 sanitized_Request(
  468.                     self._test_url('content-encoding'),
  469.                     headers={'ytdl-encoding': encoding}))
  470.             # decoded encodings are removed: only check for valid decompressed data
  471.             self.assertEqual(res.read(), b'<html><video src="/vid.mp4" /></html>')
  473.     @unittest.skipUnless(brotli, 'brotli support is not installed')
  474.     def test_brotli(self):
  475.         self.__test_compression('br')
  477.     def test_deflate(self):
  478.         self.__test_compression('deflate')
  480.     def test_gzip(self):
  481.         self.__test_compression('gzip')
  483.     def test_multiple_encodings(self):
  484.         # https://www.rfc-editor.org/rfc/rfc9110.html#section-8.4
  485.         for pair in ('gzip,deflate', 'deflate, gzip', 'gzip, gzip', 'deflate, deflate'):
  486.             self.__test_compression(pair)
  488.     def test_unsupported_encoding(self):
  489.         # it should return the raw content
  490.         with FakeYDL() as ydl:
  491.             res = ydl.urlopen(
  492.                 sanitized_Request(
  493.                     self._test_url('content-encoding'),
  494.                     headers={'ytdl-encoding': 'unsupported'}))
  495.             self.assertEqual(res.headers.get('Content-Encoding'), 'unsupported')
  496.             self.assertEqual(res.read(), b'raw')
  498.     def test_remove_dot_segments(self):
  499.         with FakeYDL() as ydl:
  500.             res = ydl.urlopen(sanitized_Request(self._test_url('a/b/./../../headers')))
  501.             self.assertEqual(compat_urllib_parse.urlparse(res.geturl()).path, '/headers')
  503.             res = ydl.urlopen(sanitized_Request(self._test_url('redirect_dotsegments')))
  504.             self.assertEqual(compat_urllib_parse.urlparse(res.geturl()).path, '/headers')
  507. def _build_proxy_handler(name):
  508.     class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler):
  509.         proxy_name = name
  511.         def log_message(self, format, *args):
  512.             pass
  514.         def do_GET(self):
  515.             self.send_response(200)
  516.             self.send_header('Content-Type', 'text/plain; charset=utf-8')
  517.             self.end_headers()
  518.             self.wfile.write('{0}: {1}'.format(self.proxy_name, self.path).encode('utf-8'))
  519.     return HTTPTestRequestHandler
  522. class TestProxy(unittest.TestCase):
  523.     def setUp(self):
  524.         self.proxy = compat_http_server.HTTPServer(
  525.             ('127.0.0.1', 0), _build_proxy_handler('normal'))
  526.         self.port = http_server_port(self.proxy)
  527.         self.proxy_thread = threading.Thread(target=self.proxy.serve_forever)
  528.         self.proxy_thread.daemon = True
  529.         self.proxy_thread.start()
  531.         self.geo_proxy = compat_http_server.HTTPServer(
  532.             ('127.0.0.1', 0), _build_proxy_handler('geo'))
  533.         self.geo_port = http_server_port(self.geo_proxy)
  534.         self.geo_proxy_thread = threading.Thread(target=self.geo_proxy.serve_forever)
  535.         self.geo_proxy_thread.daemon = True
  536.         self.geo_proxy_thread.start()
  538.     def tearDown(self):
  540.         def closer(svr):
  541.             def _closer():
  542.                 svr.shutdown()
  543.                 svr.server_close()
  544.             return _closer
  546.         shutdown_thread = threading.Thread(target=closer(self.proxy))
  547.         shutdown_thread.start()
  548.         self.proxy_thread.join(2.0)
  550.         shutdown_thread = threading.Thread(target=closer(self.geo_proxy))
  551.         shutdown_thread.start()
  552.         self.geo_proxy_thread.join(2.0)
  554.     def _test_proxy(self, host='127.0.0.1', port=None):
  555.         return '{0}:{1}'.format(
  556.             host, port if port is not None else self.port)
  558.     def test_proxy(self):
  559.         geo_proxy = self._test_proxy(port=self.geo_port)
  560.         ydl = YoutubeDL({
  561.             'proxy': self._test_proxy(),
  562.             'geo_verification_proxy': geo_proxy,
  563.         })
  564.         url = 'http://foo.com/bar'
  565.         response = ydl.urlopen(url).read().decode('utf-8')
  566.         self.assertEqual(response, 'normal: {0}'.format(url))
  568.         req = compat_urllib_request.Request(url)
  569.         req.add_header('Ytdl-request-proxy', geo_proxy)
  570.         response = ydl.urlopen(req).read().decode('utf-8')
  571.         self.assertEqual(response, 'geo: {0}'.format(url))
  573.     def test_proxy_with_idn(self):
  574.         ydl = YoutubeDL({
  575.             'proxy': self._test_proxy(),
  576.         })
  577.         url = 'http://中文.tw/'
  578.         response = ydl.urlopen(url).read().decode('utf-8')
  579.         # b'xn--fiq228c' is '中文'.encode('idna')
  580.         self.assertEqual(response, 'normal: http://xn--fiq228c.tw/')
  583. class TestFileURL(unittest.TestCase):
  584.     # See https://github.com/ytdl-org/youtube-dl/issues/8227
  585.     def test_file_urls(self):
  586.         tf = tempfile.NamedTemporaryFile(delete=False)
  587.         tf.write(b'foobar')
  588.         tf.close()
  589.         url = compat_urllib_parse.urljoin('file://', pathname2url(tf.name))
  590.         with FakeYDL() as ydl:
  591.             self.assertRaisesRegexp(
  592.                 compat_urllib_error.URLError, 'file:// scheme is explicitly disabled in youtube-dl for security reasons', ydl.urlopen, url)
  593.         # not yet implemented
  594.         """
  595.         with FakeYDL({'enable_file_urls': True}) as ydl:
  596.             res = ydl.urlopen(url)
  597.             self.assertEqual(res.read(), b'foobar')
  598.             res.close()
  599.         """
  600.         os.unlink(tf.name)
  603. if __name__ == '__main__':
  604.     unittest.main()