developer.intra2net.com Git - python-delta-tar/blob - testing/test_multivol.py

   1 # Copyright (C) 2013 Intra2net AG
   2 #
   3 # This program is free software; you can redistribute it and/or modify
   4 # it under the terms of the GNU Lesser General Public License as published
   5 # by the Free Software Foundation; either version 3 of the License, or
   6 # (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU Lesser General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program.  If not, see
  15 # <http://www.gnu.org/licenses/lgpl-3.0.html>
  16
  17
  18 import os
  19 from tempfile import TemporaryDirectory
  20
  21 import deltatar.crypto as crypto
  22 from deltatar.tarfile import TarFile, PAX_FORMAT, GNU_FORMAT, BLOCKSIZE
  23 from . import BaseTest, new_volume_handler, closing_new_volume_handler
  24 from .create_pseudo_random_files import create_file as create_random_file
  25 from .test_multivol_compression_sizes import test as multivol_compr_test_func
  26
  27 class MultivolGnuFormatTest(BaseTest):
  28     """
  29     Test multivolume support in tarfile. Tar Format is specified at class level.
  30     """
  31
  32     # used as the --format argument to tar command on tar file creation
  33     tar_command_format = "gnu"
  34
  35     # used as Tarfile.open format option argument for tar file creation
  36     tarfile_format = GNU_FORMAT
  37
  38     # overhead size used to calculate the exact maximum size of a tar file with
  39     # no extra volume that stores only one file. In case of GNU format this is
  40     # the size of three blocks:
  41     # * 1 block used to store the header information of the stored file
  42     # * 2 blocks used to mark the end of the tar file
  43     tarfile_overhead = 3*BLOCKSIZE
  44     file_overhead = 1*BLOCKSIZE
  45
  46     # overhead size used to calculate the exact maximum size of a tar volume,
  47     # corresponding with a multivolume tar file storing a single file. In the
  48     # case of GNU format this is the same as tarfile_overhead.
  49     tarvol_overhead = 3*BLOCKSIZE
  50
  51     def test_no_volume(self):
  52         """
  53         Create a tar file with only one file inside and no extra volumes
  54         """
  55
  56         # create the content of the file to compress and hash it
  57         hash = self.create_file("big", 50000)
  58
  59         # create the tar file with volumes
  60         tarobj = TarFile.open("sample.tar", mode="w", format=self.tarfile_format)
  61         tarobj.add("big")
  62         tarobj.close()
  63
  64         # check that the tar volumes were correctly created
  65         assert os.path.exists("sample.tar")
  66         assert not os.path.exists("sample.tar.1")
  67
  68         os.unlink("big")
  69         assert not os.path.exists("big")
  70
  71         # extract and check
  72         os.system("tar xfM sample.tar")
  73         assert os.path.exists("big")
  74         assert hash == self.md5sum("big")
  75
  76     def test_volume_creation1(self):
  77         """
  78         Create a tar file with two volumes, only one file inside
  79         """
  80
  81         # create the content of the file to compress and hash it
  82         hash = self.create_file("big", 50000)
  83
  84         # create the tar file with volumes
  85         tarobj = TarFile.open("sample.tar",
  86                               mode="w",
  87                               format=self.tarfile_format,
  88                               max_volume_size=30000,
  89                               new_volume_handler=new_volume_handler)
  90         tarobj.add("big")
  91         tarobj.close()
  92
  93         # check that the tar volumes were correctly created
  94         assert os.path.exists("sample.tar")
  95         assert os.path.exists("sample.tar.1")
  96         assert not os.path.exists("sample.tar.2")
  97
  98         os.unlink("big")
  99         assert not os.path.exists("big")
 100
 101         # extract with normal tar and check output
 102         os.system("tar xfM sample.tar --file=sample.tar.1")
 103         assert os.path.exists("big")
 104         assert hash == self.md5sum("big")
 105
 106     def test_volume_creation2(self):
 107         """
 108         Create a tar file with 2 extra volumes, only one file inside
 109         """
 110
 111         # create the content of the file to compress and hash it
 112         hash = self.create_file("big", 50000)
 113
 114         # create the tar file with volumes
 115         tarobj = TarFile.open("sample.tar",
 116                               mode="w",
 117                               format=self.tarfile_format,
 118                               max_volume_size=20000,
 119                               new_volume_handler=new_volume_handler)
 120         tarobj.add("big")
 121         tarobj.close()
 122
 123         # check that the tar volumes were correctly created
 124         assert os.path.exists("sample.tar")
 125         assert os.path.exists("sample.tar.1")
 126         assert os.path.exists("sample.tar.2")
 127         assert not os.path.exists("sample.tar.3")
 128
 129         os.unlink("big")
 130         assert not os.path.exists("big")
 131
 132         # extract with normal tar and check output
 133         os.system("tar xfM sample.tar --file=sample.tar.1 --file=sample.tar.2")
 134         assert os.path.exists("big")
 135         assert hash == self.md5sum("big")
 136
 137     def test_multivol_multifiles(self):
 138         '''
 139         Create a tar file with two volumes and three files inside
 140         '''
 141
 142         # create sample data
 143         hash = dict()
 144         hash["big"]  = self.create_file("big", 50000)
 145         hash["small"] = self.create_file("small", 100)
 146         hash["small2"] = self.create_file("small2", 354)
 147
 148         # create the tar file with volumes
 149         tarobj = TarFile.open("sample.tar",
 150                               mode="w",
 151                               format=self.tarfile_format,
 152                               max_volume_size=20000,
 153                               new_volume_handler=new_volume_handler)
 154         tarobj.add("big")
 155         tarobj.add("small")
 156         tarobj.add("small2")
 157         tarobj.close()
 158
 159         # check that the tar volumes were correctly created
 160         assert os.path.exists("sample.tar")
 161         assert os.path.exists("sample.tar.1")
 162         assert os.path.exists("sample.tar.2")
 163         assert not os.path.exists("sample.tar.3")
 164
 165         os.unlink("big")
 166         os.unlink("small")
 167         os.unlink("small2")
 168
 169         # extract with normal tar and check output
 170         os.system("tar xfM sample.tar --file=sample.tar.1 --file=sample.tar.2")
 171         for key, value in hash.items():
 172             assert os.path.exists(key)
 173             assert value == self.md5sum(key)
 174
 175     def test_get_file_size(self):
 176         '''
 177         Test _Stream.get_file_size which is the basis for multivol with
 178         compression.
 179         '''
 180
 181         # create test files of different sizes
 182         size_factor = 4
 183         n_sizes = 13    # 1,4,16,64,256,1KiB=4**5 ... 1MiB=4**10 ... 1GiB=4**15
 184         next_size = 1
 185         sizes = []
 186         file_names = []
 187         for exponents in range(n_sizes):
 188             sizes.append(next_size)
 189             new_name = 'size_test_{:08d}'.format(next_size)
 190             file_names.append(new_name)
 191             self.create_file(new_name, next_size)
 192             next_size *= size_factor
 193
 194         max_err = 0
 195         max_err_post = 0
 196
 197         for mode, password in [ ('w|gz', None)  , ('w|bz2', None)
 198                               , ('w|xz', None)  , ('w#gz' , None)
 199                               , ('w#gz', "test"), ('w#tar', "test")
 200                               ]:
 201             tar_file_name = "size_test.tar." + mode[2:]
 202             for size_number in range(4,n_sizes):
 203                 for order in 1,-1:   # small files first or big files first
 204                     encryptor = None
 205                     if password is None:
 206                         encryptor = None # could leak due to scoping
 207                     else:
 208                         encryptor = crypto.Encrypt (password=password, version=1,
 209                                                     paramversion=1)
 210                     tarobj = TarFile.open(tar_file_name,
 211                                           mode=mode,
 212                                           format=self.tarfile_format,
 213                                           encryption=encryptor)
 214                     for file_name in file_names[:size_number][::order]:
 215                         tarobj.add(file_name)
 216                     estimate = tarobj.fileobj.estim_file_size()
 217                     tarobj.close()
 218                     estimate_post = tarobj.fileobj.estim_file_size()
 219                     actual_size = os.stat(tar_file_name).st_size
 220                     err = abs(actual_size - estimate)
 221                     #print('mode {:>11s}, {:2} files (up to size {:9}): '
 222                     #      'estim={:9}, true={:9}, post={:9}, err={:5}'
 223                     #      .format(mode, size_number, sizes[size_number],
 224                     #              estimate, actual_size, estimate_post, err))
 225                     os.unlink(tar_file_name)
 226                     if err > max_err:
 227                         max_err = err
 228                     err = abs(actual_size - estimate_post)
 229                     if err > max_err_post:
 230                         max_err_post = err
 231
 232         #print('max err is {}, post={}'.format(max_err, max_err_post))
 233         assert max_err < 13*1024
 234         assert max_err_post == 0
 235
 236         # clean up
 237         for file_name in file_names:
 238             os.unlink(file_name)
 239
 240     def test_multivol_compress_warning(self):
 241         """ check warning being issued if compressing multivolume with w: """
 242         with self.assertWarns(UserWarning):
 243             tarobj = TarFile.open("sample.tar.gz",
 244                                   mode="w:gz",
 245                                   format=self.tarfile_format,
 246                                   max_volume_size=30000,
 247                                   new_volume_handler=new_volume_handler)
 248             tarobj.close()
 249
 250     def test_compress_single(self):
 251         ''' check creation of single volume when compression is on '''
 252
 253         # create the content of the file to compress and hash it
 254         hash = self.create_file("big", 50000)
 255
 256         # create the tar file with volumes and compression
 257         tarobj = TarFile.open("sample.tar.gz",
 258                               mode="w#gz",
 259                               format=self.tarfile_format,
 260                               max_volume_size=30000,
 261                               new_volume_handler=new_volume_handler,
 262                               debug=3)
 263         tarobj.add("big")
 264         tarobj.list()
 265         tarobj.close()
 266
 267         # data fits into a single volume -- check that no second is created
 268         assert os.path.exists("sample.tar.gz")
 269         assert not os.path.exists("sample.tar.gz.1")
 270
 271         # check size of first volume
 272         size = os.stat("sample.tar.gz").st_size
 273         arbitrary_low_size_bound  = 315 # adjust if zlib changes
 274         arbitrary_high_size_bound = 410 # adjust if zlib changes
 275         assert arbitrary_low_size_bound < size < arbitrary_high_size_bound, \
 276                'size of sample.tar.gz is {}'.format(size)
 277
 278         os.unlink("big")
 279         assert not os.path.exists("big")
 280
 281         # extract with normal tar and check output
 282         #print('unpacking:')
 283         import subprocess
 284         output = subprocess.check_output(
 285             "tar xvf sample.tar.gz".split(),
 286             universal_newlines=True)
 287         #for line in output.splitlines():
 288         #    print(line.rstrip())
 289         assert os.path.exists("big")
 290         assert hash == self.md5sum("big")
 291
 292         os.unlink('big')
 293
 294     def test_multivol_compress(self):
 295         ''' check creation of multiple volumes when compression is on '''
 296
 297         # create a random file that is not so easy to compress...
 298         filename = create_random_file('./', 100000)
 299         hash = self.md5sum(filename)
 300
 301         # need own volume handler so files maintain their suffix for gunzip
 302         def my_volume_handler(tarobj, base_name, volume_number):
 303             if not base_name.endswith('.tar.gz'):
 304                 raise ValueError('need .tar.gz file!')
 305             tarobj.fileobj.close()
 306             new_name = '{}.{}.tar.gz'.format(base_name[:-7], volume_number)
 307             tarobj.open_volume(new_name)
 308
 309         # create the tar file with volumes and compression
 310         tarobj = TarFile.open("sample.tar.gz",
 311                               mode="w#gz",
 312                               format=self.tarfile_format,
 313                               max_volume_size=30000,
 314                               new_volume_handler=my_volume_handler,
 315                               debug=3)
 316         tarobj.add(filename)
 317         tarobj.list()
 318         tarobj.close()
 319
 320         # data fits into 2 volumes -- check that no third is created
 321         assert os.path.exists("sample.tar.gz")
 322         assert os.path.exists("sample.1.tar.gz")
 323         assert not os.path.exists("sample.tar.gz.1")
 324         assert not os.path.exists("sample.tar.gz.2")
 325         assert not os.path.exists("sample.2.tar.gz.2")
 326
 327         os.unlink(filename)
 328         assert not os.path.exists(filename)
 329
 330         # extract with shell means; slightly complicated because the linux
 331         # tar/gunzip cannot do gzipped-multi-volume archives
 332         #print('unpacking:')
 333         import subprocess
 334         for cmd in 'gunzip -v sample.tar.gz', 'gunzip -v sample.1.tar.gz', \
 335                 'tar xvfM sample.tar --file=sample.1.tar':
 336             #print(cmd)
 337             output = subprocess.check_output(cmd.split(),
 338                                              universal_newlines=True)
 339             #for line in output.splitlines():
 340             #    print(line.rstrip())
 341         assert os.path.exists(filename)
 342         assert hash == self.md5sum(filename)
 343
 344         os.unlink('sample.tar')
 345         os.unlink('sample.1.tar')
 346         os.unlink(filename)
 347
 348     def test_volume_extract1(self):
 349         '''
 350         Create a tar file with multiple volumes and one file and extract it
 351         '''
 352         # create the content of the file to compress and hash it
 353         hash = self.create_file("big", 5*1024*1024)
 354
 355         # create the tar file with volumes
 356         tarobj = TarFile.open("sample.tar",
 357                               mode="w",
 358                               format=self.tarfile_format,
 359                               max_volume_size=3*1024*1024,
 360                               new_volume_handler=new_volume_handler)
 361         tarobj.add("big")
 362         tarobj.close()
 363
 364         # check that the tar volumes were correctly created
 365         assert os.path.exists("sample.tar")
 366         assert os.path.exists("sample.tar.1")
 367         assert not os.path.exists("sample.tar.2")
 368
 369         os.unlink("big")
 370         assert not os.path.exists("big")
 371
 372         # extract and check output
 373         tarobj = TarFile.open("sample.tar",
 374                               mode="r",
 375                               new_volume_handler=new_volume_handler)
 376         tarobj.extractall()
 377         tarobj.close()
 378         assert os.path.exists("big")
 379         assert hash == self.md5sum("big")
 380
 381     def test_volume_extract2(self):
 382         '''
 383         Create a multivolume tar file with gnu tar command, extract it with
 384         tarfile library
 385         '''
 386         # create the content of the file to compress and hash it
 387         hash = self.create_file("big", 5*1024*1024)
 388
 389         # create the tar file with volumes
 390         os.system("tar cM --format=%s -L 3000 big --file=sample.tar "\
 391             "--file=sample.tar.1" % self.tar_command_format)
 392
 393         # check that the tar volumes were correctly created
 394         assert os.path.exists("sample.tar")
 395         assert os.path.exists("sample.tar.1")
 396         assert not os.path.exists("sample.tar.2")
 397
 398         os.unlink("big")
 399         assert not os.path.exists("big")
 400
 401         # extract and check output
 402         tarobj = TarFile.open("sample.tar",
 403                               mode="r",
 404                               new_volume_handler=new_volume_handler)
 405         tarobj.extractall()
 406         tarobj.close()
 407         assert os.path.exists("big")
 408         assert hash == self.md5sum("big")
 409
 410     def test_volume_extract3(self):
 411         '''
 412         Create a multivolume tar file with gnu tar command with multiple
 413         files, extract it with tarfile library
 414         '''
 415         # create the content of the file to compress and hash it
 416         hash = dict()
 417         hash["big"] = self.create_file("big", 5*1024*1024)
 418         hash["small"] = self.create_file("small", 100)
 419         hash["small2"] = self.create_file("small2", 354)
 420
 421         # create the tar file with volumes
 422         os.system("tar cM --format=%s -L 3000 big small small2 --file=sample.tar "\
 423             "--file=sample.tar.1" % self.tar_command_format)
 424
 425         # check that the tar volumes were correctly created
 426         assert os.path.exists("sample.tar")
 427         assert os.path.exists("sample.tar.1")
 428         assert not os.path.exists("sample.tar.2")
 429
 430         for key, value in hash.items():
 431             os.unlink(key)
 432             assert not os.path.exists(key)
 433
 434         # extract and check output
 435         tarobj = TarFile.open("sample.tar",
 436                               mode="r",
 437                               new_volume_handler=new_volume_handler)
 438         tarobj.extractall()
 439         tarobj.close()
 440
 441         for key, value in hash.items():
 442             assert os.path.exists(key)
 443             assert value == self.md5sum(key)
 444
 445     def test_multivol_multifile_extract(self):
 446         '''
 447         create a multivolume tar file with multiple files and extracts it
 448         '''
 449
 450         # create sample data
 451         hash = dict()
 452         hash["big"] = self.create_file("big", 50000)
 453         hash["small"] = self.create_file("small", 100)
 454         hash["small2"] = self.create_file("small2", 354)
 455
 456         # create the tar file with volumes
 457         tarobj = TarFile.open("sample.tar",
 458                               mode="w",
 459                               format=self.tarfile_format,
 460                               max_volume_size=20000,
 461                               new_volume_handler=new_volume_handler)
 462         tarobj.add("big")
 463         tarobj.add("small")
 464         tarobj.add("small2")
 465         tarobj.close()
 466
 467         # check that the tar volumes were correctly created
 468         assert os.path.exists("sample.tar")
 469         assert os.path.exists("sample.tar.1")
 470         assert os.path.exists("sample.tar.2")
 471         assert not os.path.exists("sample.tar.3")
 472
 473         os.unlink("big")
 474         os.unlink("small")
 475         os.unlink("small2")
 476
 477         # extract and check output
 478         tarobj = TarFile.open("sample.tar",
 479                               mode="r",
 480                               new_volume_handler=new_volume_handler)
 481         tarobj.extractall()
 482         tarobj.close()
 483
 484         for key, value in hash.items():
 485             assert os.path.exists(key)
 486             assert value == self.md5sum(key)
 487
 488     def test_multiple_files_extract(self):
 489         '''
 490         creates a simple tar file with no volumes and with multiple files
 491         inside and extracts it
 492         '''
 493
 494         # create sample data
 495         hash = dict()
 496         hash["big"] = self.create_file("big", 50000)
 497         hash["small"] = self.create_file("small", 100)
 498         hash["small2"] = self.create_file("small2", 354)
 499
 500         # create the tar file with volumes
 501         tarobj = TarFile.open("sample.tar",
 502                               format=self.tarfile_format,
 503                               mode="w")
 504         tarobj.add("big")
 505         tarobj.add("small")
 506         tarobj.add("small2")
 507         tarobj.close()
 508
 509         # check that the tar volumes were correctly created
 510         assert os.path.exists("sample.tar")
 511         assert not os.path.exists("sample.tar.1")
 512
 513         os.unlink("big")
 514         os.unlink("small")
 515         os.unlink("small2")
 516
 517         # extract and check output
 518         tarobj = TarFile.open("sample.tar",
 519                               mode="r",
 520                               new_volume_handler=new_volume_handler)
 521         tarobj.extractall()
 522         tarobj.close()
 523
 524         for key, value in hash.items():
 525             assert os.path.exists(key)
 526             assert value == self.md5sum(key)
 527
 528     def test_corner_case_split_size1(self):
 529         '''
 530         Creates a tar file with a single file inside that contains the maximum
 531         size allowed in one volume.
 532         '''
 533         hash = self.create_file("big", 5*1024*1024)
 534
 535         # create the tar file with volumes
 536         tarobj = TarFile.open("sample.tar",
 537                               mode="w",
 538                               format=self.tarfile_format,
 539                               # see tarfile_overhead description for details
 540                               max_volume_size=5*1024*1024 + self.tarfile_overhead,
 541                               new_volume_handler=new_volume_handler)
 542         tarobj.add("big")
 543         tarobj.close()
 544
 545         # check that the tar volumes were correctly created
 546         assert os.path.exists("sample.tar")
 547         assert not os.path.exists("sample.tar.1")
 548
 549         os.unlink("big")
 550         assert not os.path.exists("big")
 551
 552         # extract and check output
 553         tarobj = TarFile.open("sample.tar",
 554                               mode="r",
 555                               new_volume_handler=new_volume_handler)
 556         tarobj.extractall()
 557         tarobj.close()
 558         assert os.path.exists("big")
 559         assert hash == self.md5sum("big")
 560
 561     def test_corner_case_split_size2(self):
 562         '''
 563         Creates a tar file with a single file inside that contains the maximum
 564         size allowed in one volume.
 565         '''
 566         hash = self.create_file("big", 4*1024*1024)
 567
 568         # create the tar file with volumes
 569         tarobj = TarFile.open("sample.tar",
 570                               mode="w",
 571                               format=self.tarfile_format,
 572                               # see tarvol_overhead description for details
 573                               max_volume_size=2*1024*1024 + self.tarvol_overhead,
 574                               new_volume_handler=new_volume_handler)
 575         tarobj.add("big")
 576         tarobj.close()
 577
 578         # check that the tar volumes were correctly created
 579         assert os.path.exists("sample.tar")
 580         assert os.path.exists("sample.tar.1")
 581         assert not os.path.exists("sample.tar.2")
 582
 583         os.unlink("big")
 584         assert not os.path.exists("big")
 585
 586         # extract and check output
 587         tarobj = TarFile.open("sample.tar",
 588                               mode="r",
 589                               new_volume_handler=new_volume_handler)
 590         tarobj.extractall()
 591         tarobj.close()
 592         assert os.path.exists("big")
 593         assert hash == self.md5sum("big")
 594
 595     def test_corner_case_split_size3(self):
 596         '''
 597         Creates a tar file with a single file inside that contains the maximum
 598         size allowed in one volume but without the overhead.
 599         '''
 600         hash = self.create_file("big", 4*1024*1024)
 601
 602         # create the tar file with volumes
 603         tarobj = TarFile.open("sample.tar",
 604                               mode="w",
 605                               format=self.tarfile_format,
 606                               max_volume_size=2*1024*1024,
 607                               new_volume_handler=new_volume_handler)
 608         tarobj.add("big")
 609         tarobj.close()
 610
 611         # check that the tar volumes were correctly created
 612         assert os.path.exists("sample.tar")
 613         assert os.path.exists("sample.tar.1")
 614         assert os.path.exists("sample.tar.2")
 615         assert not os.path.exists("sample.tar.3")
 616
 617         os.unlink("big")
 618         assert not os.path.exists("big")
 619
 620         # extract and check output
 621         tarobj = TarFile.open("sample.tar",
 622                               mode="r",
 623                               new_volume_handler=new_volume_handler)
 624         tarobj.extractall()
 625         tarobj.close()
 626         assert os.path.exists("big")
 627         assert hash == self.md5sum("big")
 628
 629     def test_corner_case_split_size4(self):
 630         '''
 631         Creates a tar file with multiple files inside that contains the maximum
 632         size allowed in one volume.
 633         '''
 634         hash = dict()
 635         hash['big'] = self.create_file("big", 3*1024*1024)
 636         hash['small'] = self.create_file("small", 1*1024*1024)
 637
 638         # create the tar file with volumes
 639         tarobj = TarFile.open("sample.tar",
 640                               mode="w",
 641                               format=self.tarfile_format,
 642                               max_volume_size=(4*1024*1024 +
 643                                                self.tarfile_overhead +
 644                                                self.file_overhead),
 645                               new_volume_handler=new_volume_handler)
 646         tarobj.add("big")
 647         tarobj.add("small")
 648         tarobj.close()
 649
 650         # check that the tar volumes were correctly created
 651         assert os.path.exists("sample.tar")
 652         assert not os.path.exists("sample.tar.1")
 653
 654         for key, value in hash.items():
 655             os.unlink(key)
 656             assert not os.path.exists(key)
 657
 658         # extract and check output
 659         tarobj = TarFile.open("sample.tar",
 660                               mode="r",
 661                               new_volume_handler=new_volume_handler)
 662         tarobj.extractall()
 663         tarobj.close()
 664
 665         for key, value in hash.items():
 666             assert os.path.exists(key)
 667             assert value == self.md5sum(key)
 668
 669     def test_corner_case_split_size5(self):
 670         '''
 671         Creates a tar file with multiple files inside that contains the maximum
 672         size allowed in one volume.
 673         '''
 674         hash = dict()
 675         hash['big'] = self.create_file("big", 3*1024*1024)
 676         hash['small'] = self.create_file("small", 1*1024*1024)
 677
 678         # create the tar file with volumes
 679         tarobj = TarFile.open("sample.tar",
 680                               mode="w",
 681                               format=self.tarfile_format,
 682                               max_volume_size=(2*1024*1024 +
 683                                                self.tarfile_overhead +
 684                                                self.file_overhead),
 685                               new_volume_handler=new_volume_handler)
 686         tarobj.add("big")
 687         tarobj.add("small")
 688         tarobj.close()
 689
 690         # check that the tar volumes were correctly created
 691         assert os.path.exists("sample.tar")
 692         assert os.path.exists("sample.tar.1")
 693         assert not os.path.exists("sample.tar.2")
 694
 695         for key, value in hash.items():
 696             os.unlink(key)
 697             assert not os.path.exists(key)
 698
 699         # extract and check output
 700         tarobj = TarFile.open("sample.tar",
 701                               mode="r",
 702                               new_volume_handler=new_volume_handler)
 703         tarobj.extractall()
 704         tarobj.close()
 705
 706         for key, value in hash.items():
 707             assert os.path.exists(key)
 708             assert value == self.md5sum(key)
 709
 710     def test_volume_not_found(self):
 711         '''
 712         Create a tar file with multiple volumes and one file and extract it, but
 713         one of the volumes is missing
 714         '''
 715         # create the content of the file to compress and hash it
 716         hash = self.create_file("big", 5*1024*1024)
 717
 718         # create the tar file with volumes
 719         tarobj = TarFile.open("sample.tar",
 720                               mode="w",
 721                               format=self.tarfile_format,
 722                               max_volume_size=2*1024*1024,
 723                               new_volume_handler=closing_new_volume_handler)
 724         tarobj.add("big")
 725         tarobj.close()
 726
 727         # check that the tar volumes were correctly created
 728         assert os.path.exists("sample.tar")
 729         assert os.path.exists("sample.tar.1")
 730         assert os.path.exists("sample.tar.2")
 731         assert not os.path.exists("sample.tar.3")
 732
 733         os.unlink("big")
 734         os.unlink("sample.tar.2")
 735
 736         class VolumeNotFound(Exception):
 737             pass
 738
 739         def new_volume_handler2(tarobj, base_name, volume_number):
 740             '''
 741             Handles the new volumes
 742             '''
 743             volume_path = "%s.%d" % (base_name, volume_number)
 744
 745             try:
 746                 tarobj.fileobj.close()
 747                 tarobj.open_volume(volume_path)
 748             except OSError as e:
 749                 # only volume number 2 is missing
 750                 assert volume_number == 2
 751                 raise e
 752
 753         # extract and check output
 754         tarobj = TarFile.open("sample.tar",
 755                               mode="r",
 756                               new_volume_handler=new_volume_handler2)
 757         try:
 758             tarobj.extractall()
 759         except OSError as e:
 760             pass
 761         tarobj.close()
 762         assert os.path.exists("big")
 763         assert hash != self.md5sum("big")
 764
 765     def test_multivol_compress_vol_size(self):
 766         """ test size of compressed volumes using "external" test routine
 767
 768         created an extensive test of this in extra file, run here just 2 short
 769         versions
 770         """
 771         # params
 772         vol_size = 3   # MiB
 773         input_size_factor = 3   # --> add 3*3 MiB of data
 774         modes = ('w#gz', None), ('w#gz', "test1234")
 775         debug_level = 0   # no debug output
 776         clean_up_if_error = True   # leave no files behind
 777
 778         with TemporaryDirectory(prefix='deltatar_test_multivol_') \
 779                 as temp_dir:     # is deleted automatically after test
 780             for mode, password in modes:
 781                 multivol_compr_test_func(vol_size, input_size_factor, mode,
 782                                          password,
 783                                          temp_dir,
 784                                          debug_level=debug_level,
 785                                          clean_up_if_error=clean_up_if_error)
 786
 787
 788 class MultivolPaxFormatTest(MultivolGnuFormatTest):
 789     """
 790     Test multivolume support in tarfile with PAX format
 791     """
 792
 793     tar_command_format = "pax"
 794
 795     tarfile_format = PAX_FORMAT
 796
 797     # overhead size used to calculate the exact maximum size of a tar file with
 798     # no extra volume that stores only one file. In case of GNU format this is
 799     # the size of three blocks:
 800     # * 1 block used to store the header information of the stored file
 801     # * 1 block used to store the header information of the pax header
 802     # * 1 block used to store the pax header
 803     # * 2 blocks used to mark the end of the tar file
 804     tarfile_overhead = 5*BLOCKSIZE
 805     file_overhead = 3*BLOCKSIZE
 806
 807     # overhead size used to calculate the exact maximum size of a tar volume,
 808     # corresponding with a multivolume tar file storing a single file. In the
 809     # case of Pax format, it's the same as tarfile_overhead plus a block for
 810     # the global header
 811     tarvol_overhead = 6*BLOCKSIZE