def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
"""Open lzma compressed tar archive name for reading or writing.
Appending is not allowed.
"""
if mode not in ("r", "w"):
raise ValueError("mode must be 'r' or 'w'")
try:
import lzma
except ImportError:
raise CompressionError("lzma module is not available")
fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)
except (lzma.LZMAError, EOFError):
fileobj.close()
raise ReadError("not an lzma file")
t._extfileobj = False
return t
# All *open() methods are registered here.
python类LZMAError()的实例源码
def test_init_bad_check(self):
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", check=b"asd")
# CHECK_UNKNOWN and anything above CHECK_ID_MAX should be invalid.
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", check=lzma.CHECK_UNKNOWN)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", check=lzma.CHECK_ID_MAX + 3)
# Cannot specify a check with mode="r".
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_NONE)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_CRC32)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_CRC64)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_SHA256)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_UNKNOWN)
def test_init_bad_preset(self):
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", preset=4.39)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", preset=10)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", preset=23)
with self.assertRaises(OverflowError):
LZMAFile(BytesIO(), "w", preset=-1)
with self.assertRaises(OverflowError):
LZMAFile(BytesIO(), "w", preset=-7)
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", preset="foo")
# Cannot specify a preset with mode="r".
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), preset=3)
def test__encode_filter_properties(self):
with self.assertRaises(TypeError):
lzma._encode_filter_properties(b"not a dict")
with self.assertRaises(ValueError):
lzma._encode_filter_properties({"id": 0x100})
with self.assertRaises(ValueError):
lzma._encode_filter_properties({"id": lzma.FILTER_LZMA2, "junk": 12})
with self.assertRaises(lzma.LZMAError):
lzma._encode_filter_properties({"id": lzma.FILTER_DELTA,
"dist": 9001})
# Test with parameters used by zipfile module.
props = lzma._encode_filter_properties({
"id": lzma.FILTER_LZMA1,
"pb": 2,
"lp": 0,
"lc": 3,
"dict_size": 8 << 20,
})
self.assertEqual(props, b"]\x00\x00\x80\x00")
def test_init_bad_check(self):
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", check=b"asd")
# CHECK_UNKNOWN and anything above CHECK_ID_MAX should be invalid.
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", check=lzma.CHECK_UNKNOWN)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", check=lzma.CHECK_ID_MAX + 3)
# Cannot specify a check with mode="r".
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_NONE)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_CRC32)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_CRC64)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_SHA256)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_UNKNOWN)
def test_init_bad_preset(self):
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", preset=4.39)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", preset=10)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", preset=23)
with self.assertRaises(OverflowError):
LZMAFile(BytesIO(), "w", preset=-1)
with self.assertRaises(OverflowError):
LZMAFile(BytesIO(), "w", preset=-7)
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", preset="foo")
# Cannot specify a preset with mode="r".
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), preset=3)
def test__encode_filter_properties(self):
with self.assertRaises(TypeError):
lzma._encode_filter_properties(b"not a dict")
with self.assertRaises(ValueError):
lzma._encode_filter_properties({"id": 0x100})
with self.assertRaises(ValueError):
lzma._encode_filter_properties({"id": lzma.FILTER_LZMA2, "junk": 12})
with self.assertRaises(lzma.LZMAError):
lzma._encode_filter_properties({"id": lzma.FILTER_DELTA,
"dist": 9001})
# Test with parameters used by zipfile module.
props = lzma._encode_filter_properties({
"id": lzma.FILTER_LZMA1,
"pb": 2,
"lp": 0,
"lc": 3,
"dict_size": 8 << 20,
})
self.assertEqual(props, b"]\x00\x00\x80\x00")
def decompress_lzma(data):
results = []
len(data)
while True:
decomp = LZMADecompressor(FORMAT_AUTO, None, None)
try:
res = decomp.decompress(data)
except LZMAError:
if results:
break # Leftover data is not a valid LZMA/XZ stream; ignore it.
else:
raise # Error on the first iteration; bail out.
results.append(res)
data = decomp.unused_data
if not data:
break
if not decomp.eof:
raise LZMAError("Compressed data ended before the end-of-stream marker was reached")
return b"".join(results)
def test_init_bad_check(self):
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", check=b"asd")
# CHECK_UNKNOWN and anything above CHECK_ID_MAX should be invalid.
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", check=lzma.CHECK_UNKNOWN)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", check=lzma.CHECK_ID_MAX + 3)
# Cannot specify a check with mode="r".
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_NONE)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_CRC32)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_CRC64)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_SHA256)
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), check=lzma.CHECK_UNKNOWN)
def test_init_bad_preset(self):
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", preset=4.39)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", preset=10)
with self.assertRaises(LZMAError):
LZMAFile(BytesIO(), "w", preset=23)
with self.assertRaises(OverflowError):
LZMAFile(BytesIO(), "w", preset=-1)
with self.assertRaises(OverflowError):
LZMAFile(BytesIO(), "w", preset=-7)
with self.assertRaises(TypeError):
LZMAFile(BytesIO(), "w", preset="foo")
# Cannot specify a preset with mode="r".
with self.assertRaises(ValueError):
LZMAFile(BytesIO(COMPRESSED_XZ), preset=3)
def test__encode_filter_properties(self):
with self.assertRaises(TypeError):
lzma._encode_filter_properties(b"not a dict")
with self.assertRaises(ValueError):
lzma._encode_filter_properties({"id": 0x100})
with self.assertRaises(ValueError):
lzma._encode_filter_properties({"id": lzma.FILTER_LZMA2, "junk": 12})
with self.assertRaises(lzma.LZMAError):
lzma._encode_filter_properties({"id": lzma.FILTER_DELTA,
"dist": 9001})
# Test with parameters used by zipfile module.
props = lzma._encode_filter_properties({
"id": lzma.FILTER_LZMA1,
"pb": 2,
"lp": 0,
"lc": 3,
"dict_size": 8 << 20,
})
self.assertEqual(props, b"]\x00\x00\x80\x00")
def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
"""Open lzma compressed tar archive name for reading or writing.
Appending is not allowed.
"""
if mode not in ("r", "w", "x"):
raise ValueError("mode must be 'r', 'w' or 'x'")
try:
import lzma
except ImportError:
raise CompressionError("lzma module is not available")
fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)
except (lzma.LZMAError, EOFError):
fileobj.close()
if mode == 'r':
raise ReadError("not an lzma file")
raise
except:
fileobj.close()
raise
t._extfileobj = False
return t
# All *open() methods are registered here.
def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
"""Open lzma compressed tar archive name for reading or writing.
Appending is not allowed.
"""
if mode not in ("r", "w", "x"):
raise ValueError("mode must be 'r', 'w' or 'x'")
try:
import lzma
except ImportError:
raise CompressionError("lzma module is not available")
fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)
except (lzma.LZMAError, EOFError):
fileobj.close()
if mode == 'r':
raise ReadError("not an lzma file")
raise
except:
fileobj.close()
raise
t._extfileobj = False
return t
# All *open() methods are registered here.
def verify_contents(thefile, tgt_hostname=None, callback=None):
"""
Given a sysstat binary data file verify that it contains a set of well
formed data values.
The optional 'tgt_hostname' argument is checked against the file header's
stored hostname value.
The optional 'callback' argument, if provided, should be an instance of
the ContentAction class, where for each magic structure, file header, file
activity set, record header and record payload read the appropriate method
will be invoked, with the 'eof' method invoked at the end.
One of the following exceptions will be raised if a problem is found with
the file:
Invalid: The file header or record header metadata values do not make
sense in relation to each other
Corruption: The file appears to be corrupted in some way
Truncated: The file does not appear to contain all the data as
described by the file header or a given record header
"""
try:
with lzma.open(thefile, "rb") as fp:
verify_contents_fp(fp, tgt_hostname, callback)
except lzma.LZMAError:
with open(thefile, "rb") as fp:
verify_contents_fp(fp, tgt_hostname, callback)
def fetch_fileheader(thefile):
"""
Fetch the sysstat FileHeader object for the given file path.
"""
try:
with lzma.open(thefile, "rb") as fp:
res = fetch_fileheader_with_fp(fp)
except lzma.LZMAError:
with open(thefile, "rb") as fp:
res = fetch_fileheader_with_fp(fp)
return res
def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
"""Open lzma compressed tar archive name for reading or writing.
Appending is not allowed.
"""
if mode not in ("r", "w", "x"):
raise ValueError("mode must be 'r', 'w' or 'x'")
try:
import lzma
except ImportError:
raise CompressionError("lzma module is not available")
fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)
except (lzma.LZMAError, EOFError):
fileobj.close()
if mode == 'r':
raise ReadError("not an lzma file")
raise
except:
fileobj.close()
raise
t._extfileobj = False
return t
# All *open() methods are registered here.
def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
"""Open lzma compressed tar archive name for reading or writing.
Appending is not allowed.
"""
if mode not in ("r", "w", "x"):
raise ValueError("mode must be 'r', 'w' or 'x'")
try:
import lzma
except ImportError:
raise CompressionError("lzma module is not available")
fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)
except (lzma.LZMAError, EOFError):
fileobj.close()
if mode == 'r':
raise ReadError("not an lzma file")
raise
except:
fileobj.close()
raise
t._extfileobj = False
return t
# All *open() methods are registered here.
def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
"""Open lzma compressed tar archive name for reading or writing.
Appending is not allowed.
"""
if mode not in ("r", "w", "x"):
raise ValueError("mode must be 'r', 'w' or 'x'")
try:
import lzma
except ImportError:
raise CompressionError("lzma module is not available")
fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)
except (lzma.LZMAError, EOFError):
fileobj.close()
if mode == 'r':
raise ReadError("not an lzma file")
raise
except:
fileobj.close()
raise
t._extfileobj = False
return t
# All *open() methods are registered here.
def test_decompressor_memlimit(self):
lzd = LZMADecompressor(memlimit=1024)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)
lzd = LZMADecompressor(lzma.FORMAT_XZ, memlimit=1024)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)
lzd = LZMADecompressor(lzma.FORMAT_ALONE, memlimit=1024)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_ALONE)
# Test LZMADecompressor on known-good input data.
def test_decompressor_bad_input(self):
lzd = LZMADecompressor()
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_RAW_1)
lzd = LZMADecompressor(lzma.FORMAT_XZ)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_ALONE)
lzd = LZMADecompressor(lzma.FORMAT_ALONE)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)
lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_1)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)
# Test that LZMACompressor->LZMADecompressor preserves the input data.
def test_decompress_memlimit(self):
with self.assertRaises(LZMAError):
lzma.decompress(COMPRESSED_XZ, memlimit=1024)
with self.assertRaises(LZMAError):
lzma.decompress(
COMPRESSED_XZ, format=lzma.FORMAT_XZ, memlimit=1024)
with self.assertRaises(LZMAError):
lzma.decompress(
COMPRESSED_ALONE, format=lzma.FORMAT_ALONE, memlimit=1024)
# Test LZMADecompressor on known-good input data.
def test_decompress_incomplete_input(self):
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_XZ[:128])
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_ALONE[:128])
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_1[:128],
format=lzma.FORMAT_RAW, filters=FILTERS_RAW_1)
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_2[:128],
format=lzma.FORMAT_RAW, filters=FILTERS_RAW_2)
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_3[:128],
format=lzma.FORMAT_RAW, filters=FILTERS_RAW_3)
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_4[:128],
format=lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
def test__decode_filter_properties(self):
with self.assertRaises(TypeError):
lzma._decode_filter_properties(lzma.FILTER_X86, {"should be": bytes})
with self.assertRaises(lzma.LZMAError):
lzma._decode_filter_properties(lzma.FILTER_DELTA, b"too long")
# Test with parameters used by zipfile module.
filterspec = lzma._decode_filter_properties(
lzma.FILTER_LZMA1, b"]\x00\x00\x80\x00")
self.assertEqual(filterspec["id"], lzma.FILTER_LZMA1)
self.assertEqual(filterspec["pb"], 2)
self.assertEqual(filterspec["lp"], 0)
self.assertEqual(filterspec["lc"], 3)
self.assertEqual(filterspec["dict_size"], 8 << 20)
def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
"""Open lzma compressed tar archive name for reading or writing.
Appending is not allowed.
"""
if mode not in ("r", "w", "x"):
raise ValueError("mode must be 'r', 'w' or 'x'")
try:
import lzma
except ImportError:
raise CompressionError("lzma module is not available")
fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)
except (lzma.LZMAError, EOFError):
fileobj.close()
if mode == 'r':
raise ReadError("not an lzma file")
raise
except:
fileobj.close()
raise
t._extfileobj = False
return t
# All *open() methods are registered here.
def xzopen(cls, name, mode="r", fileobj=None, preset=None, **kwargs):
"""Open lzma compressed tar archive name for reading or writing.
Appending is not allowed.
"""
if mode not in ("r", "w"):
raise ValueError("mode must be 'r' or 'w'")
try:
import lzma
except ImportError:
raise CompressionError("lzma module is not available")
fileobj = lzma.LZMAFile(fileobj or name, mode, preset=preset)
try:
t = cls.taropen(name, mode, fileobj, **kwargs)
except (lzma.LZMAError, EOFError):
fileobj.close()
if mode == 'r':
raise ReadError("not an lzma file")
raise
except:
fileobj.close()
raise
t._extfileobj = False
return t
# All *open() methods are registered here.
def test_decompressor_memlimit(self):
lzd = LZMADecompressor(memlimit=1024)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)
lzd = LZMADecompressor(lzma.FORMAT_XZ, memlimit=1024)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)
lzd = LZMADecompressor(lzma.FORMAT_ALONE, memlimit=1024)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_ALONE)
# Test LZMADecompressor on known-good input data.
def test_decompressor_bad_input(self):
lzd = LZMADecompressor()
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_RAW_1)
lzd = LZMADecompressor(lzma.FORMAT_XZ)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_ALONE)
lzd = LZMADecompressor(lzma.FORMAT_ALONE)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)
lzd = LZMADecompressor(lzma.FORMAT_RAW, filters=FILTERS_RAW_1)
self.assertRaises(LZMAError, lzd.decompress, COMPRESSED_XZ)
# Test that LZMACompressor->LZMADecompressor preserves the input data.
def test_decompress_memlimit(self):
with self.assertRaises(LZMAError):
lzma.decompress(COMPRESSED_XZ, memlimit=1024)
with self.assertRaises(LZMAError):
lzma.decompress(
COMPRESSED_XZ, format=lzma.FORMAT_XZ, memlimit=1024)
with self.assertRaises(LZMAError):
lzma.decompress(
COMPRESSED_ALONE, format=lzma.FORMAT_ALONE, memlimit=1024)
# Test LZMADecompressor on known-good input data.
def test_decompress_incomplete_input(self):
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_XZ[:128])
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_ALONE[:128])
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_1[:128],
format=lzma.FORMAT_RAW, filters=FILTERS_RAW_1)
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_2[:128],
format=lzma.FORMAT_RAW, filters=FILTERS_RAW_2)
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_3[:128],
format=lzma.FORMAT_RAW, filters=FILTERS_RAW_3)
self.assertRaises(LZMAError, lzma.decompress, COMPRESSED_RAW_4[:128],
format=lzma.FORMAT_RAW, filters=FILTERS_RAW_4)
def test_read_bad_data(self):
with LZMAFile(BytesIO(COMPRESSED_BOGUS)) as f:
self.assertRaises(LZMAError, f.read)