Python code coverage for Lib/test/test_bz2.py

#	count	content
1	n/a	from test import support
2	n/a	from test.support import bigmemtest, _4G
3	n/a
4	n/a	import unittest
5	n/a	from io import BytesIO, DEFAULT_BUFFER_SIZE
6	n/a	import os
7	n/a	import pickle
8	n/a	import glob
9	n/a	import pathlib
10	n/a	import random
11	n/a	import shutil
12	n/a	import subprocess
13	n/a	from test.support import unlink
14	n/a	import _compression
15	n/a
16	n/a	try:
17	n/a	import threading
18	n/a	except ImportError:
19	n/a	threading = None
20	n/a
21	n/a	# Skip tests if the bz2 module doesn't exist.
22	n/a	bz2 = support.import_module('bz2')
23	n/a	from bz2 import BZ2File, BZ2Compressor, BZ2Decompressor
24	n/a
25	n/a	has_cmdline_bunzip2 = None
26	n/a
27	n/a	def ext_decompress(data):
28	n/a	global has_cmdline_bunzip2
29	n/a	if has_cmdline_bunzip2 is None:
30	n/a	has_cmdline_bunzip2 = bool(shutil.which('bunzip2'))
31	n/a	if has_cmdline_bunzip2:
32	n/a	return subprocess.check_output(['bunzip2'], input=data)
33	n/a	else:
34	n/a	return bz2.decompress(data)
35	n/a
36	n/a	class BaseTest(unittest.TestCase):
37	n/a	"Base for other testcases."
38	n/a
39	n/a	TEXT_LINES = [
40	n/a	b'root:x:0:0:root:/root:/bin/bash\n',
41	n/a	b'bin:x:1:1:bin:/bin:\n',
42	n/a	b'daemon:x:2:2:daemon:/sbin:\n',
43	n/a	b'adm:x:3:4:adm:/var/adm:\n',
44	n/a	b'lp:x:4:7:lp:/var/spool/lpd:\n',
45	n/a	b'sync:x:5:0:sync:/sbin:/bin/sync\n',
46	n/a	b'shutdown:x:6:0:shutdown:/sbin:/sbin/shutdown\n',
47	n/a	b'halt:x:7:0:halt:/sbin:/sbin/halt\n',
48	n/a	b'mail:x:8:12:mail:/var/spool/mail:\n',
49	n/a	b'news:x:9:13:news:/var/spool/news:\n',
50	n/a	b'uucp:x:10:14:uucp:/var/spool/uucp:\n',
51	n/a	b'operator:x:11:0:operator:/root:\n',
52	n/a	b'games:x:12:100:games:/usr/games:\n',
53	n/a	b'gopher:x:13:30:gopher:/usr/lib/gopher-data:\n',
54	n/a	b'ftp:x:14:50:FTP User:/var/ftp:/bin/bash\n',
55	n/a	b'nobody:x:65534:65534:Nobody:/home:\n',
56	n/a	b'postfix:x:100:101:postfix:/var/spool/postfix:\n',
57	n/a	b'niemeyer:x:500:500::/home/niemeyer:/bin/bash\n',
58	n/a	b'postgres:x:101:102:PostgreSQL Server:/var/lib/pgsql:/bin/bash\n',
59	n/a	b'mysql:x:102:103:MySQL server:/var/lib/mysql:/bin/bash\n',
60	n/a	b'www:x:103:104::/var/www:/bin/false\n',
61	n/a	]
62	n/a	TEXT = b''.join(TEXT_LINES)
63	n/a	DATA = b'BZh91AY&SY.\xc8N\x18\x00\x01>_\x80\x00\x10@\x02\xff\xf0\x01\x07n\x00?\xe7\xff\xe00\x01\x99\xaa\x00\xc0\x03F\x86\x8c#&\x83F\x9a\x03\x06\xa6\xd0\xa6\x93M\x0fQ\xa7\xa8\x06\x804hh\x12$\x11\xa4i4\xf14S\xd2<Q\xb5\x0fH\xd3\xd4\xdd\xd5\x87\xbb\xf8\x94\r\x8f\xafI\x12\xe1\xc9\xf8/E\x00pu\x89\x12]\xc9\xbbDL\nQ\x0e\t1\x12\xdf\xa0\xc0\x97\xac2O9\x89\x13\x94\x0e\x1c7\x0ed\x95I\x0c\xaaJ\xa4\x18L\x10\x05#\x9c\xaf\xba\xbc/\x97\x8a#C\xc8\xe1\x8cW\xf9\xe2\xd0\xd6M\xa7\x8bXa<e\x84t\xcbL\xb3\xa7\xd9\xcd\xd1\xcb\x84.\xaf\xb3\xab\xab\xad`n}\xa0lh\tE,\x8eZ\x15\x17VH>\x88\xe5\xcd9gd6\x0b\n\xe9\x9b\xd5\x8a\x99\xf7\x08.K\x8ev\xfb\xf7xw\xbb\xdf\xa1\x92\xf1\xdd\|/";\xa2\xba\x9f\xd5\xb1#A\xb6\xf6\xb3o\xc9\xc5y\\\xebO\xe7\x85\x9a\xbc\xb6f8\x952\xd5\xd7"%\x89>V,\xf7\xa6z\xe2\x9f\xa3\xdf\x11\x11"\xd6E)I\xa9\x13^\xca\xf3r\xd0\x03U\x922\xf26\xec\xb6\xed\x8b\xc3U\x13\x9d\xc5\x170\xa4\xfa^\x92\xacDF\x8a\x97\xd6\x19\xfe\xdd\xb8\xbd\x1a\x9a\x19\xa3\x80ankR\x8b\xe5\xd83]\xa9\xc6\x08\x82f\xf6\xb9"6l$\xb8j@\xc0\x8a\xb0l1..\xbak\x83ls\x15\xbc\xf4\xc1\x13\xbe\xf8E\xb8\x9d\r\xa8\x9dk\x84\xd3n\xfa\xacQ\x07\xb1%y\xaav\xb4\x08\xe0z\x1b\x16\xf5\x04\xe9\xcc\xb9\x08z\x1en7.G\xfc]\xc9\x14\xe1B@\xbb!8`'
64	n/a	EMPTY_DATA = b'BZh9\x17rE8P\x90\x00\x00\x00\x00'
65	n/a	BAD_DATA = b'this is not a valid bzip2 file'
66	n/a
67	n/a	# Some tests need more than one block of uncompressed data. Since one block
68	n/a	# is at least 100 kB, we gather some data dynamically and compress it.
69	n/a	# Note that this assumes that compression works correctly, so we cannot
70	n/a	# simply use the bigger test data for all tests.
71	n/a	test_size = 0
72	n/a	BIG_TEXT = bytearray(128*1024)
73	n/a	for fname in glob.glob(os.path.join(os.path.dirname(__file__), '*.py')):
74	n/a	with open(fname, 'rb') as fh:
75	n/a	test_size += fh.readinto(memoryview(BIG_TEXT)[test_size:])
76	n/a	if test_size > 128*1024:
77	n/a	break
78	n/a	BIG_DATA = bz2.compress(BIG_TEXT, compresslevel=1)
79	n/a
80	n/a	def setUp(self):
81	n/a	self.filename = support.TESTFN
82	n/a
83	n/a	def tearDown(self):
84	n/a	if os.path.isfile(self.filename):
85	n/a	os.unlink(self.filename)
86	n/a
87	n/a
88	n/a	class BZ2FileTest(BaseTest):
89	n/a	"Test the BZ2File class."
90	n/a
91	n/a	def createTempFile(self, streams=1, suffix=b""):
92	n/a	with open(self.filename, "wb") as f:
93	n/a	f.write(self.DATA * streams)
94	n/a	f.write(suffix)
95	n/a
96	n/a	def testBadArgs(self):
97	n/a	self.assertRaises(TypeError, BZ2File, 123.456)
98	n/a	self.assertRaises(ValueError, BZ2File, os.devnull, "z")
99	n/a	self.assertRaises(ValueError, BZ2File, os.devnull, "rx")
100	n/a	self.assertRaises(ValueError, BZ2File, os.devnull, "rbt")
101	n/a	self.assertRaises(ValueError, BZ2File, os.devnull, compresslevel=0)
102	n/a	self.assertRaises(ValueError, BZ2File, os.devnull, compresslevel=10)
103	n/a
104	n/a	def testRead(self):
105	n/a	self.createTempFile()
106	n/a	with BZ2File(self.filename) as bz2f:
107	n/a	self.assertRaises(TypeError, bz2f.read, float())
108	n/a	self.assertEqual(bz2f.read(), self.TEXT)
109	n/a
110	n/a	def testReadBadFile(self):
111	n/a	self.createTempFile(streams=0, suffix=self.BAD_DATA)
112	n/a	with BZ2File(self.filename) as bz2f:
113	n/a	self.assertRaises(OSError, bz2f.read)
114	n/a
115	n/a	def testReadMultiStream(self):
116	n/a	self.createTempFile(streams=5)
117	n/a	with BZ2File(self.filename) as bz2f:
118	n/a	self.assertRaises(TypeError, bz2f.read, float())
119	n/a	self.assertEqual(bz2f.read(), self.TEXT * 5)
120	n/a
121	n/a	def testReadMonkeyMultiStream(self):
122	n/a	# Test BZ2File.read() on a multi-stream archive where a stream
123	n/a	# boundary coincides with the end of the raw read buffer.
124	n/a	buffer_size = _compression.BUFFER_SIZE
125	n/a	_compression.BUFFER_SIZE = len(self.DATA)
126	n/a	try:
127	n/a	self.createTempFile(streams=5)
128	n/a	with BZ2File(self.filename) as bz2f:
129	n/a	self.assertRaises(TypeError, bz2f.read, float())
130	n/a	self.assertEqual(bz2f.read(), self.TEXT * 5)
131	n/a	finally:
132	n/a	_compression.BUFFER_SIZE = buffer_size
133	n/a
134	n/a	def testReadTrailingJunk(self):
135	n/a	self.createTempFile(suffix=self.BAD_DATA)
136	n/a	with BZ2File(self.filename) as bz2f:
137	n/a	self.assertEqual(bz2f.read(), self.TEXT)
138	n/a
139	n/a	def testReadMultiStreamTrailingJunk(self):
140	n/a	self.createTempFile(streams=5, suffix=self.BAD_DATA)
141	n/a	with BZ2File(self.filename) as bz2f:
142	n/a	self.assertEqual(bz2f.read(), self.TEXT * 5)
143	n/a
144	n/a	def testRead0(self):
145	n/a	self.createTempFile()
146	n/a	with BZ2File(self.filename) as bz2f:
147	n/a	self.assertRaises(TypeError, bz2f.read, float())
148	n/a	self.assertEqual(bz2f.read(0), b"")
149	n/a
150	n/a	def testReadChunk10(self):
151	n/a	self.createTempFile()
152	n/a	with BZ2File(self.filename) as bz2f:
153	n/a	text = b''
154	n/a	while True:
155	n/a	str = bz2f.read(10)
156	n/a	if not str:
157	n/a	break
158	n/a	text += str
159	n/a	self.assertEqual(text, self.TEXT)
160	n/a
161	n/a	def testReadChunk10MultiStream(self):
162	n/a	self.createTempFile(streams=5)
163	n/a	with BZ2File(self.filename) as bz2f:
164	n/a	text = b''
165	n/a	while True:
166	n/a	str = bz2f.read(10)
167	n/a	if not str:
168	n/a	break
169	n/a	text += str
170	n/a	self.assertEqual(text, self.TEXT * 5)
171	n/a
172	n/a	def testRead100(self):
173	n/a	self.createTempFile()
174	n/a	with BZ2File(self.filename) as bz2f:
175	n/a	self.assertEqual(bz2f.read(100), self.TEXT[:100])
176	n/a
177	n/a	def testPeek(self):
178	n/a	self.createTempFile()
179	n/a	with BZ2File(self.filename) as bz2f:
180	n/a	pdata = bz2f.peek()
181	n/a	self.assertNotEqual(len(pdata), 0)
182	n/a	self.assertTrue(self.TEXT.startswith(pdata))
183	n/a	self.assertEqual(bz2f.read(), self.TEXT)
184	n/a
185	n/a	def testReadInto(self):
186	n/a	self.createTempFile()
187	n/a	with BZ2File(self.filename) as bz2f:
188	n/a	n = 128
189	n/a	b = bytearray(n)
190	n/a	self.assertEqual(bz2f.readinto(b), n)
191	n/a	self.assertEqual(b, self.TEXT[:n])
192	n/a	n = len(self.TEXT) - n
193	n/a	b = bytearray(len(self.TEXT))
194	n/a	self.assertEqual(bz2f.readinto(b), n)
195	n/a	self.assertEqual(b[:n], self.TEXT[-n:])
196	n/a
197	n/a	def testReadLine(self):
198	n/a	self.createTempFile()
199	n/a	with BZ2File(self.filename) as bz2f:
200	n/a	self.assertRaises(TypeError, bz2f.readline, None)
201	n/a	for line in self.TEXT_LINES:
202	n/a	self.assertEqual(bz2f.readline(), line)
203	n/a
204	n/a	def testReadLineMultiStream(self):
205	n/a	self.createTempFile(streams=5)
206	n/a	with BZ2File(self.filename) as bz2f:
207	n/a	self.assertRaises(TypeError, bz2f.readline, None)
208	n/a	for line in self.TEXT_LINES * 5:
209	n/a	self.assertEqual(bz2f.readline(), line)
210	n/a
211	n/a	def testReadLines(self):
212	n/a	self.createTempFile()
213	n/a	with BZ2File(self.filename) as bz2f:
214	n/a	self.assertRaises(TypeError, bz2f.readlines, None)
215	n/a	self.assertEqual(bz2f.readlines(), self.TEXT_LINES)
216	n/a
217	n/a	def testReadLinesMultiStream(self):
218	n/a	self.createTempFile(streams=5)
219	n/a	with BZ2File(self.filename) as bz2f:
220	n/a	self.assertRaises(TypeError, bz2f.readlines, None)
221	n/a	self.assertEqual(bz2f.readlines(), self.TEXT_LINES * 5)
222	n/a
223	n/a	def testIterator(self):
224	n/a	self.createTempFile()
225	n/a	with BZ2File(self.filename) as bz2f:
226	n/a	self.assertEqual(list(iter(bz2f)), self.TEXT_LINES)
227	n/a
228	n/a	def testIteratorMultiStream(self):
229	n/a	self.createTempFile(streams=5)
230	n/a	with BZ2File(self.filename) as bz2f:
231	n/a	self.assertEqual(list(iter(bz2f)), self.TEXT_LINES * 5)
232	n/a
233	n/a	def testClosedIteratorDeadlock(self):
234	n/a	# Issue #3309: Iteration on a closed BZ2File should release the lock.
235	n/a	self.createTempFile()
236	n/a	bz2f = BZ2File(self.filename)
237	n/a	bz2f.close()
238	n/a	self.assertRaises(ValueError, next, bz2f)
239	n/a	# This call will deadlock if the above call failed to release the lock.
240	n/a	self.assertRaises(ValueError, bz2f.readlines)
241	n/a
242	n/a	def testWrite(self):
243	n/a	with BZ2File(self.filename, "w") as bz2f:
244	n/a	self.assertRaises(TypeError, bz2f.write)
245	n/a	bz2f.write(self.TEXT)
246	n/a	with open(self.filename, 'rb') as f:
247	n/a	self.assertEqual(ext_decompress(f.read()), self.TEXT)
248	n/a
249	n/a	def testWriteChunks10(self):
250	n/a	with BZ2File(self.filename, "w") as bz2f:
251	n/a	n = 0
252	n/a	while True:
253	n/a	str = self.TEXT[n10:(n+1)10]
254	n/a	if not str:
255	n/a	break
256	n/a	bz2f.write(str)
257	n/a	n += 1
258	n/a	with open(self.filename, 'rb') as f:
259	n/a	self.assertEqual(ext_decompress(f.read()), self.TEXT)
260	n/a
261	n/a	def testWriteNonDefaultCompressLevel(self):
262	n/a	expected = bz2.compress(self.TEXT, compresslevel=5)
263	n/a	with BZ2File(self.filename, "w", compresslevel=5) as bz2f:
264	n/a	bz2f.write(self.TEXT)
265	n/a	with open(self.filename, "rb") as f:
266	n/a	self.assertEqual(f.read(), expected)
267	n/a
268	n/a	def testWriteLines(self):
269	n/a	with BZ2File(self.filename, "w") as bz2f:
270	n/a	self.assertRaises(TypeError, bz2f.writelines)
271	n/a	bz2f.writelines(self.TEXT_LINES)
272	n/a	# Issue #1535500: Calling writelines() on a closed BZ2File
273	n/a	# should raise an exception.
274	n/a	self.assertRaises(ValueError, bz2f.writelines, ["a"])
275	n/a	with open(self.filename, 'rb') as f:
276	n/a	self.assertEqual(ext_decompress(f.read()), self.TEXT)
277	n/a
278	n/a	def testWriteMethodsOnReadOnlyFile(self):
279	n/a	with BZ2File(self.filename, "w") as bz2f:
280	n/a	bz2f.write(b"abc")
281	n/a
282	n/a	with BZ2File(self.filename, "r") as bz2f:
283	n/a	self.assertRaises(OSError, bz2f.write, b"a")
284	n/a	self.assertRaises(OSError, bz2f.writelines, [b"a"])
285	n/a
286	n/a	def testAppend(self):
287	n/a	with BZ2File(self.filename, "w") as bz2f:
288	n/a	self.assertRaises(TypeError, bz2f.write)
289	n/a	bz2f.write(self.TEXT)
290	n/a	with BZ2File(self.filename, "a") as bz2f:
291	n/a	self.assertRaises(TypeError, bz2f.write)
292	n/a	bz2f.write(self.TEXT)
293	n/a	with open(self.filename, 'rb') as f:
294	n/a	self.assertEqual(ext_decompress(f.read()), self.TEXT * 2)
295	n/a
296	n/a	def testSeekForward(self):
297	n/a	self.createTempFile()
298	n/a	with BZ2File(self.filename) as bz2f:
299	n/a	self.assertRaises(TypeError, bz2f.seek)
300	n/a	bz2f.seek(150)
301	n/a	self.assertEqual(bz2f.read(), self.TEXT[150:])
302	n/a
303	n/a	def testSeekForwardAcrossStreams(self):
304	n/a	self.createTempFile(streams=2)
305	n/a	with BZ2File(self.filename) as bz2f:
306	n/a	self.assertRaises(TypeError, bz2f.seek)
307	n/a	bz2f.seek(len(self.TEXT) + 150)
308	n/a	self.assertEqual(bz2f.read(), self.TEXT[150:])
309	n/a
310	n/a	def testSeekBackwards(self):
311	n/a	self.createTempFile()
312	n/a	with BZ2File(self.filename) as bz2f:
313	n/a	bz2f.read(500)
314	n/a	bz2f.seek(-150, 1)
315	n/a	self.assertEqual(bz2f.read(), self.TEXT[500-150:])
316	n/a
317	n/a	def testSeekBackwardsAcrossStreams(self):
318	n/a	self.createTempFile(streams=2)
319	n/a	with BZ2File(self.filename) as bz2f:
320	n/a	readto = len(self.TEXT) + 100
321	n/a	while readto > 0:
322	n/a	readto -= len(bz2f.read(readto))
323	n/a	bz2f.seek(-150, 1)
324	n/a	self.assertEqual(bz2f.read(), self.TEXT[100-150:] + self.TEXT)
325	n/a
326	n/a	def testSeekBackwardsFromEnd(self):
327	n/a	self.createTempFile()
328	n/a	with BZ2File(self.filename) as bz2f:
329	n/a	bz2f.seek(-150, 2)
330	n/a	self.assertEqual(bz2f.read(), self.TEXT[len(self.TEXT)-150:])
331	n/a
332	n/a	def testSeekBackwardsFromEndAcrossStreams(self):
333	n/a	self.createTempFile(streams=2)
334	n/a	with BZ2File(self.filename) as bz2f:
335	n/a	bz2f.seek(-1000, 2)
336	n/a	self.assertEqual(bz2f.read(), (self.TEXT * 2)[-1000:])
337	n/a
338	n/a	def testSeekPostEnd(self):
339	n/a	self.createTempFile()
340	n/a	with BZ2File(self.filename) as bz2f:
341	n/a	bz2f.seek(150000)
342	n/a	self.assertEqual(bz2f.tell(), len(self.TEXT))
343	n/a	self.assertEqual(bz2f.read(), b"")
344	n/a
345	n/a	def testSeekPostEndMultiStream(self):
346	n/a	self.createTempFile(streams=5)
347	n/a	with BZ2File(self.filename) as bz2f:
348	n/a	bz2f.seek(150000)
349	n/a	self.assertEqual(bz2f.tell(), len(self.TEXT) * 5)
350	n/a	self.assertEqual(bz2f.read(), b"")
351	n/a
352	n/a	def testSeekPostEndTwice(self):
353	n/a	self.createTempFile()
354	n/a	with BZ2File(self.filename) as bz2f:
355	n/a	bz2f.seek(150000)
356	n/a	bz2f.seek(150000)
357	n/a	self.assertEqual(bz2f.tell(), len(self.TEXT))
358	n/a	self.assertEqual(bz2f.read(), b"")
359	n/a
360	n/a	def testSeekPostEndTwiceMultiStream(self):
361	n/a	self.createTempFile(streams=5)
362	n/a	with BZ2File(self.filename) as bz2f:
363	n/a	bz2f.seek(150000)
364	n/a	bz2f.seek(150000)
365	n/a	self.assertEqual(bz2f.tell(), len(self.TEXT) * 5)
366	n/a	self.assertEqual(bz2f.read(), b"")
367	n/a
368	n/a	def testSeekPreStart(self):
369	n/a	self.createTempFile()
370	n/a	with BZ2File(self.filename) as bz2f:
371	n/a	bz2f.seek(-150)
372	n/a	self.assertEqual(bz2f.tell(), 0)
373	n/a	self.assertEqual(bz2f.read(), self.TEXT)
374	n/a
375	n/a	def testSeekPreStartMultiStream(self):
376	n/a	self.createTempFile(streams=2)
377	n/a	with BZ2File(self.filename) as bz2f:
378	n/a	bz2f.seek(-150)
379	n/a	self.assertEqual(bz2f.tell(), 0)
380	n/a	self.assertEqual(bz2f.read(), self.TEXT * 2)
381	n/a
382	n/a	def testFileno(self):
383	n/a	self.createTempFile()
384	n/a	with open(self.filename, 'rb') as rawf:
385	n/a	bz2f = BZ2File(rawf)
386	n/a	try:
387	n/a	self.assertEqual(bz2f.fileno(), rawf.fileno())
388	n/a	finally:
389	n/a	bz2f.close()
390	n/a	self.assertRaises(ValueError, bz2f.fileno)
391	n/a
392	n/a	def testSeekable(self):
393	n/a	bz2f = BZ2File(BytesIO(self.DATA))
394	n/a	try:
395	n/a	self.assertTrue(bz2f.seekable())
396	n/a	bz2f.read()
397	n/a	self.assertTrue(bz2f.seekable())
398	n/a	finally:
399	n/a	bz2f.close()
400	n/a	self.assertRaises(ValueError, bz2f.seekable)
401	n/a
402	n/a	bz2f = BZ2File(BytesIO(), "w")
403	n/a	try:
404	n/a	self.assertFalse(bz2f.seekable())
405	n/a	finally:
406	n/a	bz2f.close()
407	n/a	self.assertRaises(ValueError, bz2f.seekable)
408	n/a
409	n/a	src = BytesIO(self.DATA)
410	n/a	src.seekable = lambda: False
411	n/a	bz2f = BZ2File(src)
412	n/a	try:
413	n/a	self.assertFalse(bz2f.seekable())
414	n/a	finally:
415	n/a	bz2f.close()
416	n/a	self.assertRaises(ValueError, bz2f.seekable)
417	n/a
418	n/a	def testReadable(self):
419	n/a	bz2f = BZ2File(BytesIO(self.DATA))
420	n/a	try:
421	n/a	self.assertTrue(bz2f.readable())
422	n/a	bz2f.read()
423	n/a	self.assertTrue(bz2f.readable())
424	n/a	finally:
425	n/a	bz2f.close()
426	n/a	self.assertRaises(ValueError, bz2f.readable)
427	n/a
428	n/a	bz2f = BZ2File(BytesIO(), "w")
429	n/a	try:
430	n/a	self.assertFalse(bz2f.readable())
431	n/a	finally:
432	n/a	bz2f.close()
433	n/a	self.assertRaises(ValueError, bz2f.readable)
434	n/a
435	n/a	def testWritable(self):
436	n/a	bz2f = BZ2File(BytesIO(self.DATA))
437	n/a	try:
438	n/a	self.assertFalse(bz2f.writable())
439	n/a	bz2f.read()
440	n/a	self.assertFalse(bz2f.writable())
441	n/a	finally:
442	n/a	bz2f.close()
443	n/a	self.assertRaises(ValueError, bz2f.writable)
444	n/a
445	n/a	bz2f = BZ2File(BytesIO(), "w")
446	n/a	try:
447	n/a	self.assertTrue(bz2f.writable())
448	n/a	finally:
449	n/a	bz2f.close()
450	n/a	self.assertRaises(ValueError, bz2f.writable)
451	n/a
452	n/a	def testOpenDel(self):
453	n/a	self.createTempFile()
454	n/a	for i in range(10000):
455	n/a	o = BZ2File(self.filename)
456	n/a	del o
457	n/a
458	n/a	def testOpenNonexistent(self):
459	n/a	self.assertRaises(OSError, BZ2File, "/non/existent")
460	n/a
461	n/a	def testReadlinesNoNewline(self):
462	n/a	# Issue #1191043: readlines() fails on a file containing no newline.
463	n/a	data = b'BZh91AY&SY\xd9b\x89]\x00\x00\x00\x03\x80\x04\x00\x02\x00\x0c\x00 \x00!\x9ah3M\x13<]\xc9\x14\xe1BCe\x8a%t'
464	n/a	with open(self.filename, "wb") as f:
465	n/a	f.write(data)
466	n/a	with BZ2File(self.filename) as bz2f:
467	n/a	lines = bz2f.readlines()
468	n/a	self.assertEqual(lines, [b'Test'])
469	n/a	with BZ2File(self.filename) as bz2f:
470	n/a	xlines = list(bz2f.readlines())
471	n/a	self.assertEqual(xlines, [b'Test'])
472	n/a
473	n/a	def testContextProtocol(self):
474	n/a	f = None
475	n/a	with BZ2File(self.filename, "wb") as f:
476	n/a	f.write(b"xxx")
477	n/a	f = BZ2File(self.filename, "rb")
478	n/a	f.close()
479	n/a	try:
480	n/a	with f:
481	n/a	pass
482	n/a	except ValueError:
483	n/a	pass
484	n/a	else:
485	n/a	self.fail("__enter__ on a closed file didn't raise an exception")
486	n/a	try:
487	n/a	with BZ2File(self.filename, "wb") as f:
488	n/a	1/0
489	n/a	except ZeroDivisionError:
490	n/a	pass
491	n/a	else:
492	n/a	self.fail("1/0 didn't raise an exception")
493	n/a
494	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
495	n/a	def testThreading(self):
496	n/a	# Issue #7205: Using a BZ2File from several threads shouldn't deadlock.
497	n/a	data = b"1" * 2**20
498	n/a	nthreads = 10
499	n/a	with BZ2File(self.filename, 'wb') as f:
500	n/a	def comp():
501	n/a	for i in range(5):
502	n/a	f.write(data)
503	n/a	threads = [threading.Thread(target=comp) for i in range(nthreads)]
504	n/a	with support.start_threads(threads):
505	n/a	pass
506	n/a
507	n/a	def testWithoutThreading(self):
508	n/a	module = support.import_fresh_module("bz2", blocked=("threading",))
509	n/a	with module.BZ2File(self.filename, "wb") as f:
510	n/a	f.write(b"abc")
511	n/a	with module.BZ2File(self.filename, "rb") as f:
512	n/a	self.assertEqual(f.read(), b"abc")
513	n/a
514	n/a	def testMixedIterationAndReads(self):
515	n/a	self.createTempFile()
516	n/a	linelen = len(self.TEXT_LINES[0])
517	n/a	halflen = linelen // 2
518	n/a	with BZ2File(self.filename) as bz2f:
519	n/a	bz2f.read(halflen)
520	n/a	self.assertEqual(next(bz2f), self.TEXT_LINES[0][halflen:])
521	n/a	self.assertEqual(bz2f.read(), self.TEXT[linelen:])
522	n/a	with BZ2File(self.filename) as bz2f:
523	n/a	bz2f.readline()
524	n/a	self.assertEqual(next(bz2f), self.TEXT_LINES[1])
525	n/a	self.assertEqual(bz2f.readline(), self.TEXT_LINES[2])
526	n/a	with BZ2File(self.filename) as bz2f:
527	n/a	bz2f.readlines()
528	n/a	self.assertRaises(StopIteration, next, bz2f)
529	n/a	self.assertEqual(bz2f.readlines(), [])
530	n/a
531	n/a	def testMultiStreamOrdering(self):
532	n/a	# Test the ordering of streams when reading a multi-stream archive.
533	n/a	data1 = b"foo" * 1000
534	n/a	data2 = b"bar" * 1000
535	n/a	with BZ2File(self.filename, "w") as bz2f:
536	n/a	bz2f.write(data1)
537	n/a	with BZ2File(self.filename, "a") as bz2f:
538	n/a	bz2f.write(data2)
539	n/a	with BZ2File(self.filename) as bz2f:
540	n/a	self.assertEqual(bz2f.read(), data1 + data2)
541	n/a
542	n/a	def testOpenBytesFilename(self):
543	n/a	str_filename = self.filename
544	n/a	try:
545	n/a	bytes_filename = str_filename.encode("ascii")
546	n/a	except UnicodeEncodeError:
547	n/a	self.skipTest("Temporary file name needs to be ASCII")
548	n/a	with BZ2File(bytes_filename, "wb") as f:
549	n/a	f.write(self.DATA)
550	n/a	with BZ2File(bytes_filename, "rb") as f:
551	n/a	self.assertEqual(f.read(), self.DATA)
552	n/a	# Sanity check that we are actually operating on the right file.
553	n/a	with BZ2File(str_filename, "rb") as f:
554	n/a	self.assertEqual(f.read(), self.DATA)
555	n/a
556	n/a	def testOpenPathLikeFilename(self):
557	n/a	filename = pathlib.Path(self.filename)
558	n/a	with BZ2File(filename, "wb") as f:
559	n/a	f.write(self.DATA)
560	n/a	with BZ2File(filename, "rb") as f:
561	n/a	self.assertEqual(f.read(), self.DATA)
562	n/a
563	n/a	def testDecompressLimited(self):
564	n/a	"""Decompressed data buffering should be limited"""
565	n/a	bomb = bz2.compress(b'\0' * int(2e6), compresslevel=9)
566	n/a	self.assertLess(len(bomb), _compression.BUFFER_SIZE)
567	n/a
568	n/a	decomp = BZ2File(BytesIO(bomb))
569	n/a	self.assertEqual(decomp.read(1), b'\0')
570	n/a	max_decomp = 1 + DEFAULT_BUFFER_SIZE
571	n/a	self.assertLessEqual(decomp._buffer.raw.tell(), max_decomp,
572	n/a	"Excessive amount of data was decompressed")
573	n/a
574	n/a
575	n/a	# Tests for a BZ2File wrapping another file object:
576	n/a
577	n/a	def testReadBytesIO(self):
578	n/a	with BytesIO(self.DATA) as bio:
579	n/a	with BZ2File(bio) as bz2f:
580	n/a	self.assertRaises(TypeError, bz2f.read, float())
581	n/a	self.assertEqual(bz2f.read(), self.TEXT)
582	n/a	self.assertFalse(bio.closed)
583	n/a
584	n/a	def testPeekBytesIO(self):
585	n/a	with BytesIO(self.DATA) as bio:
586	n/a	with BZ2File(bio) as bz2f:
587	n/a	pdata = bz2f.peek()
588	n/a	self.assertNotEqual(len(pdata), 0)
589	n/a	self.assertTrue(self.TEXT.startswith(pdata))
590	n/a	self.assertEqual(bz2f.read(), self.TEXT)
591	n/a
592	n/a	def testWriteBytesIO(self):
593	n/a	with BytesIO() as bio:
594	n/a	with BZ2File(bio, "w") as bz2f:
595	n/a	self.assertRaises(TypeError, bz2f.write)
596	n/a	bz2f.write(self.TEXT)
597	n/a	self.assertEqual(ext_decompress(bio.getvalue()), self.TEXT)
598	n/a	self.assertFalse(bio.closed)
599	n/a
600	n/a	def testSeekForwardBytesIO(self):
601	n/a	with BytesIO(self.DATA) as bio:
602	n/a	with BZ2File(bio) as bz2f:
603	n/a	self.assertRaises(TypeError, bz2f.seek)
604	n/a	bz2f.seek(150)
605	n/a	self.assertEqual(bz2f.read(), self.TEXT[150:])
606	n/a
607	n/a	def testSeekBackwardsBytesIO(self):
608	n/a	with BytesIO(self.DATA) as bio:
609	n/a	with BZ2File(bio) as bz2f:
610	n/a	bz2f.read(500)
611	n/a	bz2f.seek(-150, 1)
612	n/a	self.assertEqual(bz2f.read(), self.TEXT[500-150:])
613	n/a
614	n/a	def test_read_truncated(self):
615	n/a	# Drop the eos_magic field (6 bytes) and CRC (4 bytes).
616	n/a	truncated = self.DATA[:-10]
617	n/a	with BZ2File(BytesIO(truncated)) as f:
618	n/a	self.assertRaises(EOFError, f.read)
619	n/a	with BZ2File(BytesIO(truncated)) as f:
620	n/a	self.assertEqual(f.read(len(self.TEXT)), self.TEXT)
621	n/a	self.assertRaises(EOFError, f.read, 1)
622	n/a	# Incomplete 4-byte file header, and block header of at least 146 bits.
623	n/a	for i in range(22):
624	n/a	with BZ2File(BytesIO(truncated[:i])) as f:
625	n/a	self.assertRaises(EOFError, f.read, 1)
626	n/a
627	n/a
628	n/a	class BZ2CompressorTest(BaseTest):
629	n/a	def testCompress(self):
630	n/a	bz2c = BZ2Compressor()
631	n/a	self.assertRaises(TypeError, bz2c.compress)
632	n/a	data = bz2c.compress(self.TEXT)
633	n/a	data += bz2c.flush()
634	n/a	self.assertEqual(ext_decompress(data), self.TEXT)
635	n/a
636	n/a	def testCompressEmptyString(self):
637	n/a	bz2c = BZ2Compressor()
638	n/a	data = bz2c.compress(b'')
639	n/a	data += bz2c.flush()
640	n/a	self.assertEqual(data, self.EMPTY_DATA)
641	n/a
642	n/a	def testCompressChunks10(self):
643	n/a	bz2c = BZ2Compressor()
644	n/a	n = 0
645	n/a	data = b''
646	n/a	while True:
647	n/a	str = self.TEXT[n10:(n+1)10]
648	n/a	if not str:
649	n/a	break
650	n/a	data += bz2c.compress(str)
651	n/a	n += 1
652	n/a	data += bz2c.flush()
653	n/a	self.assertEqual(ext_decompress(data), self.TEXT)
654	n/a
655	n/a	@bigmemtest(size=_4G + 100, memuse=2)
656	n/a	def testCompress4G(self, size):
657	n/a	# "Test BZ2Compressor.compress()/flush() with >4GiB input"
658	n/a	bz2c = BZ2Compressor()
659	n/a	data = b"x" * size
660	n/a	try:
661	n/a	compressed = bz2c.compress(data)
662	n/a	compressed += bz2c.flush()
663	n/a	finally:
664	n/a	data = None # Release memory
665	n/a	data = bz2.decompress(compressed)
666	n/a	try:
667	n/a	self.assertEqual(len(data), size)
668	n/a	self.assertEqual(len(data.strip(b"x")), 0)
669	n/a	finally:
670	n/a	data = None
671	n/a
672	n/a	def testPickle(self):
673	n/a	for proto in range(pickle.HIGHEST_PROTOCOL + 1):
674	n/a	with self.assertRaises(TypeError):
675	n/a	pickle.dumps(BZ2Compressor(), proto)
676	n/a
677	n/a
678	n/a	class BZ2DecompressorTest(BaseTest):
679	n/a	def test_Constructor(self):
680	n/a	self.assertRaises(TypeError, BZ2Decompressor, 42)
681	n/a
682	n/a	def testDecompress(self):
683	n/a	bz2d = BZ2Decompressor()
684	n/a	self.assertRaises(TypeError, bz2d.decompress)
685	n/a	text = bz2d.decompress(self.DATA)
686	n/a	self.assertEqual(text, self.TEXT)
687	n/a
688	n/a	def testDecompressChunks10(self):
689	n/a	bz2d = BZ2Decompressor()
690	n/a	text = b''
691	n/a	n = 0
692	n/a	while True:
693	n/a	str = self.DATA[n10:(n+1)10]
694	n/a	if not str:
695	n/a	break
696	n/a	text += bz2d.decompress(str)
697	n/a	n += 1
698	n/a	self.assertEqual(text, self.TEXT)
699	n/a
700	n/a	def testDecompressUnusedData(self):
701	n/a	bz2d = BZ2Decompressor()
702	n/a	unused_data = b"this is unused data"
703	n/a	text = bz2d.decompress(self.DATA+unused_data)
704	n/a	self.assertEqual(text, self.TEXT)
705	n/a	self.assertEqual(bz2d.unused_data, unused_data)
706	n/a
707	n/a	def testEOFError(self):
708	n/a	bz2d = BZ2Decompressor()
709	n/a	text = bz2d.decompress(self.DATA)
710	n/a	self.assertRaises(EOFError, bz2d.decompress, b"anything")
711	n/a	self.assertRaises(EOFError, bz2d.decompress, b"")
712	n/a
713	n/a	@bigmemtest(size=_4G + 100, memuse=3.3)
714	n/a	def testDecompress4G(self, size):
715	n/a	# "Test BZ2Decompressor.decompress() with >4GiB input"
716	n/a	blocksize = 10 * 1024 * 1024
717	n/a	block = random.getrandbits(blocksize * 8).to_bytes(blocksize, 'little')
718	n/a	try:
719	n/a	data = block * (size // blocksize + 1)
720	n/a	compressed = bz2.compress(data)
721	n/a	bz2d = BZ2Decompressor()
722	n/a	decompressed = bz2d.decompress(compressed)
723	n/a	self.assertTrue(decompressed == data)
724	n/a	finally:
725	n/a	data = None
726	n/a	compressed = None
727	n/a	decompressed = None
728	n/a
729	n/a	def testPickle(self):
730	n/a	for proto in range(pickle.HIGHEST_PROTOCOL + 1):
731	n/a	with self.assertRaises(TypeError):
732	n/a	pickle.dumps(BZ2Decompressor(), proto)
733	n/a
734	n/a	def testDecompressorChunksMaxsize(self):
735	n/a	bzd = BZ2Decompressor()
736	n/a	max_length = 100
737	n/a	out = []
738	n/a
739	n/a	# Feed some input
740	n/a	len_ = len(self.BIG_DATA) - 64
741	n/a	out.append(bzd.decompress(self.BIG_DATA[:len_],
742	n/a	max_length=max_length))
743	n/a	self.assertFalse(bzd.needs_input)
744	n/a	self.assertEqual(len(out[-1]), max_length)
745	n/a
746	n/a	# Retrieve more data without providing more input
747	n/a	out.append(bzd.decompress(b'', max_length=max_length))
748	n/a	self.assertFalse(bzd.needs_input)
749	n/a	self.assertEqual(len(out[-1]), max_length)
750	n/a
751	n/a	# Retrieve more data while providing more input
752	n/a	out.append(bzd.decompress(self.BIG_DATA[len_:],
753	n/a	max_length=max_length))
754	n/a	self.assertLessEqual(len(out[-1]), max_length)
755	n/a
756	n/a	# Retrieve remaining uncompressed data
757	n/a	while not bzd.eof:
758	n/a	out.append(bzd.decompress(b'', max_length=max_length))
759	n/a	self.assertLessEqual(len(out[-1]), max_length)
760	n/a
761	n/a	out = b"".join(out)
762	n/a	self.assertEqual(out, self.BIG_TEXT)
763	n/a	self.assertEqual(bzd.unused_data, b"")
764	n/a
765	n/a	def test_decompressor_inputbuf_1(self):
766	n/a	# Test reusing input buffer after moving existing
767	n/a	# contents to beginning
768	n/a	bzd = BZ2Decompressor()
769	n/a	out = []
770	n/a
771	n/a	# Create input buffer and fill it
772	n/a	self.assertEqual(bzd.decompress(self.DATA[:100],
773	n/a	max_length=0), b'')
774	n/a
775	n/a	# Retrieve some results, freeing capacity at beginning
776	n/a	# of input buffer
777	n/a	out.append(bzd.decompress(b'', 2))
778	n/a
779	n/a	# Add more data that fits into input buffer after
780	n/a	# moving existing data to beginning
781	n/a	out.append(bzd.decompress(self.DATA[100:105], 15))
782	n/a
783	n/a	# Decompress rest of data
784	n/a	out.append(bzd.decompress(self.DATA[105:]))
785	n/a	self.assertEqual(b''.join(out), self.TEXT)
786	n/a
787	n/a	def test_decompressor_inputbuf_2(self):
788	n/a	# Test reusing input buffer by appending data at the
789	n/a	# end right away
790	n/a	bzd = BZ2Decompressor()
791	n/a	out = []
792	n/a
793	n/a	# Create input buffer and empty it
794	n/a	self.assertEqual(bzd.decompress(self.DATA[:200],
795	n/a	max_length=0), b'')
796	n/a	out.append(bzd.decompress(b''))
797	n/a
798	n/a	# Fill buffer with new data
799	n/a	out.append(bzd.decompress(self.DATA[200:280], 2))
800	n/a
801	n/a	# Append some more data, not enough to require resize
802	n/a	out.append(bzd.decompress(self.DATA[280:300], 2))
803	n/a
804	n/a	# Decompress rest of data
805	n/a	out.append(bzd.decompress(self.DATA[300:]))
806	n/a	self.assertEqual(b''.join(out), self.TEXT)
807	n/a
808	n/a	def test_decompressor_inputbuf_3(self):
809	n/a	# Test reusing input buffer after extending it
810	n/a
811	n/a	bzd = BZ2Decompressor()
812	n/a	out = []
813	n/a
814	n/a	# Create almost full input buffer
815	n/a	out.append(bzd.decompress(self.DATA[:200], 5))
816	n/a
817	n/a	# Add even more data to it, requiring resize
818	n/a	out.append(bzd.decompress(self.DATA[200:300], 5))
819	n/a
820	n/a	# Decompress rest of data
821	n/a	out.append(bzd.decompress(self.DATA[300:]))
822	n/a	self.assertEqual(b''.join(out), self.TEXT)
823	n/a
824	n/a	def test_failure(self):
825	n/a	bzd = BZ2Decompressor()
826	n/a	self.assertRaises(Exception, bzd.decompress, self.BAD_DATA * 30)
827	n/a	# Previously, a second call could crash due to internal inconsistency
828	n/a	self.assertRaises(Exception, bzd.decompress, self.BAD_DATA * 30)
829	n/a
830	n/a	class CompressDecompressTest(BaseTest):
831	n/a	def testCompress(self):
832	n/a	data = bz2.compress(self.TEXT)
833	n/a	self.assertEqual(ext_decompress(data), self.TEXT)
834	n/a
835	n/a	def testCompressEmptyString(self):
836	n/a	text = bz2.compress(b'')
837	n/a	self.assertEqual(text, self.EMPTY_DATA)
838	n/a
839	n/a	def testDecompress(self):
840	n/a	text = bz2.decompress(self.DATA)
841	n/a	self.assertEqual(text, self.TEXT)
842	n/a
843	n/a	def testDecompressEmpty(self):
844	n/a	text = bz2.decompress(b"")
845	n/a	self.assertEqual(text, b"")
846	n/a
847	n/a	def testDecompressToEmptyString(self):
848	n/a	text = bz2.decompress(self.EMPTY_DATA)
849	n/a	self.assertEqual(text, b'')
850	n/a
851	n/a	def testDecompressIncomplete(self):
852	n/a	self.assertRaises(ValueError, bz2.decompress, self.DATA[:-10])
853	n/a
854	n/a	def testDecompressBadData(self):
855	n/a	self.assertRaises(OSError, bz2.decompress, self.BAD_DATA)
856	n/a
857	n/a	def testDecompressMultiStream(self):
858	n/a	text = bz2.decompress(self.DATA * 5)
859	n/a	self.assertEqual(text, self.TEXT * 5)
860	n/a
861	n/a	def testDecompressTrailingJunk(self):
862	n/a	text = bz2.decompress(self.DATA + self.BAD_DATA)
863	n/a	self.assertEqual(text, self.TEXT)
864	n/a
865	n/a	def testDecompressMultiStreamTrailingJunk(self):
866	n/a	text = bz2.decompress(self.DATA * 5 + self.BAD_DATA)
867	n/a	self.assertEqual(text, self.TEXT * 5)
868	n/a
869	n/a
870	n/a	class OpenTest(BaseTest):
871	n/a	"Test the open function."
872	n/a
873	n/a	def open(self, args, *kwargs):
874	n/a	return bz2.open(args, *kwargs)
875	n/a
876	n/a	def test_binary_modes(self):
877	n/a	for mode in ("wb", "xb"):
878	n/a	if mode == "xb":
879	n/a	unlink(self.filename)
880	n/a	with self.open(self.filename, mode) as f:
881	n/a	f.write(self.TEXT)
882	n/a	with open(self.filename, "rb") as f:
883	n/a	file_data = ext_decompress(f.read())
884	n/a	self.assertEqual(file_data, self.TEXT)
885	n/a	with self.open(self.filename, "rb") as f:
886	n/a	self.assertEqual(f.read(), self.TEXT)
887	n/a	with self.open(self.filename, "ab") as f:
888	n/a	f.write(self.TEXT)
889	n/a	with open(self.filename, "rb") as f:
890	n/a	file_data = ext_decompress(f.read())
891	n/a	self.assertEqual(file_data, self.TEXT * 2)
892	n/a
893	n/a	def test_implicit_binary_modes(self):
894	n/a	# Test implicit binary modes (no "b" or "t" in mode string).
895	n/a	for mode in ("w", "x"):
896	n/a	if mode == "x":
897	n/a	unlink(self.filename)
898	n/a	with self.open(self.filename, mode) as f:
899	n/a	f.write(self.TEXT)
900	n/a	with open(self.filename, "rb") as f:
901	n/a	file_data = ext_decompress(f.read())
902	n/a	self.assertEqual(file_data, self.TEXT)
903	n/a	with self.open(self.filename, "r") as f:
904	n/a	self.assertEqual(f.read(), self.TEXT)
905	n/a	with self.open(self.filename, "a") as f:
906	n/a	f.write(self.TEXT)
907	n/a	with open(self.filename, "rb") as f:
908	n/a	file_data = ext_decompress(f.read())
909	n/a	self.assertEqual(file_data, self.TEXT * 2)
910	n/a
911	n/a	def test_text_modes(self):
912	n/a	text = self.TEXT.decode("ascii")
913	n/a	text_native_eol = text.replace("\n", os.linesep)
914	n/a	for mode in ("wt", "xt"):
915	n/a	if mode == "xt":
916	n/a	unlink(self.filename)
917	n/a	with self.open(self.filename, mode) as f:
918	n/a	f.write(text)
919	n/a	with open(self.filename, "rb") as f:
920	n/a	file_data = ext_decompress(f.read()).decode("ascii")
921	n/a	self.assertEqual(file_data, text_native_eol)
922	n/a	with self.open(self.filename, "rt") as f:
923	n/a	self.assertEqual(f.read(), text)
924	n/a	with self.open(self.filename, "at") as f:
925	n/a	f.write(text)
926	n/a	with open(self.filename, "rb") as f:
927	n/a	file_data = ext_decompress(f.read()).decode("ascii")
928	n/a	self.assertEqual(file_data, text_native_eol * 2)
929	n/a
930	n/a	def test_x_mode(self):
931	n/a	for mode in ("x", "xb", "xt"):
932	n/a	unlink(self.filename)
933	n/a	with self.open(self.filename, mode) as f:
934	n/a	pass
935	n/a	with self.assertRaises(FileExistsError):
936	n/a	with self.open(self.filename, mode) as f:
937	n/a	pass
938	n/a
939	n/a	def test_fileobj(self):
940	n/a	with self.open(BytesIO(self.DATA), "r") as f:
941	n/a	self.assertEqual(f.read(), self.TEXT)
942	n/a	with self.open(BytesIO(self.DATA), "rb") as f:
943	n/a	self.assertEqual(f.read(), self.TEXT)
944	n/a	text = self.TEXT.decode("ascii")
945	n/a	with self.open(BytesIO(self.DATA), "rt") as f:
946	n/a	self.assertEqual(f.read(), text)
947	n/a
948	n/a	def test_bad_params(self):
949	n/a	# Test invalid parameter combinations.
950	n/a	self.assertRaises(ValueError,
951	n/a	self.open, self.filename, "wbt")
952	n/a	self.assertRaises(ValueError,
953	n/a	self.open, self.filename, "xbt")
954	n/a	self.assertRaises(ValueError,
955	n/a	self.open, self.filename, "rb", encoding="utf-8")
956	n/a	self.assertRaises(ValueError,
957	n/a	self.open, self.filename, "rb", errors="ignore")
958	n/a	self.assertRaises(ValueError,
959	n/a	self.open, self.filename, "rb", newline="\n")
960	n/a
961	n/a	def test_encoding(self):
962	n/a	# Test non-default encoding.
963	n/a	text = self.TEXT.decode("ascii")
964	n/a	text_native_eol = text.replace("\n", os.linesep)
965	n/a	with self.open(self.filename, "wt", encoding="utf-16-le") as f:
966	n/a	f.write(text)
967	n/a	with open(self.filename, "rb") as f:
968	n/a	file_data = ext_decompress(f.read()).decode("utf-16-le")
969	n/a	self.assertEqual(file_data, text_native_eol)
970	n/a	with self.open(self.filename, "rt", encoding="utf-16-le") as f:
971	n/a	self.assertEqual(f.read(), text)
972	n/a
973	n/a	def test_encoding_error_handler(self):
974	n/a	# Test with non-default encoding error handler.
975	n/a	with self.open(self.filename, "wb") as f:
976	n/a	f.write(b"foo\xffbar")
977	n/a	with self.open(self.filename, "rt", encoding="ascii", errors="ignore") \
978	n/a	as f:
979	n/a	self.assertEqual(f.read(), "foobar")
980	n/a
981	n/a	def test_newline(self):
982	n/a	# Test with explicit newline (universal newline mode disabled).
983	n/a	text = self.TEXT.decode("ascii")
984	n/a	with self.open(self.filename, "wt", newline="\n") as f:
985	n/a	f.write(text)
986	n/a	with self.open(self.filename, "rt", newline="\r") as f:
987	n/a	self.assertEqual(f.readlines(), [text])
988	n/a
989	n/a
990	n/a	def test_main():
991	n/a	support.run_unittest(
992	n/a	BZ2FileTest,
993	n/a	BZ2CompressorTest,
994	n/a	BZ2DecompressorTest,
995	n/a	CompressDecompressTest,
996	n/a	OpenTest,
997	n/a	)
998	n/a	support.reap_children()
999	n/a
1000	n/a	if __name__ == '__main__':
1001	n/a	test_main()