import unittest, os, re, sys, time from commontest import * from rdiff_backup import Globals, log, rpath, robust, FilenameMapping """Regression tests""" Globals.exclude_mirror_regexps = [re.compile(".*/rdiff-backup-data")] log.Log.setverbosity(3) lc = Globals.local_connection class Local: """This is just a place to put increments relative to the local connection""" def get_local_rp(extension): return rpath.RPath(Globals.local_connection, "testfiles/" + extension) vftrp = get_local_rp('various_file_types') inc1rp = get_local_rp('increment1') inc2rp = get_local_rp('increment2') inc3rp = get_local_rp('increment3') inc4rp = get_local_rp('increment4') rpout = get_local_rp('output') rpout_inc = get_local_rp('output_inc') rpout1 = get_local_rp('restoretarget1') rpout2 = get_local_rp('restoretarget2') rpout3 = get_local_rp('restoretarget3') rpout4 = get_local_rp('restoretarget4') prefix = get_local_rp('.') vft_in = get_local_rp('vft_out') vft_out = get_local_rp('increment2/various_file_types') vft2_in = get_local_rp('vft2_out') timbar_in = get_local_rp('increment1/timbar.pyc') timbar_out = get_local_rp('../timbar.pyc') # in cur directory class PathSetter(unittest.TestCase): def setUp(self): self.reset_schema() def reset_schema(self): self.rb_schema = (SourceDir + "/../rdiff-backup -v3 --no-compare-inode " "--remote-schema './chdir-wrapper2 %s' ") def refresh(self, *rp_list): """Reread data for the given rps""" for rp in rp_list: rp.setdata() def set_connections(self, src_pre, src_back, dest_pre, dest_back): """Set source and destination prefixes""" if src_pre: self.src_prefix = "%s::%s" % (src_pre, src_back) else: self.src_prefix = './' if dest_pre: self.dest_prefix = "%s::%s" % (dest_pre, dest_back) else: self.dest_prefix = './' def exec_rb(self, time, *args): """Run rdiff-backup on given arguments""" arglist = [] if time: arglist.extend(["--current-time", str(time)]) arglist.append(self.src_prefix + args[0]) if len(args) > 1: arglist.append(self.dest_prefix + args[1]) assert len(args) == 2 argstring = ' '.join(map(lambda s: "'%s'" % (s,), arglist)) cmdstr = self.rb_schema + argstring print "executing " + cmdstr assert not os.system(cmdstr) def exec_rb_extra_args(self, time, extra_args, *args): """Run rdiff-backup on given arguments""" arglist = [] if time: arglist.extend(["--current-time", str(time)]) arglist.append(self.src_prefix + args[0]) if len(args) > 1: arglist.append(self.dest_prefix + args[1]) assert len(args) == 2 cmdstr = "%s %s %s" % (self.rb_schema, extra_args, ' '.join(arglist)) print "executing " + cmdstr assert not os.system(cmdstr) def exec_rb_restore(self, time, *args): """Restore using rdiff-backup's new syntax and given time""" arglist = [] arglist.append("--restore-as-of %s" % str(time)) arglist.append(self.src_prefix + args[0]) if len(args) > 1: arglist.append(self.dest_prefix + args[1]) assert len(args) == 2 cmdstr = self.rb_schema + " ".join(arglist) print "Restoring via cmdline: " + cmdstr assert not os.system(cmdstr) def delete_tmpdirs(self): """Remove any temp directories created by previous tests""" assert not os.system(MiscDir + '/myrm testfiles/output* ' 'testfiles/restoretarget* testfiles/vft_out ' 'timbar.pyc testfiles/vft2_out') def runtest(self): self.delete_tmpdirs() # Backing up increment1 self.exec_rb(10000, 'testfiles/increment1', 'testfiles/output') assert CompareRecursive(Local.inc1rp, Local.rpout) time.sleep(1) # Backing up increment2 self.exec_rb(20000, 'testfiles/increment2', 'testfiles/output') assert CompareRecursive(Local.inc2rp, Local.rpout) time.sleep(1) # Backing up increment3 self.exec_rb(30000, 'testfiles/increment3', 'testfiles/output') assert CompareRecursive(Local.inc3rp, Local.rpout) time.sleep(1) # Backing up increment4 self.exec_rb(40000, 'testfiles/increment4', 'testfiles/output') assert CompareRecursive(Local.inc4rp, Local.rpout) # Getting restore rps inc_paths = self.getinc_paths("increments.", "testfiles/output/rdiff-backup-data") assert len(inc_paths) == 3 # Restoring increment1 self.exec_rb(None, inc_paths[0], 'testfiles/restoretarget1') assert CompareRecursive(Local.inc1rp, Local.rpout1) # Restoring increment2 self.exec_rb(None, inc_paths[1], 'testfiles/restoretarget2') assert CompareRecursive(Local.inc2rp, Local.rpout2) # Restoring increment3 self.exec_rb(None, inc_paths[2], 'testfiles/restoretarget3') assert CompareRecursive(Local.inc3rp, Local.rpout3) # Test restoration of a few random files vft_paths = self.getinc_paths("various_file_types.", "testfiles/output/rdiff-backup-data/increments") self.exec_rb(None, vft_paths[1], 'testfiles/vft_out') self.refresh(Local.vft_in, Local.vft_out) assert CompareRecursive(Local.vft_in, Local.vft_out) timbar_paths = self.getinc_paths("timbar.pyc.", "testfiles/output/rdiff-backup-data/increments") self.exec_rb(None, timbar_paths[0]) self.refresh(Local.timbar_in, Local.timbar_out) assert rpath.cmp_with_attribs(Local.timbar_in, Local.timbar_out) self.exec_rb_restore(25000, 'testfiles/output/various_file_types', 'testfiles/vft2_out') self.refresh(Local.vft2_in, Local.vft_out) assert CompareRecursive(Local.vft2_in, Local.vft_out) # Make sure too many increment files not created assert len(self.getinc_paths("nochange.", "testfiles/output/rdiff-backup-data/increments")) == 0 nochange_incs = len(self.getinc_paths("", "testfiles/output/rdiff-backup-data/increments/nochange")) assert nochange_incs == 1 or nochange_incs == 0, nochange_incs def getinc_paths(self, basename, directory, quoted = 0): """Return increment.______.dir paths""" if quoted: FilenameMapping.set_init_quote_vals() dirrp = FilenameMapping.QuotedRPath(Globals.local_connection, directory) else: dirrp = rpath.RPath(Globals.local_connection, directory) incbasenames = [filename for filename in robust.listrp(dirrp) if filename.startswith(basename)] incbasenames.sort() incrps = map(dirrp.append, incbasenames) return map(lambda x: x.path, filter(lambda incrp: incrp.isincfile(), incrps)) class Final(PathSetter): def testLocal(self): """Run test sequence everything local""" self.set_connections(None, None, None, None) self.runtest() def testRemoteAll(self): """Run test sequence everything remote""" self.set_connections("test1/", '../', 'test2/tmp/', '../../') self.runtest() def testRemoteSource(self): """Run test sequence when remote side is source""" self.set_connections("test1/", "../", None, None) self.runtest() def testRemoteDest(self): """Run test sequence when remote side is destination""" self.set_connections(None, None, "test2/tmp", "../../") self.runtest() def testProcLocal(self): """Test initial backup of /proc locally""" Myrm("testfiles/procoutput") self.set_connections(None, None, None, None) self.exec_rb(None, '../../../../../../proc', 'testfiles/procoutput') def testProcRemote(self): """Test mirroring proc remote""" Myrm("testfiles/procoutput") self.set_connections(None, None, "test2/tmp/", "../../") self.exec_rb(None, '../../../../../../proc', 'testfiles/procoutput') def testProcRemote2(self): """Test mirroring proc, this time when proc is remote, dest local""" Myrm("testfiles/procoutput") self.set_connections("test1/", "../", None, None) self.exec_rb(None, '../../../../../../proc', 'testfiles/procoutput') def testWindowsMode(self): """Test backup with the --windows-mode option""" old_schema = self.rb_schema self.rb_schema = old_schema + " --windows-mode " self.set_connections(None, None, None, None) self.delete_tmpdirs() # Back up increment2, this contains a file with colons self.exec_rb(20000, 'testfiles/increment2', 'testfiles/output') time.sleep(1) # Back up increment3 self.exec_rb(30000, 'testfiles/increment3', 'testfiles/output') Globals.time_separator = "_" inc_paths = self.getinc_paths("increments.", "testfiles/output/rdiff-backup-data", 1) Globals.time_separator = ":" assert len(inc_paths) == 1, inc_paths # Restore increment2 self.exec_rb(None, inc_paths[0], 'testfiles/restoretarget2') assert CompareRecursive(Local.inc2rp, Local.rpout2, compare_hardlinks = 0) # Now check to make sure no ":" in output directory popen_fp = os.popen("find testfiles/output -name '*:*' | wc") wc_output = popen_fp.read() popen_fp.close() assert wc_output.split() == ["0", "0", "0"], wc_output class FinalSelection(PathSetter): """Test selection options""" def run(self, cmd): print "Executing: ", cmd assert not os.system(cmd) def testSelLocal(self): """Quick backup testing a few selection options""" self.delete_tmpdirs() # Test --include option assert not \ os.system(self.rb_schema + "--current-time 10000 " "--include testfiles/increment2/various_file_types " "--exclude '**' " "testfiles/increment2 testfiles/output") assert os.lstat("testfiles/output/various_file_types/regular_file") self.assertRaises(OSError, os.lstat, "testfiles/output/test.py") # Now try reading list of files fp = os.popen(self.rb_schema + "--current-time 20000 " "--include-filelist-stdin --exclude '**' " "testfiles/increment2 testfiles/output", "w") fp.write(""" testfiles/increment2/test.py testfiles/increment2/changed_dir""") assert not fp.close() assert os.lstat("testfiles/output/changed_dir") assert os.lstat("testfiles/output/test.py") self.assertRaises(OSError, os.lstat, "testfiles/output/various_file_types") self.assertRaises(OSError, os.lstat, "testfiles/output/changed_dir/foo") # Test selective restoring mirror_rp = rpath.RPath(Globals.local_connection, "testfiles/output") restore_filename = get_increment_rp(mirror_rp, 10000).path self.run(self.rb_schema + "--include testfiles/restoretarget1/various_file_types/" "regular_file " "--exclude '**' " + restore_filename + " testfiles/restoretarget1") assert os.lstat("testfiles/restoretarget1/various_file_types/" "regular_file") self.assertRaises(OSError, os.lstat, "testfiles/restoretarget1/tester") self.assertRaises(OSError, os.lstat, "testfiles/restoretarget1/various_file_types/executable") fp = os.popen(self.rb_schema + "--include-filelist-stdin " + restore_filename + " testfiles/restoretarget2", "w") fp.write(""" - testfiles/restoretarget2/various_file_types/executable""") assert not fp.close() assert os.lstat("testfiles/restoretarget2/various_file_types/" "regular_file") self.assertRaises(OSError, os.lstat, "testfiles/restoretarget2/various_file_types/executable") def testSelFilesRemote(self): """Test for bug found in 0.7.[34] - filelist where source remote""" self.delete_tmpdirs() self.set_connections("test1/", "../", 'test2/tmp/', '../../') self.rb_schema += ("--exclude-filelist testfiles/vft_out/exclude " "--include-filelist testfiles/vft_out/include " "--exclude '**' ") # Make an exclude list os.mkdir("testfiles/vft_out") excluderp = rpath.RPath(Globals.local_connection, "testfiles/vft_out/exclude") fp = excluderp.open("w") fp.write(""" ../testfiles/various_file_types/regular_file ../testfiles/various_file_types/test """) assert not fp.close() # Make an include list includerp = rpath.RPath(Globals.local_connection, "testfiles/vft_out/include") fp = includerp.open("w") fp.write(""" ../testfiles/various_file_types/executable ../testfiles/various_file_types/symbolic_link ../testfiles/various_file_types/regular_file ../testfiles/various_file_types/test """) assert not fp.close() self.exec_rb(None, 'testfiles/various_file_types', 'testfiles/output') self.reset_schema() self.exec_rb_restore("now", 'testfiles/output', 'testfiles/restoretarget1') assert os.lstat('testfiles/restoretarget1/executable') assert os.lstat('testfiles/restoretarget1/symbolic_link') self.assertRaises(OSError, os.lstat, 'testfiles/restoretarget1/regular_file') self.assertRaises(OSError, os.lstat, 'testfiles/restoretarget1/executable2') if __name__ == "__main__": unittest.main()