|
|
@ -3,11 +3,12 @@
|
|
|
|
# Allow direct execution
|
|
|
|
# Allow direct execution
|
|
|
|
import os
|
|
|
|
import os
|
|
|
|
import sys
|
|
|
|
import sys
|
|
|
|
|
|
|
|
import time
|
|
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
|
|
|
|
from test.helper import verify_address_availability
|
|
|
|
from test.helper import verify_address_availability
|
|
|
|
from yt_dlp.networking.common import Features
|
|
|
|
from yt_dlp.networking.common import Features, DEFAULT_TIMEOUT
|
|
|
|
|
|
|
|
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
|
|
|
|
|
|
|
@ -202,11 +203,26 @@ class TestWebsSocketRequestHandlerConformance:
|
|
|
|
({'timeout': sys.float_info.min}, {}),
|
|
|
|
({'timeout': sys.float_info.min}, {}),
|
|
|
|
({}, {'timeout': sys.float_info.min}),
|
|
|
|
({}, {'timeout': sys.float_info.min}),
|
|
|
|
])
|
|
|
|
])
|
|
|
|
def test_timeout(self, handler, params, extensions):
|
|
|
|
def test_read_timeout(self, handler, params, extensions):
|
|
|
|
with handler(**params) as rh:
|
|
|
|
with handler(**params) as rh:
|
|
|
|
with pytest.raises(TransportError):
|
|
|
|
with pytest.raises(TransportError):
|
|
|
|
ws_validate_and_send(rh, Request(self.ws_base_url, extensions=extensions))
|
|
|
|
ws_validate_and_send(rh, Request(self.ws_base_url, extensions=extensions))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_connect_timeout(self, handler):
|
|
|
|
|
|
|
|
# nothing should be listening on this port
|
|
|
|
|
|
|
|
connect_timeout_url = 'ws://10.255.255.255'
|
|
|
|
|
|
|
|
with handler(timeout=0.01) as rh, pytest.raises(TransportError):
|
|
|
|
|
|
|
|
now = time.time()
|
|
|
|
|
|
|
|
ws_validate_and_send(rh, Request(connect_timeout_url))
|
|
|
|
|
|
|
|
assert time.time() - now < DEFAULT_TIMEOUT
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Per request timeout, should override handler timeout
|
|
|
|
|
|
|
|
request = Request(connect_timeout_url, extensions={'timeout': 0.01})
|
|
|
|
|
|
|
|
with handler() as rh, pytest.raises(TransportError):
|
|
|
|
|
|
|
|
now = time.time()
|
|
|
|
|
|
|
|
ws_validate_and_send(rh, request)
|
|
|
|
|
|
|
|
assert time.time() - now < DEFAULT_TIMEOUT
|
|
|
|
|
|
|
|
|
|
|
|
def test_cookies(self, handler):
|
|
|
|
def test_cookies(self, handler):
|
|
|
|
cookiejar = YoutubeDLCookieJar()
|
|
|
|
cookiejar = YoutubeDLCookieJar()
|
|
|
|
cookiejar.set_cookie(http.cookiejar.Cookie(
|
|
|
|
cookiejar.set_cookie(http.cookiejar.Cookie(
|
|
|
|