importosimporttempfilefromos.pathimportjoin,existsfromrsync_backupimportbackup,rsync,dirshiftdefrf(fname):"""read the content of a file and return it"""withopen(fname,"rb")asf:returnf.read()defwf(fname,content):"""write content into a file"""withopen(fname,"wb")asf:f.write(content)deftest_flat():src=tempfile.mkdtemp()dst=tempfile.mkdtemp()wf(join(src,"1"),"eins")wf(join(src,"2"),"zwei")rsync(src,dst,src)assertrf(join(dst,"1"))=="eins"assertrf(join(dst,"2"))=="zwei"deftest_recursive():src=tempfile.mkdtemp()dst=tempfile.mkdtemp()wf(join(src,"1"),"eins")src_sub=join(src,"sub")os.mkdir(src_sub)wf(join(src_sub,"2"),"zwei")rsync(src,dst,src)assertrf(join(dst,"1"))=="eins"assertrf(join(dst,"sub","2"))=="zwei"deftest_dirshift():tmp=tempfile.mkdtemp()N=10foriinrange(N):# 0..N-1os.mkdir(join(tmp,str(i)))wf(join(tmp,str(i),str(i)),"")dirshift(tmp,N,"")assertexists(join(tmp,"0"))foriinrange(1,N):# 1..N-1assertexists(join(tmp,str(i),str(i-1)))assertnotexists(join(tmp,str(N)))deftest_backup():src=tempfile.mkdtemp()dst=tempfile.mkdtemp()N=3backup(src,dst,N)assertos.listdir(join(dst,"0"))==[]wf(join(src,"1"),"eins")wf(join(src,"2"),"zwei")backup(src,dst,N)assertsorted(os.listdir(join(dst,"0")))==["1","2"]assertrf(join(dst,"0","1"))=="eins"assertrf(join(dst,"0","2"))=="zwei"wf(join(src,"1"),"einseins")backup(src,dst,N)assertsorted(os.listdir(join(dst,"0")))==["1","2"]assertrf(join(dst,"0","1"))=="einseins"assertrf(join(dst,"0","2"))=="zwei"st=os.stat(join(dst,"0","1"))assertst.st_nlink==1st=os.stat(join(dst,"0","2"))assertst.st_nlink==2os.remove(join(src,"2"))backup(src,dst,N)assertsorted(os.listdir(join(dst,"0")))==["1"]assertrf(join(dst,"0","1"))=="einseins"st=os.stat(join(dst,"1","2"))assertst.st_nlink==2