Goffi's cp, a fancy file copier
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

test_gcp.py 7.9KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226
  1. #!/usr/bin/python
  2. # -*- coding: utf-8 -*-
  3. """
  4. gcp: Goffi's CoPier
  5. Copyright (C) 2010, 2011 Jérôme Poisson <goffi@goffi.org>
  6. This program is free software: you can redistribute it and/or modify
  7. it under the terms of the GNU General Public License as published by
  8. the Free Software Foundation, either version 3 of the License, or
  9. (at your option) any later version.
  10. This program is distributed in the hope that it will be useful,
  11. but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. GNU General Public License for more details.
  14. You should have received a copy of the GNU General Public License
  15. along with this program. If not, see <http://www.gnu.org/licenses/>.
  16. """
  17. from __future__ import with_statement
  18. import tempfile
  19. import unittest
  20. from os import getcwd, chdir, system, mkdir, makedirs, listdir
  21. from os.path import join, isdir
  22. from shutil import rmtree
  23. from random import randrange
  24. from hashlib import sha1
  25. #size shorcuts
  26. S10K = 1024 * 10
  27. S100K = 1024 * 100
  28. S1M = 1024 * 1024
  29. S10M = 1024 * 1024 * 10
  30. S100M = 1024 * 1024 * 100
  31. def sha1sum(filename, buf_size=4096):
  32. """Return the SHA1 hash of a file
  33. @param filename: path to the file
  34. @param buf_size: size of the buffer to use for calculation"""
  35. csum = sha1()
  36. with open(filename) as fd:
  37. data = fd.read(buf_size)
  38. while data:
  39. csum.update(data)
  40. data = fd.read(buf_size)
  41. return csum.digest()
  42. def dirCheck(dir_path):
  43. """Recursively calculate SHA1 sum of a dir
  44. @param path: path of the dir to check
  45. @return: a dict in the form [{filepath: sum,...}]
  46. """
  47. def recursive_sum(directory, result):
  48. for current_path in listdir(directory):
  49. full_path = join(directory, current_path)
  50. if isdir(full_path):
  51. recursive_sum(full_path, result)
  52. else:
  53. result[full_path] = sha1sum(full_path)
  54. result = {}
  55. _ori_dir = getcwd()
  56. chdir(dir_path)
  57. recursive_sum(".", result)
  58. chdir(_ori_dir)
  59. return result
  60. #def makeRandomFile(path, size, buf_size=4096):
  61. # """Create a fake file
  62. # @param path: where the file is created
  63. # @param size: size of the file to create in bytes"""
  64. # def seq(size):
  65. # return ''.join(chr(randrange(256)) for i in range(size))
  66. # fd = open(path, 'w')
  67. # for byte in range(size//buf_size):
  68. # fd.write(seq(buf_size))
  69. # fd.write(seq(size%buf_size))
  70. # fd.close()
  71. def makeRandomFile(path, size=S10K, buf_size=4096):
  72. """Create a fake file using /dev/urandom
  73. @param path: where the file must be created
  74. @param size: size of the file to create in bytes"""
  75. source = open('/dev/urandom','r')
  76. dest = open(path, 'w')
  77. for byte in range(size//buf_size):
  78. dest.write(source.read(buf_size))
  79. dest.write(source.read(size%buf_size))
  80. dest.close()
  81. def makeTestDir(path):
  82. """Helper method to easily create a test dir
  83. @param path: where the dir must be created"""
  84. for i in range(2):
  85. subdir = join(path,'subdir_%d' % i)
  86. makedirs(subdir)
  87. for j in range(2):
  88. makeRandomFile(join(subdir, 'file_%d' % j), S10K)
  89. for i in range(2):
  90. makeRandomFile(join(path,'file_%d' % i), S10K)
  91. class TestCopyCases(unittest.TestCase):
  92. """Test basic copy use cases, using gcp externally
  93. gcp must be available in the PATH"""
  94. #TODO: check journal
  95. def setUp(self):
  96. self.ori_dir = getcwd()
  97. self.tmp_dir = tempfile.mkdtemp()
  98. chdir(self.tmp_dir)
  99. def tearDown(self):
  100. chdir(self.ori_dir)
  101. rmtree(self.tmp_dir)
  102. def test_one_file_copy(self):
  103. """Copy one file and test the result"""
  104. makeRandomFile('file_1', S10K)
  105. ori_sum = sha1sum('file_1')
  106. ret = system("gcp file_1 file_2")
  107. self.assertEqual(ret,0)
  108. dest_sum = sha1sum('file_2')
  109. self.assertEqual(ori_sum, dest_sum)
  110. def test_one_file_copy_already_exists(self):
  111. """Check that an existing file is not overwritten"""
  112. makeRandomFile('file_1', S10K)
  113. makeRandomFile('file_2', S10K)
  114. file_2_sum = sha1sum('file_2')
  115. ret = system("gcp file_1 file_2")
  116. self.assertNotEqual(ret,0)
  117. file_2_sum_bis = sha1sum('file_2')
  118. self.assertEqual(file_2_sum, file_2_sum_bis)
  119. def test_one_file_copy_already_exists_force(self):
  120. """Check that an existing file is overwritten with --force"""
  121. makeRandomFile('file_1', S10K)
  122. makeRandomFile('file_2', S10K)
  123. file_1_sum = sha1sum('file_1')
  124. ret = system("gcp -f file_1 file_2")
  125. self.assertEqual(ret,0)
  126. file_2_sum_bis = sha1sum('file_2')
  127. self.assertEqual(file_1_sum, file_2_sum_bis)
  128. def test_one_dir_copy(self):
  129. """Check copy of one dir to a non existant path"""
  130. makeTestDir('dir_1')
  131. check_1 = dirCheck('dir_1')
  132. ret = system("gcp -r dir_1 dir_2")
  133. self.assertEqual(ret,0)
  134. check_2 = dirCheck('dir_2')
  135. self.assertEqual(check_1, check_2)
  136. def test_one_dir_copy_nocopy(self):
  137. """Check that a dir is not copied without the recursive option"""
  138. makeTestDir('dir_1')
  139. check_before = dirCheck('.')
  140. ret = system("gcp dir_1 dir_2")
  141. self.assertEqual(ret,0)
  142. check_after = dirCheck('.')
  143. self.assertEqual(check_before, check_after)
  144. def test_one_dir_copy_existing_dest(self):
  145. """Check that a dir is copied inside an existing destination"""
  146. makeTestDir('dir_1')
  147. mkdir('dir_2')
  148. check_1 = dirCheck('dir_1')
  149. ret = system("gcp -r dir_1 dir_2")
  150. self.assertEqual(ret,0)
  151. self.assertEqual(listdir('dir_2'), ['dir_1'])
  152. check_2 = dirCheck('dir_2/dir_1')
  153. self.assertEqual(check_1, check_2)
  154. def test_mixt_copy_existing_dest(self):
  155. """Check that a mixt copy (files + dir) to an existing dest work as expected"""
  156. for i in range(2):
  157. makeRandomFile('file_%d' % i, S10K)
  158. makeTestDir('dir_%d' % i)
  159. check_1 = dirCheck('.')
  160. mkdir('dest_dir')
  161. ret = system("gcp -r file_0 file_1 dir_0 dir_1 dest_dir")
  162. self.assertEqual(ret,0)
  163. check_2 = dirCheck('dest_dir')
  164. self.assertEqual(check_1, check_2)
  165. def test_mixt_copy_nonexisting_dest(self):
  166. """Check that a mixt copy (files + dir) to an non existing dest work as expected
  167. /!\\ the behavious is different of the one of cp in this case ! (cp doesn't copy at all, while gcp create the dest)"""
  168. for i in range(2):
  169. makeRandomFile('file_%d' % i, S10K)
  170. makeTestDir('dir_%d' % i)
  171. check_1 = dirCheck('.')
  172. ret = system("gcp -r file_0 file_1 dir_0 dir_1 dest_dir")
  173. self.assertEqual(ret,0)
  174. check_2 = dirCheck('dest_dir')
  175. self.assertEqual(check_1, check_2)
  176. def test_mixt_copy_existing_dest_nonrecursive(self):
  177. """Check that a mixt copy (files + dir) to an existing dest without the recursive option work as expected"""
  178. for i in range(2):
  179. makeRandomFile('file_%d' % i, S10K)
  180. makeTestDir('dir_%d' % i)
  181. mkdir('dest_dir')
  182. ret = system("gcp file_0 file_1 dir_0 dir_1 dest_dir")
  183. self.assertEqual(ret,0)
  184. self.assertEqual(set(listdir('dest_dir')), set(['file_0', 'file_1']))
  185. self.assertEqual(sha1sum('file_0'), sha1sum('dest_dir/file_0'))
  186. self.assertEqual(sha1sum('file_1'), sha1sum('dest_dir/file_1'))
  187. def test_mixt_copy_nonexisting_dest_nonrecursive(self):
  188. """Check that a mixt copy (files + dir) to an existing dest without the recursive option work as expected"""
  189. for i in range(2):
  190. makeRandomFile('file_%d' % i, S10K)
  191. makeTestDir('dir_%d' % i)
  192. check_before = dirCheck('.')
  193. ret = system("gcp file_0 file_1 dir_0 dir_1 dest_dir")
  194. self.assertEqual(ret >> 8, 1)
  195. check_after = dirCheck('.')
  196. self.assertEqual(check_before, check_after)
  197. if __name__ == '__main__':
  198. unittest.main()