File: test_stream_wrappers.py

package info (click to toggle)
streamlink 8.1.2-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 5,568 kB
  • sloc: python: 51,299; sh: 184; makefile: 152
file content (18 lines) | stat: -rw-r--r-- 565 bytes parent folder | download | duplicates (4) 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
 
from streamlink.stream.wrappers import StreamIOIterWrapper


class TestPluginStream:
    def test_iter(self):
        def generator():
            yield b"1" * 8192
            yield b"2" * 4096
            yield b"3" * 2048

        fd = StreamIOIterWrapper(generator())
        assert fd.read(4096) == b"1" * 4096
        assert fd.read(2048) == b"1" * 2048
        assert fd.read(2048) == b"1" * 2048
        assert fd.read(1) == b"2"
        assert fd.read(4095) == b"2" * 4095
        assert fd.read(1536) == b"3" * 1536
        assert fd.read() == b"3" * 512