#!/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__)))) import contextlib import gzip import io import ssl import tempfile import threading import zlib # avoid deprecated alias assertRaisesRegexp if hasattr(unittest.TestCase, 'assertRaisesRegex'): unittest.TestCase.assertRaisesRegexp = unittest.TestCase.assertRaisesRegex try: import brotli except ImportError: brotli = None try: from urllib.request import pathname2url except ImportError: from urllib import pathname2url from youtube_dl.compat import ( compat_http_cookiejar_Cookie, compat_http_server, compat_str as str, compat_urllib_error, compat_urllib_HTTPError, compat_urllib_parse, compat_urllib_request, ) from youtube_dl.utils import ( sanitized_Request, urlencode_postdata, ) from test.helper import ( expectedFailureIf, FakeYDL, FakeLogger, http_server_port, ) from youtube_dl import YoutubeDL TEST_DIR = os.path.dirname(os.path.abspath(__file__)) class HTTPTestRequestHandler(compat_http_server.BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' # work-around old/new -style class inheritance def super(self, meth_name, *args, **kwargs): from types import MethodType try: super() fn = lambda s, m, *a, **k: getattr(super(), m)(*a, **k) except TypeError: fn = lambda s, m, *a, **k: getattr(compat_http_server.BaseHTTPRequestHandler, m)(s, *a, **k) self.super = MethodType(fn, self) return self.super(meth_name, *args, **kwargs) def log_message(self, format, *args): pass def _headers(self): payload = str(self.headers).encode('utf-8') self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(payload))) self.end_headers() self.wfile.write(payload) def _redirect(self): self.send_response(int(self.path[len('/redirect_'):])) self.send_header('Location', '/method') self.send_header('Content-Length', '0') self.end_headers() def _method(self, method, payload=None): self.send_response(200) self.send_header('Content-Length', str(len(payload or ''))) self.send_header('Method', method) self.end_headers() if payload: self.wfile.write(payload) def _status(self, status): payload = '{0} NOT FOUND'.format(status).encode('utf-8') self.send_response(int(status)) self.send_header('Content-Type', 'text/html; charset=utf-8') self.send_header('Content-Length', str(len(payload))) self.end_headers() self.wfile.write(payload) def _read_data(self): if 'Content-Length' in self.headers: return self.rfile.read(int(self.headers['Content-Length'])) def _test_url(self, path, host='127.0.0.1', scheme='http', port=None): return '{0}://{1}:{2}/{3}'.format( scheme, host, port if port is not None else http_server_port(self.server), path) def do_POST(self): data = self._read_data() if self.path.startswith('/redirect_'): self._redirect() elif self.path.startswith('/method'): self._method('POST', data) elif self.path.startswith('/headers'): self._headers() else: self._status(404) def do_HEAD(self): if self.path.startswith('/redirect_'): self._redirect() elif self.path.startswith('/method'): self._method('HEAD') else: self._status(404) def do_PUT(self): data = self._read_data() if self.path.startswith('/redirect_'): self._redirect() elif self.path.startswith('/method'): self._method('PUT', data) else: self._status(404) def do_GET(self): def respond(payload=b'