Python tarfile 模块,StreamError() 实例源码

我们从Python开源项目中,提取了以下22个代码示例,用于说明如何使用tarfile.StreamError()

项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def test_compare_members(self):
        tar1 = tarfile.open(tarname, encoding="iso8859-1")
        try:
            tar2 = self.tar

            while True:
                t1 = tar1.next()
                t2 = tar2.next()
                if t1 is None:
                    break
                self.assertTrue(t2 is not None, "stream.next() failed.")

                if t2.islnk() or t2.issym():
                    self.assertRaises(tarfile.StreamError, tar2.extractfile, t2)
                    continue

                v1 = tar1.extractfile(t1)
                v2 = tar2.extractfile(t2)
                if v1 is None:
                    continue
                self.assertTrue(v2 is not None, "stream.extractfile() failed")
                self.assertEqual(v1.read(), v2.read(), "stream extraction failed")
        finally:
            tar1.close()
项目:oil    作者:oilshell    | 项目源码 | 文件源码
def test_compare_members(self):
        tar1 = tarfile.open(tarname, encoding="iso8859-1")
        try:
            tar2 = self.tar

            while True:
                t1 = tar1.next()
                t2 = tar2.next()
                if t1 is None:
                    break
                self.assertTrue(t2 is not None, "stream.next() failed.")

                if t2.islnk() or t2.issym():
                    self.assertRaises(tarfile.StreamError, tar2.extractfile, t2)
                    continue

                v1 = tar1.extractfile(t1)
                v2 = tar2.extractfile(t2)
                if v1 is None:
                    continue
                self.assertTrue(v2 is not None, "stream.extractfile() failed")
                self.assertTrue(v1.read() == v2.read(), "stream extraction failed")
        finally:
            tar1.close()
项目:python2-tracer    作者:extremecoders-re    | 项目源码 | 文件源码
def test_compare_members(self):
        tar1 = tarfile.open(tarname, encoding="iso8859-1")
        try:
            tar2 = self.tar

            while True:
                t1 = tar1.next()
                t2 = tar2.next()
                if t1 is None:
                    break
                self.assertTrue(t2 is not None, "stream.next() failed.")

                if t2.islnk() or t2.issym():
                    self.assertRaises(tarfile.StreamError, tar2.extractfile, t2)
                    continue

                v1 = tar1.extractfile(t1)
                v2 = tar2.extractfile(t2)
                if v1 is None:
                    continue
                self.assertTrue(v2 is not None, "stream.extractfile() failed")
                self.assertTrue(v1.read() == v2.read(), "stream extraction failed")
        finally:
            tar1.close()
项目:ndk-python    作者:gittor    | 项目源码 | 文件源码
def test_compare_members(self):
        tar1 = tarfile.open(tarname, encoding="iso8859-1")
        tar2 = self.tar

        while True:
            t1 = tar1.next()
            t2 = tar2.next()
            if t1 is None:
                break
            self.assertTrue(t2 is not None, "stream.next() failed.")

            if t2.islnk() or t2.issym():
                self.assertRaises(tarfile.StreamError, tar2.extractfile, t2)
                continue

            v1 = tar1.extractfile(t1)
            v2 = tar2.extractfile(t2)
            if v1 is None:
                continue
            self.assertTrue(v2 is not None, "stream.extractfile() failed")
            self.assertTrue(v1.read() == v2.read(), "stream extraction failed")

        tar1.close()
项目:android-backup-tools    作者:bluec0re    | 项目源码 | 文件源码
def test_compressed_stream(self):
        with AndroidBackup(io.BytesIO(TEST_DATA_NONENC)) as ab:
            self.assertEqual(ab.version, 3)
            self.assertEqual(ab.encryption, EncryptionType.NONE)
            self.assertEqual(ab.compression, CompressionType.ZLIB)

            names = list(map(lambda f: f.name, ab.get_files()))
            self.assertListEqual(names, TEST_MEMBERS_NAMES)

            tar = ab.read_data()
            with self.assertRaisesRegex(tarfile.StreamError, 'seeking backwards is not allowed'):
                tar.extractfile(
                    'apps/eu.bluec0re.android-backup/r/settings.cfg').read()
项目:android-backup-tools    作者:bluec0re    | 项目源码 | 文件源码
def test_encrypted_stream(self):
        with AndroidBackup(io.BytesIO(TEST_DATA_ENC_TEST), password='test') as ab:
            self.assertEqual(ab.version, 3)
            self.assertEqual(ab.encryption, EncryptionType.AES256)
            self.assertEqual(ab.compression, CompressionType.ZLIB)

            names = list(map(lambda f: f.name, ab.get_files()))
            self.assertListEqual(names, TEST_MEMBERS_NAMES)

            tar = ab.read_data()
            with self.assertRaisesRegex(tarfile.StreamError, 'seeking backwards is not allowed'):
                tar.extractfile(
                    'apps/eu.bluec0re.android-backup/r/settings.cfg').read()
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def test_read_through(self):
        # Issue #11224: A poorly designed _FileInFile.read() method
        # caused seeking errors with stream tar files.
        for tarinfo in self.tar:
            if not tarinfo.isreg():
                continue
            fobj = self.tar.extractfile(tarinfo)
            while True:
                try:
                    buf = fobj.read(512)
                except tarfile.StreamError:
                    self.fail("simple read-through using TarFile.extractfile() failed")
                if not buf:
                    break
            fobj.close()
项目:zippy    作者:securesystemslab    | 项目源码 | 文件源码
def test_provoke_stream_error(self):
        tarinfos = self.tar.getmembers()
        f = self.tar.extractfile(tarinfos[0]) # read the first member
        self.assertRaises(tarfile.StreamError, f.read)
项目:oil    作者:oilshell    | 项目源码 | 文件源码
def test_provoke_stream_error(self):
        tarinfos = self.tar.getmembers()
        f = self.tar.extractfile(tarinfos[0]) # read the first member
        self.assertRaises(tarfile.StreamError, f.read)
项目:python2-tracer    作者:extremecoders-re    | 项目源码 | 文件源码
def test_provoke_stream_error(self):
        tarinfos = self.tar.getmembers()
        f = self.tar.extractfile(tarinfos[0]) # read the first member
        self.assertRaises(tarfile.StreamError, f.read)
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def test_read_through(self):
        # Issue #11224: A poorly designed _FileInFile.read() method
        # caused seeking errors with stream tar files.
        for tarinfo in self.tar:
            if not tarinfo.isreg():
                continue
            with self.tar.extractfile(tarinfo) as fobj:
                while True:
                    try:
                        buf = fobj.read(512)
                    except tarfile.StreamError:
                        self.fail("simple read-through using "
                                  "TarFile.extractfile() failed")
                    if not buf:
                        break
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def test_provoke_stream_error(self):
        tarinfos = self.tar.getmembers()
        with self.tar.extractfile(tarinfos[0]) as f: # read the first member
            self.assertRaises(tarfile.StreamError, f.read)
项目:web_ctp    作者:molebot    | 项目源码 | 文件源码
def test_compare_members(self):
        tar1 = tarfile.open(tarname, encoding="iso8859-1")
        try:
            tar2 = self.tar

            while True:
                t1 = tar1.next()
                t2 = tar2.next()
                if t1 is None:
                    break
                self.assertIsNotNone(t2, "stream.next() failed.")

                if t2.islnk() or t2.issym():
                    with self.assertRaises(tarfile.StreamError):
                        tar2.extractfile(t2)
                    continue

                v1 = tar1.extractfile(t1)
                v2 = tar2.extractfile(t2)
                if v1 is None:
                    continue
                self.assertIsNotNone(v2, "stream.extractfile() failed")
                self.assertEqual(v1.read(), v2.read(),
                        "stream extraction failed")
        finally:
            tar1.close()
项目:pefile.pypy    作者:cloudtracer    | 项目源码 | 文件源码
def test_provoke_stream_error(self):
        tarinfos = self.tar.getmembers()
        f = self.tar.extractfile(tarinfos[0]) # read the first member
        self.assertRaises(tarfile.StreamError, f.read)
项目:pefile.pypy    作者:cloudtracer    | 项目源码 | 文件源码
def test_compare_members(self):
        tar1 = tarfile.open(tarname, encoding="iso8859-1")
        try:
            tar2 = self.tar

            while True:
                t1 = tar1.next()
                t2 = tar2.next()
                if t1 is None:
                    break
                self.assertIsNotNone(t2, "stream.next() failed.")

                if t2.islnk() or t2.issym():
                    with self.assertRaises(tarfile.StreamError):
                        tar2.extractfile(t2)
                    continue

                v1 = tar1.extractfile(t1)
                v2 = tar2.extractfile(t2)
                if v1 is None:
                    continue
                self.assertIsNotNone(v2, "stream.extractfile() failed")
                self.assertEqual(v1.read(), v2.read(),
                        "stream extraction failed")
        finally:
            tar1.close()
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def test_read_through(self):
        # Issue #11224: A poorly designed _FileInFile.read() method
        # caused seeking errors with stream tar files.
        for tarinfo in self.tar:
            if not tarinfo.isreg():
                continue
            with self.tar.extractfile(tarinfo) as fobj:
                while True:
                    try:
                        buf = fobj.read(512)
                    except tarfile.StreamError:
                        self.fail("simple read-through using "
                                  "TarFile.extractfile() failed")
                    if not buf:
                        break
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def test_provoke_stream_error(self):
        tarinfos = self.tar.getmembers()
        with self.tar.extractfile(tarinfos[0]) as f: # read the first member
            self.assertRaises(tarfile.StreamError, f.read)
项目:ouroboros    作者:pybee    | 项目源码 | 文件源码
def test_compare_members(self):
        tar1 = tarfile.open(tarname, encoding="iso8859-1")
        try:
            tar2 = self.tar

            while True:
                t1 = tar1.next()
                t2 = tar2.next()
                if t1 is None:
                    break
                self.assertIsNotNone(t2, "stream.next() failed.")

                if t2.islnk() or t2.issym():
                    with self.assertRaises(tarfile.StreamError):
                        tar2.extractfile(t2)
                    continue

                v1 = tar1.extractfile(t1)
                v2 = tar2.extractfile(t2)
                if v1 is None:
                    continue
                self.assertIsNotNone(v2, "stream.extractfile() failed")
                self.assertEqual(v1.read(), v2.read(),
                        "stream extraction failed")
        finally:
            tar1.close()
项目:ndk-python    作者:gittor    | 项目源码 | 文件源码
def test_provoke_stream_error(self):
        tarinfos = self.tar.getmembers()
        f = self.tar.extractfile(tarinfos[0]) # read the first member
        self.assertRaises(tarfile.StreamError, f.read)
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def test_read_through(self):
        # Issue #11224: A poorly designed _FileInFile.read() method
        # caused seeking errors with stream tar files.
        for tarinfo in self.tar:
            if not tarinfo.isreg():
                continue
            with self.tar.extractfile(tarinfo) as fobj:
                while True:
                    try:
                        buf = fobj.read(512)
                    except tarfile.StreamError:
                        self.fail("simple read-through using "
                                  "TarFile.extractfile() failed")
                    if not buf:
                        break
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def test_provoke_stream_error(self):
        tarinfos = self.tar.getmembers()
        with self.tar.extractfile(tarinfos[0]) as f: # read the first member
            self.assertRaises(tarfile.StreamError, f.read)
项目:kbe_server    作者:xiaohaoppy    | 项目源码 | 文件源码
def test_compare_members(self):
        tar1 = tarfile.open(tarname, encoding="iso8859-1")
        try:
            tar2 = self.tar

            while True:
                t1 = tar1.next()
                t2 = tar2.next()
                if t1 is None:
                    break
                self.assertIsNotNone(t2, "stream.next() failed.")

                if t2.islnk() or t2.issym():
                    with self.assertRaises(tarfile.StreamError):
                        tar2.extractfile(t2)
                    continue

                v1 = tar1.extractfile(t1)
                v2 = tar2.extractfile(t2)
                if v1 is None:
                    continue
                self.assertIsNotNone(v2, "stream.extractfile() failed")
                self.assertEqual(v1.read(), v2.read(),
                        "stream extraction failed")
        finally:
            tar1.close()