1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
import unittest, random
from commontest import *
import librsync
def MakeRandomFile(path):
"""Writes a random file of length between 10000 and 100000"""
fp = open(path, "w")
randseq = []
for i in xrange(random.randrange(5000, 30000)):
randseq.append(chr(random.randrange(256)))
fp.write("".join(randseq))
fp.close()
class LibrsyncTest(unittest.TestCase):
"""Test various librsync wrapper functions"""
basis = RPath(Globals.local_connection, "testfiles/basis")
new = RPath(Globals.local_connection, "testfiles/new")
new2 = RPath(Globals.local_connection, "testfiles/new2")
sig = RPath(Globals.local_connection, "testfiles/signature")
sig2 = RPath(Globals.local_connection, "testfiles/signature2")
delta = RPath(Globals.local_connection, "testfiles/delta")
def testSigFile(self):
"""Make sure SigFile generates same data as rdiff"""
for i in range(5):
MakeRandomFile(self.basis.path)
self.sig.delete()
assert not os.system("rdiff signature %s %s" %
(self.basis.path, self.sig.path))
fp = self.sig.open("rb")
rdiff_sig = fp.read()
fp.close()
sf = librsync.SigFile(self.basis.open("rb"))
librsync_sig = sf.read()
sf.close()
assert rdiff_sig == librsync_sig, \
(len(rdiff_sig), len(librsync_sig))
def testSigGenerator(self):
"""Test SigGenerator, make sure it's same as SigFile"""
for i in range(5):
MakeRandomFile(self.basis.path)
sf = librsync.SigFile(self.basis.open("rb"))
sigfile_string = sf.read()
sf.close()
sig_gen = librsync.SigGenerator()
infile = self.basis.open("rb")
while 1:
buf = infile.read(1000)
if not buf: break
sig_gen.update(buf)
siggen_string = sig_gen.getsig()
assert sigfile_string == siggen_string, \
(len(sigfile_string), len(siggen_string))
def OldtestDelta(self):
"""Test delta generation against Rdiff"""
MakeRandomFile(self.basis.path)
assert not os.system("rdiff signature %s %s" %
(self.basis.path, self.sig.path))
for i in range(5):
MakeRandomFile(self.new.path)
assert not os.system("rdiff delta %s %s %s" %
(self.sig.path, self.new.path, self.delta.path))
fp = self.delta.open("rb")
rdiff_delta = fp.read()
fp.close()
df = librsync.DeltaFile(self.sig.open("rb"), self.new.open("rb"))
librsync_delta = df.read()
df.close()
print len(rdiff_delta), len(librsync_delta)
print repr(rdiff_delta[:100])
print repr(librsync_delta[:100])
assert rdiff_delta == librsync_delta
def testDelta(self):
"""Test delta generation by making sure rdiff can process output
There appears to be some undeterminism so we can't just
byte-compare the deltas produced by rdiff and DeltaFile.
"""
MakeRandomFile(self.basis.path)
assert not os.system("rdiff signature %s %s" %
(self.basis.path, self.sig.path))
for i in range(5):
MakeRandomFile(self.new.path)
df = librsync.DeltaFile(self.sig.open("rb"), self.new.open("rb"))
librsync_delta = df.read()
df.close()
fp = self.delta.open("wb")
fp.write(librsync_delta)
fp.close()
assert not os.system("rdiff patch %s %s %s" %
(self.basis.path, self.delta.path,
self.new2.path))
new_fp = self.new.open("rb")
new = new_fp.read()
new_fp.close()
new2_fp = self.new2.open("rb")
new2 = new2_fp.read()
new2_fp.close()
assert new == new2, (len(new), len(new2))
def testPatch(self):
"""Test patching against Rdiff"""
MakeRandomFile(self.basis.path)
assert not os.system("rdiff signature %s %s" %
(self.basis.path, self.sig.path))
for i in range(5):
MakeRandomFile(self.new.path)
assert not os.system("rdiff delta %s %s %s" %
(self.sig.path, self.new.path, self.delta.path))
fp = self.new.open("rb")
real_new = fp.read()
fp.close()
pf = librsync.PatchedFile(self.basis.open("rb"),
self.delta.open("rb"))
librsync_new = pf.read()
pf.close()
assert real_new == librsync_new, \
(len(real_new), len(librsync_new))
if __name__ == "__main__": unittest.main()
|
