1414from contextlib import ExitStack
1515from os .path import splitdrive
1616from distutils .spawn import find_executable , spawn
17- from shutil import (_make_tarball , _make_zipfile , make_archive ,
17+ from shutil import (make_archive ,
1818 register_archive_format , unregister_archive_format ,
1919 get_archive_formats , Error , unpack_archive ,
2020 register_unpack_format , RegistryError ,
@@ -92,6 +92,18 @@ def read_file(path, binary=False):
9292 with open (path , 'rb' if binary else 'r' ) as fp :
9393 return fp .read ()
9494
95+ def rlistdir (path ):
96+ res = []
97+ for name in sorted (os .listdir (path )):
98+ p = os .path .join (path , name )
99+ if os .path .isdir (p ) and not os .path .islink (p ):
100+ res .append (name + '/' )
101+ for n in rlistdir (p ):
102+ res .append (name + '/' + n )
103+ else :
104+ res .append (name )
105+ return res
106+
95107
96108class TestShutil (unittest .TestCase ):
97109
@@ -957,114 +969,105 @@ def test_copy2(self):
957969 @requires_zlib
958970 def test_make_tarball (self ):
959971 # creating something to tar
960- tmpdir = self .mkdtemp ()
961- write_file ((tmpdir , 'file1' ), 'xxx' )
962- write_file ((tmpdir , 'file2' ), 'xxx' )
963- os .mkdir (os .path .join (tmpdir , 'sub' ))
964- write_file ((tmpdir , 'sub' , 'file3' ), 'xxx' )
972+ root_dir , base_dir = self ._create_files ('' )
965973
966974 tmpdir2 = self .mkdtemp ()
967975 # force shutil to create the directory
968976 os .rmdir (tmpdir2 )
969- unittest .skipUnless (splitdrive (tmpdir )[0 ] == splitdrive (tmpdir2 )[0 ],
977+ unittest .skipUnless (splitdrive (root_dir )[0 ] == splitdrive (tmpdir2 )[0 ],
970978 "source and target should be on same drive" )
971979
972980 base_name = os .path .join (tmpdir2 , 'archive' )
973981
974982 # working with relative paths to avoid tar warnings
975- with support .change_cwd (tmpdir ):
976- _make_tarball (splitdrive (base_name )[1 ], '.' )
983+ make_archive (splitdrive (base_name )[1 ], 'gztar' , root_dir , '.' )
977984
978985 # check if the compressed tarball was created
979986 tarball = base_name + '.tar.gz'
980- self .assertTrue (os .path .exists (tarball ))
987+ self .assertTrue (os .path .isfile (tarball ))
988+ self .assertTrue (tarfile .is_tarfile (tarball ))
989+ with tarfile .open (tarball , 'r:gz' ) as tf :
990+ self .assertCountEqual (tf .getnames (),
991+ ['.' , './sub' , './sub2' ,
992+ './file1' , './file2' , './sub/file3' ])
981993
982994 # trying an uncompressed one
983995 base_name = os .path .join (tmpdir2 , 'archive' )
984- with support .change_cwd (tmpdir ):
985- _make_tarball (splitdrive (base_name )[1 ], '.' , compress = None )
996+ make_archive (splitdrive (base_name )[1 ], 'tar' , root_dir , '.' )
986997 tarball = base_name + '.tar'
987- self .assertTrue (os .path .exists (tarball ))
998+ self .assertTrue (os .path .isfile (tarball ))
999+ self .assertTrue (tarfile .is_tarfile (tarball ))
1000+ with tarfile .open (tarball , 'r' ) as tf :
1001+ self .assertCountEqual (tf .getnames (),
1002+ ['.' , './sub' , './sub2' ,
1003+ './file1' , './file2' , './sub/file3' ])
9881004
9891005 def _tarinfo (self , path ):
990- tar = tarfile .open (path )
991- try :
1006+ with tarfile .open (path ) as tar :
9921007 names = tar .getnames ()
9931008 names .sort ()
9941009 return tuple (names )
995- finally :
996- tar .close ()
9971010
998- def _create_files (self ):
1011+ def _create_files (self , base_dir = 'dist' ):
9991012 # creating something to tar
1000- tmpdir = self .mkdtemp ()
1001- dist = os .path .join (tmpdir , 'dist' )
1002- os .mkdir (dist )
1013+ root_dir = self .mkdtemp ()
1014+ dist = os .path .join (root_dir , base_dir )
1015+ os .makedirs (dist , exist_ok = True )
10031016 write_file ((dist , 'file1' ), 'xxx' )
10041017 write_file ((dist , 'file2' ), 'xxx' )
10051018 os .mkdir (os .path .join (dist , 'sub' ))
10061019 write_file ((dist , 'sub' , 'file3' ), 'xxx' )
10071020 os .mkdir (os .path .join (dist , 'sub2' ))
1008- tmpdir2 = self . mkdtemp ()
1009- base_name = os . path . join ( tmpdir2 , 'archive ' )
1010- return tmpdir , tmpdir2 , base_name
1021+ if base_dir :
1022+ write_file (( root_dir , 'outer' ), 'xxx ' )
1023+ return root_dir , base_dir
10111024
10121025 @requires_zlib
1013- @unittest .skipUnless (find_executable ('tar' ) and find_executable ( 'gzip' ) ,
1026+ @unittest .skipUnless (find_executable ('tar' ),
10141027 'Need the tar command to run' )
10151028 def test_tarfile_vs_tar (self ):
1016- tmpdir , tmpdir2 , base_name = self ._create_files ()
1017- with support . change_cwd ( tmpdir ):
1018- _make_tarball (base_name , 'dist' )
1029+ root_dir , base_dir = self ._create_files ()
1030+ base_name = os . path . join ( self . mkdtemp (), 'archive' )
1031+ make_archive (base_name , 'gztar' , root_dir , base_dir )
10191032
10201033 # check if the compressed tarball was created
10211034 tarball = base_name + '.tar.gz'
1022- self .assertTrue (os .path .exists (tarball ))
1035+ self .assertTrue (os .path .isfile (tarball ))
10231036
10241037 # now create another tarball using `tar`
1025- tarball2 = os .path .join (tmpdir , 'archive2.tar.gz' )
1026- tar_cmd = ['tar' , '-cf' , 'archive2.tar' , 'dist' ]
1027- gzip_cmd = ['gzip' , '-f9' , 'archive2.tar' ]
1028- with support .change_cwd (tmpdir ):
1029- with captured_stdout () as s :
1030- spawn (tar_cmd )
1031- spawn (gzip_cmd )
1032-
1033- self .assertTrue (os .path .exists (tarball2 ))
1038+ tarball2 = os .path .join (root_dir , 'archive2.tar' )
1039+ tar_cmd = ['tar' , '-cf' , 'archive2.tar' , base_dir ]
1040+ with support .change_cwd (root_dir ), captured_stdout ():
1041+ spawn (tar_cmd )
1042+
1043+ self .assertTrue (os .path .isfile (tarball2 ))
10341044 # let's compare both tarballs
10351045 self .assertEqual (self ._tarinfo (tarball ), self ._tarinfo (tarball2 ))
10361046
10371047 # trying an uncompressed one
1038- base_name = os .path .join (tmpdir2 , 'archive' )
1039- with support .change_cwd (tmpdir ):
1040- _make_tarball (base_name , 'dist' , compress = None )
1048+ make_archive (base_name , 'tar' , root_dir , base_dir )
10411049 tarball = base_name + '.tar'
1042- self .assertTrue (os .path .exists (tarball ))
1050+ self .assertTrue (os .path .isfile (tarball ))
10431051
10441052 # now for a dry_run
1045- base_name = os .path .join (tmpdir2 , 'archive' )
1046- with support .change_cwd (tmpdir ):
1047- _make_tarball (base_name , 'dist' , compress = None , dry_run = True )
1053+ make_archive (base_name , 'tar' , root_dir , base_dir , dry_run = True )
10481054 tarball = base_name + '.tar'
1049- self .assertTrue (os .path .exists (tarball ))
1055+ self .assertTrue (os .path .isfile (tarball ))
10501056
10511057 @requires_zlib
10521058 @unittest .skipUnless (ZIP_SUPPORT , 'Need zip support to run' )
10531059 def test_make_zipfile (self ):
1054- # creating something to tar
1055- tmpdir = self .mkdtemp ()
1056- write_file ((tmpdir , 'file1' ), 'xxx' )
1057- write_file ((tmpdir , 'file2' ), 'xxx' )
1058-
1059- tmpdir2 = self .mkdtemp ()
1060- # force shutil to create the directory
1061- os .rmdir (tmpdir2 )
1062- base_name = os .path .join (tmpdir2 , 'archive' )
1063- _make_zipfile (base_name , tmpdir )
1060+ # creating something to zip
1061+ root_dir , base_dir = self ._create_files ()
1062+ base_name = os .path .join (self .mkdtemp (), 'archive' )
1063+ res = make_archive (base_name , 'zip' , root_dir , 'dist' )
10641064
1065- # check if the compressed tarball was created
1066- tarball = base_name + '.zip'
1067- self .assertTrue (os .path .exists (tarball ))
1065+ self .assertEqual (res , base_name + '.zip' )
1066+ self .assertTrue (os .path .isfile (res ))
1067+ self .assertTrue (zipfile .is_zipfile (res ))
1068+ with zipfile .ZipFile (res ) as zf :
1069+ self .assertCountEqual (zf .namelist (),
1070+ ['dist/file1' , 'dist/file2' , 'dist/sub/file3' ])
10681071
10691072
10701073 def test_make_archive (self ):
@@ -1082,36 +1085,37 @@ def test_make_archive_owner_group(self):
10821085 else :
10831086 group = owner = 'root'
10841087
1085- base_dir , root_dir , base_name = self ._create_files ()
1086- base_name = os .path .join (self .mkdtemp () , 'archive' )
1088+ root_dir , base_dir = self ._create_files ()
1089+ base_name = os .path .join (self .mkdtemp (), 'archive' )
10871090 res = make_archive (base_name , 'zip' , root_dir , base_dir , owner = owner ,
10881091 group = group )
1089- self .assertTrue (os .path .exists (res ))
1092+ self .assertTrue (os .path .isfile (res ))
10901093
10911094 res = make_archive (base_name , 'zip' , root_dir , base_dir )
1092- self .assertTrue (os .path .exists (res ))
1095+ self .assertTrue (os .path .isfile (res ))
10931096
10941097 res = make_archive (base_name , 'tar' , root_dir , base_dir ,
10951098 owner = owner , group = group )
1096- self .assertTrue (os .path .exists (res ))
1099+ self .assertTrue (os .path .isfile (res ))
10971100
10981101 res = make_archive (base_name , 'tar' , root_dir , base_dir ,
10991102 owner = 'kjhkjhkjg' , group = 'oihohoh' )
1100- self .assertTrue (os .path .exists (res ))
1103+ self .assertTrue (os .path .isfile (res ))
11011104
11021105
11031106 @requires_zlib
11041107 @unittest .skipUnless (UID_GID_SUPPORT , "Requires grp and pwd support" )
11051108 def test_tarfile_root_owner (self ):
1106- tmpdir , tmpdir2 , base_name = self ._create_files ()
1109+ root_dir , base_dir = self ._create_files ()
1110+ base_name = os .path .join (self .mkdtemp (), 'archive' )
11071111 group = grp .getgrgid (0 )[0 ]
11081112 owner = pwd .getpwuid (0 )[0 ]
1109- with support .change_cwd (tmpdir ):
1110- archive_name = _make_tarball (base_name , 'dist ' , compress = None ,
1111- owner = owner , group = group )
1113+ with support .change_cwd (root_dir ):
1114+ archive_name = make_archive (base_name , 'gztar ' , root_dir , 'dist' ,
1115+ owner = owner , group = group )
11121116
11131117 # check if the compressed tarball was created
1114- self .assertTrue (os .path .exists (archive_name ))
1118+ self .assertTrue (os .path .isfile (archive_name ))
11151119
11161120 # now checks the rights
11171121 archive = tarfile .open (archive_name )
@@ -1168,18 +1172,6 @@ def test_register_archive_format(self):
11681172 formats = [name for name , params in get_archive_formats ()]
11691173 self .assertNotIn ('xxx' , formats )
11701174
1171- def _compare_dirs (self , dir1 , dir2 ):
1172- # check that dir1 and dir2 are equivalent,
1173- # return the diff
1174- diff = []
1175- for root , dirs , files in os .walk (dir1 ):
1176- for file_ in files :
1177- path = os .path .join (root , file_ )
1178- target_path = os .path .join (dir2 , os .path .split (path )[- 1 ])
1179- if not os .path .exists (target_path ):
1180- diff .append (file_ )
1181- return diff
1182-
11831175 @requires_zlib
11841176 def test_unpack_archive (self ):
11851177 formats = ['tar' , 'gztar' , 'zip' ]
@@ -1188,22 +1180,24 @@ def test_unpack_archive(self):
11881180 if LZMA_SUPPORTED :
11891181 formats .append ('xztar' )
11901182
1183+ root_dir , base_dir = self ._create_files ()
11911184 for format in formats :
1192- tmpdir = self .mkdtemp ()
1193- base_dir , root_dir , base_name = self ._create_files ()
1194- tmpdir2 = self .mkdtemp ()
1185+ expected = rlistdir (root_dir )
1186+ expected .remove ('outer' )
1187+ if format == 'zip' :
1188+ expected .remove ('dist/sub2/' )
1189+ base_name = os .path .join (self .mkdtemp (), 'archive' )
11951190 filename = make_archive (base_name , format , root_dir , base_dir )
11961191
11971192 # let's try to unpack it now
1193+ tmpdir2 = self .mkdtemp ()
11981194 unpack_archive (filename , tmpdir2 )
1199- diff = self ._compare_dirs (tmpdir , tmpdir2 )
1200- self .assertEqual (diff , [])
1195+ self .assertEqual (rlistdir (tmpdir2 ), expected )
12011196
12021197 # and again, this time with the format specified
12031198 tmpdir3 = self .mkdtemp ()
12041199 unpack_archive (filename , tmpdir3 , format = format )
1205- diff = self ._compare_dirs (tmpdir , tmpdir3 )
1206- self .assertEqual (diff , [])
1200+ self .assertEqual (rlistdir (tmpdir3 ), expected )
12071201 self .assertRaises (shutil .ReadError , unpack_archive , TESTFN )
12081202 self .assertRaises (ValueError , unpack_archive , TESTFN , format = 'xxx' )
12091203
0 commit comments