blob: c608b1112c8b602c7534310ee917d82f0d7170fc (
plain)
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
|
#
# This file is part of pyasn1 software.
#
# Copyright (c) 2005-2019, Ilya Etingof <etingof@gmail.com>
# License: http://snmplabs.com/pyasn1/license.html
#
import io
import sys
try:
import unittest2 as unittest
except ImportError:
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)
|
