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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
import unittest
from streamlink.buffers import Buffer, RingBuffer
class TestBuffer(unittest.TestCase):
def setUp(self):
self.buffer = Buffer()
def test_write(self):
self.buffer.write(b"1" * 8192)
self.buffer.write(b"2" * 4096)
self.assertEqual(self.buffer.length, 8192 + 4096)
def test_read(self):
self.buffer.write(b"1" * 8192)
self.buffer.write(b"2" * 4096)
self.assertEqual(self.buffer.length, 8192 + 4096)
self.assertEqual(self.buffer.read(4096), b"1" * 4096)
self.assertEqual(self.buffer.read(4096), b"1" * 4096)
self.assertEqual(self.buffer.read(), b"2" * 4096)
self.assertEqual(self.buffer.read(4096), b"")
self.assertEqual(self.buffer.read(), b"")
self.assertEqual(self.buffer.length, 0)
def test_readwrite(self):
self.buffer.write(b"1" * 8192)
self.assertEqual(self.buffer.length, 8192)
self.assertEqual(self.buffer.read(4096), b"1" * 4096)
self.assertEqual(self.buffer.length, 4096)
self.buffer.write(b"2" * 4096)
self.assertEqual(self.buffer.length, 8192)
self.assertEqual(self.buffer.read(1), b"1")
self.assertEqual(self.buffer.read(4095), b"1" * 4095)
self.assertEqual(self.buffer.read(8192), b"2" * 4096)
self.assertEqual(self.buffer.read(8192), b"")
self.assertEqual(self.buffer.read(), b"")
self.assertEqual(self.buffer.length, 0)
def test_close(self):
self.buffer.write(b"1" * 8192)
self.assertEqual(self.buffer.length, 8192)
self.buffer.close()
self.buffer.write(b"2" * 8192)
self.assertEqual(self.buffer.length, 8192)
def test_reuse_input(self):
"""Objects should be reusable after write()"""
original = b"original"
tests = [bytearray(original)]
try:
m = memoryview(bytearray(original))
except NameError: # Python 2.6 does not have "memoryview"
pass
else:
# Python 2.7 doesn't do bytes(memoryview) properly
if bytes(m) == original:
tests.append(m)
for data in tests:
self.buffer.write(data)
data[:] = b"reused!!"
self.assertEqual(self.buffer.read(), original)
def test_read_empty(self):
self.assertRaises(
StopIteration,
lambda: next(self.buffer._iterate_chunks(10)))
class TestRingBuffer(unittest.TestCase):
BUFFER_SIZE = 8192 * 4
def setUp(self):
self.buffer = RingBuffer(size=self.BUFFER_SIZE)
def test_write(self):
self.buffer.write(b"1" * 8192)
self.buffer.write(b"2" * 4096)
self.assertEqual(self.buffer.length, 8192 + 4096)
def test_read(self):
self.buffer.write(b"1" * 8192)
self.buffer.write(b"2" * 4096)
self.assertEqual(self.buffer.length, 8192 + 4096)
self.assertEqual(self.buffer.read(4096), b"1" * 4096)
self.assertEqual(self.buffer.read(4096), b"1" * 4096)
self.assertEqual(self.buffer.read(), b"2" * 4096)
self.assertEqual(self.buffer.length, 0)
def test_read_timeout(self):
self.assertRaises(
IOError,
self.buffer.read, timeout=0.1)
def test_write_after_close(self):
self.buffer.close()
self.buffer.write(b"1" * 8192)
self.assertEqual(self.buffer.length, 0)
self.assertTrue(self.buffer.closed)
def test_resize(self):
self.assertEqual(self.buffer.buffer_size, self.BUFFER_SIZE)
self.buffer.resize(self.BUFFER_SIZE * 2)
self.assertEqual(self.buffer.buffer_size, self.BUFFER_SIZE * 2)
def test_free(self):
self.assertEqual(self.buffer.free, self.BUFFER_SIZE)
self.buffer.write(b'1' * 100)
self.assertEqual(self.buffer.free, self.BUFFER_SIZE - 100)
|
