1import unittest
2
3from livestreamer.buffers import Buffer
4
5class TestBuffer(unittest.TestCase):
6    def setUp(self):
7        self.buffer = Buffer()
8
9    def test_write(self):
10        self.buffer.write(b"1" * 8192)
11        self.buffer.write(b"2" * 4096)
12
13        self.assertEqual(self.buffer.length, 8192 + 4096)
14
15    def test_read(self):
16        self.buffer.write(b"1" * 8192)
17        self.buffer.write(b"2" * 4096)
18
19        self.assertEqual(self.buffer.length, 8192 + 4096)
20        self.assertEqual(self.buffer.read(4096), b"1" * 4096)
21        self.assertEqual(self.buffer.read(4096), b"1" * 4096)
22        self.assertEqual(self.buffer.read(), b"2" * 4096)
23        self.assertEqual(self.buffer.read(4096), b"")
24        self.assertEqual(self.buffer.read(), b"")
25        self.assertEqual(self.buffer.length, 0)
26
27    def test_readwrite(self):
28        self.buffer.write(b"1" * 8192)
29        self.assertEqual(self.buffer.length, 8192)
30        self.assertEqual(self.buffer.read(4096), b"1" * 4096)
31        self.assertEqual(self.buffer.length, 4096)
32
33        self.buffer.write(b"2" * 4096)
34        self.assertEqual(self.buffer.length, 8192)
35        self.assertEqual(self.buffer.read(1), b"1")
36        self.assertEqual(self.buffer.read(4095), b"1" * 4095)
37        self.assertEqual(self.buffer.read(8192), b"2" * 4096)
38        self.assertEqual(self.buffer.read(8192), b"")
39        self.assertEqual(self.buffer.read(), b"")
40        self.assertEqual(self.buffer.length, 0)
41
42    def test_close(self):
43        self.buffer.write(b"1" * 8192)
44        self.assertEqual(self.buffer.length, 8192)
45
46        self.buffer.close()
47        self.buffer.write(b"2" * 8192)
48        self.assertEqual(self.buffer.length, 8192)
49
50    def test_reuse_input(self):
51        """Objects should be reusable after write()"""
52
53        original = b"original"
54        tests = [bytearray(original)]
55        try:
56            m = memoryview(bytearray(original))
57        except NameError:  # Python 2.6 does not have "memoryview"
58            pass
59        else:
60            # Python 2.7 doesn't do bytes(memoryview) properly
61            if bytes(m) == original:
62                tests.append(m)
63
64        for data in tests:
65            self.buffer.write(data)
66            data[:] = b"reused!!"
67            self.assertEqual(self.buffer.read(), original)
68
69
70if __name__ == "__main__":
71    unittest.main()
72
73