-
Notifications
You must be signed in to change notification settings - Fork 582
/
test_buffer.py
72 lines (56 loc) · 2.32 KB
/
test_buffer.py
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
import unittest
from livestreamer.buffers import Buffer
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)
if __name__ == "__main__":
unittest.main()