Default page

Your IP: 216.73.216.118
Server IP: 84.32.84.115
Server: Linux sg-nme-web1518.main-hosting.eu 5.14.0-611.16.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Mon Dec 22 03:40:39 EST 2025 x86_64
Server Software: LiteSpeed
PHP Version: 8.3.28
Buat File | Buat Folder

Dir : ~/opt/gsutil/third_party/pyasn1/tests/codec/

View File Name : test_streaming.py

#
# This file is part of pyasn1 software.
#
# Copyright (c) 2005-2019, Ilya Etingof <etingof@gmail.com>
# License: https://pyasn1.readthedocs.io/en/latest/license.html
#
import io
import sys
import unittest

from tests.base import BaseTestCase

from pyasn1.codec import streaming


class CachingStreamWrapperTestCase(BaseTestCase):
    def setUp(self):
        self.shortText = b"abcdefghij"
        self.longText = self.shortText * (io.DEFAULT_BUFFER_SIZE * 5)
        self.shortStream = io.BytesIO(self.shortText)
        self.longStream = io.BytesIO(self.longText)

    def testReadJustFromCache(self):
        wrapper = streaming.CachingStreamWrapper(self.shortStream)
        wrapper.read(6)
        wrapper.seek(3)
        assert wrapper.read(1) == b"d"
        assert wrapper.read(1) == b"e"
        assert wrapper.tell() == 5

    def testReadFromCacheAndStream(self):
        wrapper = streaming.CachingStreamWrapper(self.shortStream)
        wrapper.read(6)
        wrapper.seek(3)
        assert wrapper.read(4) == b"defg"
        assert wrapper.tell() == 7

    def testReadJustFromStream(self):
        wrapper = streaming.CachingStreamWrapper(self.shortStream)
        assert wrapper.read(6) == b"abcdef"
        assert wrapper.tell() == 6

    def testPeek(self):
        wrapper = streaming.CachingStreamWrapper(self.longStream)
        read_bytes = wrapper.peek(io.DEFAULT_BUFFER_SIZE + 73)
        assert len(read_bytes) == io.DEFAULT_BUFFER_SIZE + 73
        assert read_bytes.startswith(b"abcdefg")
        assert wrapper.tell() == 0
        assert wrapper.read(4) == b"abcd"

    def testMarkedPositionResets(self):
        wrapper = streaming.CachingStreamWrapper(self.longStream)
        wrapper.read(10)
        wrapper.markedPosition = wrapper.tell()
        assert wrapper.markedPosition == 10

        # Reach the maximum capacity of cache
        wrapper.read(io.DEFAULT_BUFFER_SIZE)
        assert wrapper.tell() == 10 + io.DEFAULT_BUFFER_SIZE

        # The following should clear the cache
        wrapper.markedPosition = wrapper.tell()
        assert wrapper.markedPosition == 0
        assert len(wrapper._cache.getvalue()) == 0


suite = unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])

if __name__ == '__main__':
    unittest.TextTestRunner(verbosity=2).run(suite)