]> git.proxmox.com Git - mirror_edk2.git/blame - AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_bz2.py
projects / mirror_edk2.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
EmbeddedPkg: Extend NvVarStoreFormattedLib LIBRARY_CLASS
[mirror_edk2.git] / AppPkg / Applications / Python / Python-2.7.2 / Lib / test / test_bz2.py
CommitLineData
4710c53d 1#!/usr/bin/env python\r
2from test import test_support\r
3from test.test_support import TESTFN, import_module\r
4\r
5import unittest\r
6from cStringIO import StringIO\r
7import os\r
8import subprocess\r
9import sys\r
10\r
11try:\r
12 import threading\r
13except ImportError:\r
14 threading = None\r
15\r
16bz2 = import_module('bz2')\r
17from bz2 import BZ2File, BZ2Compressor, BZ2Decompressor\r
18\r
19has_cmdline_bunzip2 = sys.platform not in ("win32", "os2emx", "riscos")\r
20\r
21class BaseTest(unittest.TestCase):\r
22 "Base for other testcases."\r
23 TEXT = 'root:x:0:0:root:/root:/bin/bash\nbin:x:1:1:bin:/bin:\ndaemon:x:2:2:daemon:/sbin:\nadm:x:3:4:adm:/var/adm:\nlp:x:4:7:lp:/var/spool/lpd:\nsync:x:5:0:sync:/sbin:/bin/sync\nshutdown:x:6:0:shutdown:/sbin:/sbin/shutdown\nhalt:x:7:0:halt:/sbin:/sbin/halt\nmail:x:8:12:mail:/var/spool/mail:\nnews:x:9:13:news:/var/spool/news:\nuucp:x:10:14:uucp:/var/spool/uucp:\noperator:x:11:0:operator:/root:\ngames:x:12:100:games:/usr/games:\ngopher:x:13:30:gopher:/usr/lib/gopher-data:\nftp:x:14:50:FTP User:/var/ftp:/bin/bash\nnobody:x:65534:65534:Nobody:/home:\npostfix:x:100:101:postfix:/var/spool/postfix:\nniemeyer:x:500:500::/home/niemeyer:/bin/bash\npostgres:x:101:102:PostgreSQL Server:/var/lib/pgsql:/bin/bash\nmysql:x:102:103:MySQL server:/var/lib/mysql:/bin/bash\nwww:x:103:104::/var/www:/bin/false\n'\r
24 DATA = '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`'\r
25 DATA_CRLF = 'BZh91AY&SY\xaez\xbbN\x00\x01H\xdf\x80\x00\x12@\x02\xff\xf0\x01\x07n\x00?\xe7\xff\xe0@\x01\xbc\xc6`\x86*\x8d=M\xa9\x9a\x86\xd0L@\x0fI\xa6!\xa1\x13\xc8\x88jdi\x8d@\x03@\x1a\x1a\x0c\x0c\x83 \x00\xc4h2\x19\x01\x82D\x84e\t\xe8\x99\x89\x19\x1ah\x00\r\x1a\x11\xaf\x9b\x0fG\xf5(\x1b\x1f?\t\x12\xcf\xb5\xfc\x95E\x00ps\x89\x12^\xa4\xdd\xa2&\x05(\x87\x04\x98\x89u\xe40%\xb6\x19\'\x8c\xc4\x89\xca\x07\x0e\x1b!\x91UIFU%C\x994!DI\xd2\xfa\xf0\xf1N8W\xde\x13A\xf5\x9cr%?\x9f3;I45A\xd1\x8bT\xb1<l\xba\xcb_\xc00xY\x17r\x17\x88\x08\x08@\xa0\ry@\x10\x04$)`\xf2\xce\x89z\xb0s\xec\x9b.iW\x9d\x81\xb5-+t\x9f\x1a\'\x97dB\xf5x\xb5\xbe.[.\xd7\x0e\x81\xe7\x08\x1cN`\x88\x10\xca\x87\xc3!"\x80\x92R\xa1/\xd1\xc0\xe6mf\xac\xbd\x99\xcca\xb3\x8780>\xa4\xc7\x8d\x1a\\"\xad\xa1\xabyBg\x15\xb9l\x88\x88\x91k"\x94\xa4\xd4\x89\xae*\xa6\x0b\x10\x0c\xd6\xd4m\xe86\xec\xb5j\x8a\x86j\';\xca.\x01I\xf2\xaaJ\xe8\x88\x8cU+t3\xfb\x0c\n\xa33\x13r2\r\x16\xe0\xb3(\xbf\x1d\x83r\xe7M\xf0D\x1365\xd8\x88\xd3\xa4\x92\xcb2\x06\x04\\\xc1\xb0\xea//\xbek&\xd8\xe6+t\xe5\xa1\x13\xada\x16\xder5"w]\xa2i\xb7[\x97R \xe2IT\xcd;Z\x04dk4\xad\x8a\t\xd3\x81z\x10\xf1:^`\xab\x1f\xc5\xdc\x91N\x14$+\x9e\xae\xd3\x80'\r
26\r
27 if has_cmdline_bunzip2:\r
28 def decompress(self, data):\r
29 pop = subprocess.Popen("bunzip2", shell=True,\r
30 stdin=subprocess.PIPE,\r
31 stdout=subprocess.PIPE,\r
32 stderr=subprocess.STDOUT)\r
33 pop.stdin.write(data)\r
34 pop.stdin.close()\r
35 ret = pop.stdout.read()\r
36 pop.stdout.close()\r
37 if pop.wait() != 0:\r
38 ret = bz2.decompress(data)\r
39 return ret\r
40\r
41 else:\r
42 # bunzip2 isn't available to run on Windows.\r
43 def decompress(self, data):\r
44 return bz2.decompress(data)\r
45\r
46class BZ2FileTest(BaseTest):\r
47 "Test BZ2File type miscellaneous methods."\r
48\r
49 def setUp(self):\r
50 self.filename = TESTFN\r
51\r
52 def tearDown(self):\r
53 if os.path.isfile(self.filename):\r
54 os.unlink(self.filename)\r
55\r
56 def createTempFile(self, crlf=0):\r
57 with open(self.filename, "wb") as f:\r
58 if crlf:\r
59 data = self.DATA_CRLF\r
60 else:\r
61 data = self.DATA\r
62 f.write(data)\r
63\r
64 def testRead(self):\r
65 # "Test BZ2File.read()"\r
66 self.createTempFile()\r
67 with BZ2File(self.filename) as bz2f:\r
68 self.assertRaises(TypeError, bz2f.read, None)\r
69 self.assertEqual(bz2f.read(), self.TEXT)\r
70\r
71 def testRead0(self):\r
72 # Test BBZ2File.read(0)"\r
73 self.createTempFile()\r
74 with BZ2File(self.filename) as bz2f:\r
75 self.assertRaises(TypeError, bz2f.read, None)\r
76 self.assertEqual(bz2f.read(0), "")\r
77\r
78 def testReadChunk10(self):\r
79 # "Test BZ2File.read() in chunks of 10 bytes"\r
80 self.createTempFile()\r
81 with BZ2File(self.filename) as bz2f:\r
82 text = ''\r
83 while 1:\r
84 str = bz2f.read(10)\r
85 if not str:\r
86 break\r
87 text += str\r
88 self.assertEqual(text, self.TEXT)\r
89\r
90 def testRead100(self):\r
91 # "Test BZ2File.read(100)"\r
92 self.createTempFile()\r
93 with BZ2File(self.filename) as bz2f:\r
94 self.assertEqual(bz2f.read(100), self.TEXT[:100])\r
95\r
96 def testReadLine(self):\r
97 # "Test BZ2File.readline()"\r
98 self.createTempFile()\r
99 with BZ2File(self.filename) as bz2f:\r
100 self.assertRaises(TypeError, bz2f.readline, None)\r
101 sio = StringIO(self.TEXT)\r
102 for line in sio.readlines():\r
103 self.assertEqual(bz2f.readline(), line)\r
104\r
105 def testReadLines(self):\r
106 # "Test BZ2File.readlines()"\r
107 self.createTempFile()\r
108 with BZ2File(self.filename) as bz2f:\r
109 self.assertRaises(TypeError, bz2f.readlines, None)\r
110 sio = StringIO(self.TEXT)\r
111 self.assertEqual(bz2f.readlines(), sio.readlines())\r
112\r
113 def testIterator(self):\r
114 # "Test iter(BZ2File)"\r
115 self.createTempFile()\r
116 with BZ2File(self.filename) as bz2f:\r
117 sio = StringIO(self.TEXT)\r
118 self.assertEqual(list(iter(bz2f)), sio.readlines())\r
119\r
120 def testClosedIteratorDeadlock(self):\r
121 # "Test that iteration on a closed bz2file releases the lock."\r
122 # http://bugs.python.org/issue3309\r
123 self.createTempFile()\r
124 bz2f = BZ2File(self.filename)\r
125 bz2f.close()\r
126 self.assertRaises(ValueError, bz2f.next)\r
127 # This call will deadlock of the above .next call failed to\r
128 # release the lock.\r
129 self.assertRaises(ValueError, bz2f.readlines)\r
130\r
131 def testXReadLines(self):\r
132 # "Test BZ2File.xreadlines()"\r
133 self.createTempFile()\r
134 bz2f = BZ2File(self.filename)\r
135 sio = StringIO(self.TEXT)\r
136 self.assertEqual(list(bz2f.xreadlines()), sio.readlines())\r
137 bz2f.close()\r
138\r
139 def testUniversalNewlinesLF(self):\r
140 # "Test BZ2File.read() with universal newlines (\\n)"\r
141 self.createTempFile()\r
142 bz2f = BZ2File(self.filename, "rU")\r
143 self.assertEqual(bz2f.read(), self.TEXT)\r
144 self.assertEqual(bz2f.newlines, "\n")\r
145 bz2f.close()\r
146\r
147 def testUniversalNewlinesCRLF(self):\r
148 # "Test BZ2File.read() with universal newlines (\\r\\n)"\r
149 self.createTempFile(crlf=1)\r
150 bz2f = BZ2File(self.filename, "rU")\r
151 self.assertEqual(bz2f.read(), self.TEXT)\r
152 self.assertEqual(bz2f.newlines, "\r\n")\r
153 bz2f.close()\r
154\r
155 def testWrite(self):\r
156 # "Test BZ2File.write()"\r
157 with BZ2File(self.filename, "w") as bz2f:\r
158 self.assertRaises(TypeError, bz2f.write)\r
159 bz2f.write(self.TEXT)\r
160 with open(self.filename, 'rb') as f:\r
161 self.assertEqual(self.decompress(f.read()), self.TEXT)\r
162\r
163 def testWriteChunks10(self):\r
164 # "Test BZ2File.write() with chunks of 10 bytes"\r
165 with BZ2File(self.filename, "w") as bz2f:\r
166 n = 0\r
167 while 1:\r
168 str = self.TEXT[n*10:(n+1)*10]\r
169 if not str:\r
170 break\r
171 bz2f.write(str)\r
172 n += 1\r
173 with open(self.filename, 'rb') as f:\r
174 self.assertEqual(self.decompress(f.read()), self.TEXT)\r
175\r
176 def testWriteLines(self):\r
177 # "Test BZ2File.writelines()"\r
178 with BZ2File(self.filename, "w") as bz2f:\r
179 self.assertRaises(TypeError, bz2f.writelines)\r
180 sio = StringIO(self.TEXT)\r
181 bz2f.writelines(sio.readlines())\r
182 # patch #1535500\r
183 self.assertRaises(ValueError, bz2f.writelines, ["a"])\r
184 with open(self.filename, 'rb') as f:\r
185 self.assertEqual(self.decompress(f.read()), self.TEXT)\r
186\r
187 def testWriteMethodsOnReadOnlyFile(self):\r
188 with BZ2File(self.filename, "w") as bz2f:\r
189 bz2f.write("abc")\r
190\r
191 with BZ2File(self.filename, "r") as bz2f:\r
192 self.assertRaises(IOError, bz2f.write, "a")\r
193 self.assertRaises(IOError, bz2f.writelines, ["a"])\r
194\r
195 def testSeekForward(self):\r
196 # "Test BZ2File.seek(150, 0)"\r
197 self.createTempFile()\r
198 with BZ2File(self.filename) as bz2f:\r
199 self.assertRaises(TypeError, bz2f.seek)\r
200 bz2f.seek(150)\r
201 self.assertEqual(bz2f.read(), self.TEXT[150:])\r
202\r
203 def testSeekBackwards(self):\r
204 # "Test BZ2File.seek(-150, 1)"\r
205 self.createTempFile()\r
206 with BZ2File(self.filename) as bz2f:\r
207 bz2f.read(500)\r
208 bz2f.seek(-150, 1)\r
209 self.assertEqual(bz2f.read(), self.TEXT[500-150:])\r
210\r
211 def testSeekBackwardsFromEnd(self):\r
212 # "Test BZ2File.seek(-150, 2)"\r
213 self.createTempFile()\r
214 with BZ2File(self.filename) as bz2f:\r
215 bz2f.seek(-150, 2)\r
216 self.assertEqual(bz2f.read(), self.TEXT[len(self.TEXT)-150:])\r
217\r
218 def testSeekPostEnd(self):\r
219 # "Test BZ2File.seek(150000)"\r
220 self.createTempFile()\r
221 with BZ2File(self.filename) as bz2f:\r
222 bz2f.seek(150000)\r
223 self.assertEqual(bz2f.tell(), len(self.TEXT))\r
224 self.assertEqual(bz2f.read(), "")\r
225\r
226 def testSeekPostEndTwice(self):\r
227 # "Test BZ2File.seek(150000) twice"\r
228 self.createTempFile()\r
229 with BZ2File(self.filename) as bz2f:\r
230 bz2f.seek(150000)\r
231 bz2f.seek(150000)\r
232 self.assertEqual(bz2f.tell(), len(self.TEXT))\r
233 self.assertEqual(bz2f.read(), "")\r
234\r
235 def testSeekPreStart(self):\r
236 # "Test BZ2File.seek(-150, 0)"\r
237 self.createTempFile()\r
238 with BZ2File(self.filename) as bz2f:\r
239 bz2f.seek(-150)\r
240 self.assertEqual(bz2f.tell(), 0)\r
241 self.assertEqual(bz2f.read(), self.TEXT)\r
242\r
243 def testOpenDel(self):\r
244 # "Test opening and deleting a file many times"\r
245 self.createTempFile()\r
246 for i in xrange(10000):\r
247 o = BZ2File(self.filename)\r
248 del o\r
249\r
250 def testOpenNonexistent(self):\r
251 # "Test opening a nonexistent file"\r
252 self.assertRaises(IOError, BZ2File, "/non/existent")\r
253\r
254 def testModeU(self):\r
255 # Bug #1194181: bz2.BZ2File opened for write with mode "U"\r
256 self.createTempFile()\r
257 bz2f = BZ2File(self.filename, "U")\r
258 bz2f.close()\r
259 f = file(self.filename)\r
260 f.seek(0, 2)\r
261 self.assertEqual(f.tell(), len(self.DATA))\r
262 f.close()\r
263\r
264 def testBug1191043(self):\r
265 # readlines() for files containing no newline\r
266 data = 'BZh91AY&SY\xd9b\x89]\x00\x00\x00\x03\x80\x04\x00\x02\x00\x0c\x00 \x00!\x9ah3M\x13<]\xc9\x14\xe1BCe\x8a%t'\r
267 with open(self.filename, "wb") as f:\r
268 f.write(data)\r
269 with BZ2File(self.filename) as bz2f:\r
270 lines = bz2f.readlines()\r
271 self.assertEqual(lines, ['Test'])\r
272 with BZ2File(self.filename) as bz2f:\r
273 xlines = list(bz2f.readlines())\r
274 self.assertEqual(xlines, ['Test'])\r
275\r
276 def testContextProtocol(self):\r
277 # BZ2File supports the context management protocol\r
278 f = None\r
279 with BZ2File(self.filename, "wb") as f:\r
280 f.write(b"xxx")\r
281 f = BZ2File(self.filename, "rb")\r
282 f.close()\r
283 try:\r
284 with f:\r
285 pass\r
286 except ValueError:\r
287 pass\r
288 else:\r
289 self.fail("__enter__ on a closed file didn't raise an exception")\r
290 try:\r
291 with BZ2File(self.filename, "wb") as f:\r
292 1 // 0\r
293 except ZeroDivisionError:\r
294 pass\r
295 else:\r
296 self.fail("1 // 0 didn't raise an exception")\r
297\r
298 @unittest.skipUnless(threading, 'Threading required for this test.')\r
299 def testThreading(self):\r
300 # Using a BZ2File from several threads doesn't deadlock (issue #7205).\r
301 data = "1" * 2**20\r
302 nthreads = 10\r
303 with bz2.BZ2File(self.filename, 'wb') as f:\r
304 def comp():\r
305 for i in range(5):\r
306 f.write(data)\r
307 threads = [threading.Thread(target=comp) for i in range(nthreads)]\r
308 for t in threads:\r
309 t.start()\r
310 for t in threads:\r
311 t.join()\r
312\r
313 def testMixedIterationReads(self):\r
314 # Issue #8397: mixed iteration and reads should be forbidden.\r
315 with bz2.BZ2File(self.filename, 'wb') as f:\r
316 # The internal buffer size is hard-wired to 8192 bytes, we must\r
317 # write out more than that for the test to stop half through\r
318 # the buffer.\r
319 f.write(self.TEXT * 100)\r
320 with bz2.BZ2File(self.filename, 'rb') as f:\r
321 next(f)\r
322 self.assertRaises(ValueError, f.read)\r
323 self.assertRaises(ValueError, f.readline)\r
324 self.assertRaises(ValueError, f.readlines)\r
325\r
326class BZ2CompressorTest(BaseTest):\r
327 def testCompress(self):\r
328 # "Test BZ2Compressor.compress()/flush()"\r
329 bz2c = BZ2Compressor()\r
330 self.assertRaises(TypeError, bz2c.compress)\r
331 data = bz2c.compress(self.TEXT)\r
332 data += bz2c.flush()\r
333 self.assertEqual(self.decompress(data), self.TEXT)\r
334\r
335 def testCompressChunks10(self):\r
336 # "Test BZ2Compressor.compress()/flush() with chunks of 10 bytes"\r
337 bz2c = BZ2Compressor()\r
338 n = 0\r
339 data = ''\r
340 while 1:\r
341 str = self.TEXT[n*10:(n+1)*10]\r
342 if not str:\r
343 break\r
344 data += bz2c.compress(str)\r
345 n += 1\r
346 data += bz2c.flush()\r
347 self.assertEqual(self.decompress(data), self.TEXT)\r
348\r
349class BZ2DecompressorTest(BaseTest):\r
350 def test_Constructor(self):\r
351 self.assertRaises(TypeError, BZ2Decompressor, 42)\r
352\r
353 def testDecompress(self):\r
354 # "Test BZ2Decompressor.decompress()"\r
355 bz2d = BZ2Decompressor()\r
356 self.assertRaises(TypeError, bz2d.decompress)\r
357 text = bz2d.decompress(self.DATA)\r
358 self.assertEqual(text, self.TEXT)\r
359\r
360 def testDecompressChunks10(self):\r
361 # "Test BZ2Decompressor.decompress() with chunks of 10 bytes"\r
362 bz2d = BZ2Decompressor()\r
363 text = ''\r
364 n = 0\r
365 while 1:\r
366 str = self.DATA[n*10:(n+1)*10]\r
367 if not str:\r
368 break\r
369 text += bz2d.decompress(str)\r
370 n += 1\r
371 self.assertEqual(text, self.TEXT)\r
372\r
373 def testDecompressUnusedData(self):\r
374 # "Test BZ2Decompressor.decompress() with unused data"\r
375 bz2d = BZ2Decompressor()\r
376 unused_data = "this is unused data"\r
377 text = bz2d.decompress(self.DATA+unused_data)\r
378 self.assertEqual(text, self.TEXT)\r
379 self.assertEqual(bz2d.unused_data, unused_data)\r
380\r
381 def testEOFError(self):\r
382 # "Calling BZ2Decompressor.decompress() after EOS must raise EOFError"\r
383 bz2d = BZ2Decompressor()\r
384 text = bz2d.decompress(self.DATA)\r
385 self.assertRaises(EOFError, bz2d.decompress, "anything")\r
386\r
387\r
388class FuncTest(BaseTest):\r
389 "Test module functions"\r
390\r
391 def testCompress(self):\r
392 # "Test compress() function"\r
393 data = bz2.compress(self.TEXT)\r
394 self.assertEqual(self.decompress(data), self.TEXT)\r
395\r
396 def testDecompress(self):\r
397 # "Test decompress() function"\r
398 text = bz2.decompress(self.DATA)\r
399 self.assertEqual(text, self.TEXT)\r
400\r
401 def testDecompressEmpty(self):\r
402 # "Test decompress() function with empty string"\r
403 text = bz2.decompress("")\r
404 self.assertEqual(text, "")\r
405\r
406 def testDecompressIncomplete(self):\r
407 # "Test decompress() function with incomplete data"\r
408 self.assertRaises(ValueError, bz2.decompress, self.DATA[:-10])\r
409\r
410def test_main():\r
411 test_support.run_unittest(\r
412 BZ2FileTest,\r
413 BZ2CompressorTest,\r
414 BZ2DecompressorTest,\r
415 FuncTest\r
416 )\r
417 test_support.reap_children()\r
418\r
419if __name__ == '__main__':\r
420 test_main()\r
421\r
422# vim:ts=4:sw=4\r
EFI Development Kit II mirror for PVE