mirror of
https://github.com/gentoo-mirror/guru.git
synced 2026-10-24 18:37:44 -04:00
Package-Manager: Portage-3.0.30, Repoman-3.0.3 Signed-off-by: Marco Scardovi <marco@scardovi.com>
264 lines
10 KiB
Python
264 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
# vim: ts=4 sw=4 et ai si
|
|
#
|
|
# Copyright (c) 2012-2014 Intel, Inc.
|
|
# License: GPLv2
|
|
# Author: Artem Bityutskiy <artem.bityutskiy@linux.intel.com>
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License, version 2,
|
|
# as published by the Free Software Foundation.
|
|
#
|
|
# This program is distributed in the hope that it will be useful, but
|
|
# WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
|
# General Public License for more details.
|
|
|
|
"""
|
|
This test verifies the base bmap creation and copying API functionality. It
|
|
generates a random sparse file, then creates a bmap fir this file and copies it
|
|
to a different file using the bmap. Then it compares the original random sparse
|
|
file and the copy and verifies that they are identical.
|
|
"""
|
|
|
|
# Disable the following pylint recommendations:
|
|
# * Too many public methods (R0904)
|
|
# * Too many local variables (R0914)
|
|
# * Too many statements (R0915)
|
|
# pylint: disable=R0904
|
|
# pylint: disable=R0914
|
|
# pylint: disable=R0915
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import filecmp
|
|
import subprocess
|
|
from six.moves import zip_longest
|
|
from tests import helpers
|
|
from bmaptools import BmapHelpers, BmapCreate, Filemap
|
|
|
|
# This is a work-around for Centos 6
|
|
try:
|
|
import unittest2 as unittest # pylint: disable=F0401
|
|
except ImportError:
|
|
import unittest
|
|
|
|
|
|
class Error(Exception):
|
|
"""A class for exceptions generated by this test."""
|
|
pass
|
|
|
|
|
|
def _compare_holes(file1, file2):
|
|
"""
|
|
Make sure that files 'file1' and 'file2' have holes at the same places.
|
|
The 'file1' and 'file2' arguments may be full file paths or file objects.
|
|
"""
|
|
|
|
filemap1 = Filemap.filemap(file1)
|
|
filemap2 = Filemap.filemap(file2)
|
|
|
|
iterator1 = filemap1.get_unmapped_ranges(0, filemap1.blocks_cnt)
|
|
iterator2 = filemap2.get_unmapped_ranges(0, filemap2.blocks_cnt)
|
|
|
|
iterator = zip_longest(iterator1, iterator2)
|
|
for range1, range2 in iterator:
|
|
if range1 != range2:
|
|
raise Error("mismatch for hole %d-%d, it is %d-%d in file2"
|
|
% (range1[0], range1[1], range2[0], range2[1]))
|
|
|
|
|
|
def _generate_compressed_files(file_path, delete=True):
|
|
"""
|
|
This is a generator which yields compressed versions of a file
|
|
'file_path'.
|
|
The 'delete' argument specifies whether the compressed files that this
|
|
generator yields have to be automatically deleted.
|
|
"""
|
|
|
|
# Make sure the temporary files start with the same name as 'file_obj' in
|
|
# order to simplify debugging.
|
|
prefix = os.path.splitext(os.path.basename(file_path))[0] + '.'
|
|
# Put the temporary files in the directory with 'file_obj'
|
|
directory = os.path.dirname(file_path)
|
|
|
|
compressors = [("bzip2", None, ".bz2", "-c -k"),
|
|
("pbzip2", None, ".p.bz2", "-c -k"),
|
|
("gzip", None, ".gz", "-c"),
|
|
("pigz", None, ".p.gz", "-c -k"),
|
|
("xz", None, ".xz", "-c -k"),
|
|
("lzop", None, ".lzo", "-c -k"),
|
|
("lz4", None, ".lz4", "-c -k"),
|
|
("zstd", None, ".zst", "-c -k"),
|
|
# The "-P -C /" trick is used to avoid silly warnings:
|
|
# "tar: Removing leading `/' from member names"
|
|
("bzip2", "tar", ".tar.bz2", "-c -j -O -P -C /"),
|
|
("gzip", "tar", ".tar.gz", "-c -z -O -P -C /"),
|
|
("xz", "tar", ".tar.xz", "-c -J -O -P -C /"),
|
|
("lzop", "tar", ".tar.lzo", "-c --lzo -O -P -C /"),
|
|
("lz4", "tar", ".tar.lz4", "-c -Ilz4 -O -P -C /"),
|
|
("zstd", "tar", ".tar.zst", "-c -Izstd -O -P -C /"),
|
|
("zip", None, ".zip", "-q -j -")]
|
|
|
|
for decompressor, archiver, suffix, options in compressors:
|
|
if not BmapHelpers.program_is_available(decompressor):
|
|
continue
|
|
if archiver and not BmapHelpers.program_is_available(archiver):
|
|
continue
|
|
|
|
tmp_file_obj = tempfile.NamedTemporaryFile('wb+', prefix=prefix,
|
|
delete=delete, dir=directory,
|
|
suffix=suffix)
|
|
|
|
if archiver:
|
|
args = archiver + " " + options + " " + file_path
|
|
else:
|
|
args = decompressor + " " + options + " " + file_path
|
|
child_process = subprocess.Popen(args, shell=True, stdout=tmp_file_obj)
|
|
child_process.wait()
|
|
tmp_file_obj.flush()
|
|
yield tmp_file_obj.name
|
|
tmp_file_obj.close()
|
|
|
|
|
|
def _do_test(image, image_size, delete=True):
|
|
"""
|
|
A basic test for the bmap creation and copying functionality. It first
|
|
generates a bmap for file 'image', and then copies the sparse file to a
|
|
different file, and then checks that the original file and the copy are
|
|
identical.
|
|
The 'image_size' argument is size of the image in bytes. The 'delete'
|
|
argument specifies whether the temporary files that this function creates
|
|
have to be automatically deleted.
|
|
"""
|
|
|
|
try:
|
|
Filemap.filemap(image)
|
|
except Filemap.ErrorNotSupp as e:
|
|
sys.stderr.write('%s\n' % e)
|
|
return
|
|
|
|
# Make sure the temporary files start with the same name as 'image' in
|
|
# order to simplify debugging.
|
|
prefix = os.path.splitext(os.path.basename(image))[0] + '.'
|
|
# Put the temporary files in the directory with the image
|
|
directory = os.path.dirname(image)
|
|
|
|
# Create and open a temporary file for a copy of the image
|
|
f_copy = tempfile.NamedTemporaryFile("wb+", prefix=prefix,
|
|
delete=delete, dir=directory,
|
|
suffix=".copy")
|
|
|
|
# Create and open 2 temporary files for the bmap
|
|
f_bmap1 = tempfile.NamedTemporaryFile("w+", prefix=prefix,
|
|
delete=delete, dir=directory,
|
|
suffix=".bmap1")
|
|
f_bmap2 = tempfile.NamedTemporaryFile("w+", prefix=prefix,
|
|
delete=delete, dir=directory,
|
|
suffix=".bmap2")
|
|
|
|
image_chksum = helpers.calculate_chksum(image)
|
|
|
|
#
|
|
# Pass 1: generate the bmap, copy and compare
|
|
#
|
|
|
|
# Create bmap for the random sparse file
|
|
creator = BmapCreate.BmapCreate(image, f_bmap1.name)
|
|
creator.generate()
|
|
|
|
helpers.copy_and_verify_image(image, f_copy.name, f_bmap1.name,
|
|
image_chksum, image_size)
|
|
|
|
# Make sure that holes in the copy are identical to holes in the random
|
|
# sparse file.
|
|
_compare_holes(image, f_copy.name)
|
|
|
|
#
|
|
# Pass 2: same as pass 1, but use file objects instead of paths
|
|
#
|
|
|
|
creator = BmapCreate.BmapCreate(image, f_bmap2)
|
|
creator.generate()
|
|
helpers.copy_and_verify_image(image, f_copy.name, f_bmap2.name,
|
|
image_chksum, image_size)
|
|
_compare_holes(image, f_copy.name)
|
|
|
|
# Make sure the bmap files generated at pass 1 and pass 2 are identical
|
|
assert filecmp.cmp(f_bmap1.name, f_bmap2.name, False)
|
|
|
|
#
|
|
# Pass 3: test compressed files copying with bmap
|
|
#
|
|
|
|
for compressed in _generate_compressed_files(image, delete=delete):
|
|
helpers.copy_and_verify_image(compressed, f_copy.name,
|
|
f_bmap1.name, image_chksum, image_size)
|
|
|
|
# Test without setting the size
|
|
helpers.copy_and_verify_image(compressed, f_copy.name, f_bmap1.name,
|
|
image_chksum, None)
|
|
|
|
# Append a "file:" prefixe to make BmapCopy use urllib
|
|
compressed = "file:" + compressed
|
|
helpers.copy_and_verify_image(compressed, f_copy.name, f_bmap1.name,
|
|
image_chksum, image_size)
|
|
helpers.copy_and_verify_image(compressed, f_copy.name, f_bmap1.name,
|
|
image_chksum, None)
|
|
|
|
#
|
|
# Pass 5: copy without bmap and make sure it is identical to the original
|
|
# file.
|
|
|
|
helpers.copy_and_verify_image(image, f_copy.name, None, image_chksum,
|
|
image_size)
|
|
helpers.copy_and_verify_image(image, f_copy.name, None, image_chksum, None)
|
|
|
|
#
|
|
# Pass 6: test compressed files copying without bmap
|
|
#
|
|
|
|
for compressed in _generate_compressed_files(image, delete=delete):
|
|
helpers.copy_and_verify_image(compressed, f_copy.name, f_bmap1.name,
|
|
image_chksum, image_size)
|
|
|
|
# Test without setting the size
|
|
helpers.copy_and_verify_image(compressed, f_copy.name, f_bmap1.name,
|
|
image_chksum, None)
|
|
|
|
# Append a "file:" prefix to make BmapCopy use urllib
|
|
helpers.copy_and_verify_image(compressed, f_copy.name, f_bmap1.name,
|
|
image_chksum, image_size)
|
|
helpers.copy_and_verify_image(compressed, f_copy.name, f_bmap1.name,
|
|
image_chksum, None)
|
|
|
|
# Close temporary files, which will also remove them
|
|
f_copy.close()
|
|
f_bmap1.close()
|
|
f_bmap2.close()
|
|
|
|
|
|
class TestCreateCopy(unittest.TestCase):
|
|
"""
|
|
The test class for this unit tests. Basically executes the '_do_test()'
|
|
function for different sparse files.
|
|
"""
|
|
|
|
def test(self): # pylint: disable=R0201
|
|
"""
|
|
The test entry point. Executes the '_do_test()' function for files of
|
|
different sizes, holes distribution and format.
|
|
"""
|
|
|
|
# Delete all the test-related temporary files automatically
|
|
delete = True
|
|
# Create all the test-related temporary files in current directory
|
|
directory = '.'
|
|
|
|
iterator = helpers.generate_test_files(delete=delete,
|
|
directory=directory)
|
|
for f_image, image_size, _, _ in iterator:
|
|
assert image_size == os.path.getsize(f_image.name)
|
|
_do_test(f_image.name, image_size, delete=delete)
|