Python code coverage for Lib/_compression.py

#	count	content
1	n/a	"""Internal classes used by the gzip, lzma and bz2 modules"""
2	n/a
3	n/a	import io
4	n/a
5	n/a
6	n/a	BUFFER_SIZE = io.DEFAULT_BUFFER_SIZE # Compressed data read chunk size
7	n/a
8	n/a
9	n/a	class BaseStream(io.BufferedIOBase):
10	n/a	"""Mode-checking helper functions."""
11	n/a
12	n/a	def _check_not_closed(self):
13	n/a	if self.closed:
14	n/a	raise ValueError("I/O operation on closed file")
15	n/a
16	n/a	def _check_can_read(self):
17	n/a	if not self.readable():
18	n/a	raise io.UnsupportedOperation("File not open for reading")
19	n/a
20	n/a	def _check_can_write(self):
21	n/a	if not self.writable():
22	n/a	raise io.UnsupportedOperation("File not open for writing")
23	n/a
24	n/a	def _check_can_seek(self):
25	n/a	if not self.readable():
26	n/a	raise io.UnsupportedOperation("Seeking is only supported "
27	n/a	"on files open for reading")
28	n/a	if not self.seekable():
29	n/a	raise io.UnsupportedOperation("The underlying file object "
30	n/a	"does not support seeking")
31	n/a
32	n/a
33	n/a	class DecompressReader(io.RawIOBase):
34	n/a	"""Adapts the decompressor API to a RawIOBase reader API"""
35	n/a
36	n/a	def readable(self):
37	n/a	return True
38	n/a
39	n/a	def __init__(self, fp, decomp_factory, trailing_error=(), **decomp_args):
40	n/a	self._fp = fp
41	n/a	self._eof = False
42	n/a	self._pos = 0 # Current offset in decompressed stream
43	n/a
44	n/a	# Set to size of decompressed stream once it is known, for SEEK_END
45	n/a	self._size = -1
46	n/a
47	n/a	# Save the decompressor factory and arguments.
48	n/a	# If the file contains multiple compressed streams, each
49	n/a	# stream will need a separate decompressor object. A new decompressor
50	n/a	# object is also needed when implementing a backwards seek().
51	n/a	self._decomp_factory = decomp_factory
52	n/a	self._decomp_args = decomp_args
53	n/a	self._decompressor = self._decomp_factory(**self._decomp_args)
54	n/a
55	n/a	# Exception class to catch from decompressor signifying invalid
56	n/a	# trailing data to ignore
57	n/a	self._trailing_error = trailing_error
58	n/a
59	n/a	def close(self):
60	n/a	self._decompressor = None
61	n/a	return super().close()
62	n/a
63	n/a	def seekable(self):
64	n/a	return self._fp.seekable()
65	n/a
66	n/a	def readinto(self, b):
67	n/a	with memoryview(b) as view, view.cast("B") as byte_view:
68	n/a	data = self.read(len(byte_view))
69	n/a	byte_view[:len(data)] = data
70	n/a	return len(data)
71	n/a
72	n/a	def read(self, size=-1):
73	n/a	if size < 0:
74	n/a	return self.readall()
75	n/a
76	n/a	if not size or self._eof:
77	n/a	return b""
78	n/a	data = None # Default if EOF is encountered
79	n/a	# Depending on the input data, our call to the decompressor may not
80	n/a	# return any data. In this case, try again after reading another block.
81	n/a	while True:
82	n/a	if self._decompressor.eof:
83	n/a	rawblock = (self._decompressor.unused_data or
84	n/a	self._fp.read(BUFFER_SIZE))
85	n/a	if not rawblock:
86	n/a	break
87	n/a	# Continue to next stream.
88	n/a	self._decompressor = self._decomp_factory(
89	n/a	**self._decomp_args)
90	n/a	try:
91	n/a	data = self._decompressor.decompress(rawblock, size)
92	n/a	except self._trailing_error:
93	n/a	# Trailing data isn't a valid compressed stream; ignore it.
94	n/a	break
95	n/a	else:
96	n/a	if self._decompressor.needs_input:
97	n/a	rawblock = self._fp.read(BUFFER_SIZE)
98	n/a	if not rawblock:
99	n/a	raise EOFError("Compressed file ended before the "
100	n/a	"end-of-stream marker was reached")
101	n/a	else:
102	n/a	rawblock = b""
103	n/a	data = self._decompressor.decompress(rawblock, size)
104	n/a	if data:
105	n/a	break
106	n/a	if not data:
107	n/a	self._eof = True
108	n/a	self._size = self._pos
109	n/a	return b""
110	n/a	self._pos += len(data)
111	n/a	return data
112	n/a
113	n/a	# Rewind the file to the beginning of the data stream.
114	n/a	def _rewind(self):
115	n/a	self._fp.seek(0)
116	n/a	self._eof = False
117	n/a	self._pos = 0
118	n/a	self._decompressor = self._decomp_factory(**self._decomp_args)
119	n/a
120	n/a	def seek(self, offset, whence=io.SEEK_SET):
121	n/a	# Recalculate offset as an absolute file position.
122	n/a	if whence == io.SEEK_SET:
123	n/a	pass
124	n/a	elif whence == io.SEEK_CUR:
125	n/a	offset = self._pos + offset
126	n/a	elif whence == io.SEEK_END:
127	n/a	# Seeking relative to EOF - we need to know the file's size.
128	n/a	if self._size < 0:
129	n/a	while self.read(io.DEFAULT_BUFFER_SIZE):
130	n/a	pass
131	n/a	offset = self._size + offset
132	n/a	else:
133	n/a	raise ValueError("Invalid value for whence: {}".format(whence))
134	n/a
135	n/a	# Make it so that offset is the number of bytes to skip forward.
136	n/a	if offset < self._pos:
137	n/a	self._rewind()
138	n/a	else:
139	n/a	offset -= self._pos
140	n/a
141	n/a	# Read and discard data until we reach the desired position.
142	n/a	while offset > 0:
143	n/a	data = self.read(min(io.DEFAULT_BUFFER_SIZE, offset))
144	n/a	if not data:
145	n/a	break
146	n/a	offset -= len(data)
147	n/a
148	n/a	return self._pos
149	n/a
150	n/a	def tell(self):
151	n/a	"""Return the current file position."""
152	n/a	return self._pos