1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
import threading
import time
from http.server import BaseHTTPRequestHandler
from socketserver import TCPServer
import av
from .common import TestCase, fate_suite
PORT = 8002
CONTENT = open(fate_suite("mpeg2/mpeg2_field_encoding.ts"), "rb").read()
# Needs to be long enough for all host OSes to deal.
TIMEOUT = 0.25
DELAY = 4 * TIMEOUT
class HttpServer(TCPServer):
allow_reuse_address = True
def handle_error(self, request: object, client_address: object) -> None:
pass
class SlowRequestHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
time.sleep(DELAY)
self.send_response(200)
self.send_header("Content-Length", str(len(CONTENT)))
self.end_headers()
self.wfile.write(CONTENT)
def log_message(self, format: object, *args: object) -> None:
pass
class TestTimeout(TestCase):
def setUp(cls) -> None:
cls._server = HttpServer(("", PORT), SlowRequestHandler)
cls._thread = threading.Thread(target=cls._server.handle_request)
cls._thread.daemon = True # Make sure the tests will exit.
cls._thread.start()
def tearDown(cls) -> None:
cls._thread.join(1) # Can't wait forever or the tests will never exit.
cls._server.server_close()
def test_no_timeout(self) -> None:
start = time.time()
av.open(f"http://localhost:{PORT}/mpeg2_field_encoding.ts")
duration = time.time() - start
assert duration > DELAY
def test_open_timeout(self) -> None:
with self.assertRaises(av.ExitError):
start = time.time()
av.open(f"http://localhost:{PORT}/mpeg2_field_encoding.ts", timeout=TIMEOUT)
duration = time.time() - start
assert duration < DELAY
def test_open_timeout_2(self) -> None:
with self.assertRaises(av.ExitError):
start = time.time()
av.open(
f"http://localhost:{PORT}/mpeg2_field_encoding.ts",
timeout=(TIMEOUT, None),
)
duration = time.time() - start
assert duration < DELAY
|
