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
|
import unittest
execfile("commontest.py")
rbexec("highlevel.py")
#Log.setverbosity(8)
class index:
"""This is just used below to test the iter tree reducer"""
def __init__(self, index):
self.index = index
class RORPIterTest(unittest.TestCase):
def setUp(self):
self.lc = Globals.local_connection
self.inc0rp = RPath(self.lc, "testfiles/empty", ())
self.inc1rp = RPath(self.lc, "testfiles/inc-reg-perms1", ())
self.inc2rp = RPath(self.lc, "testfiles/inc-reg-perms2", ())
self.output = RPath(self.lc, "testfiles/output", ())
def testCollateIterators(self):
"""Test basic collating"""
indicies = map(index, [0,1,2,3])
helper = lambda i: indicies[i]
makeiter1 = lambda: iter(indicies)
makeiter2 = lambda: iter(map(helper, [0,1,3]))
makeiter3 = lambda: iter(map(helper, [1,2]))
outiter = RORPIter.CollateIterators(makeiter1(), makeiter2())
assert Iter.equal(outiter,
iter([(indicies[0], indicies[0]),
(indicies[1], indicies[1]),
(indicies[2], None),
(indicies[3], indicies[3])]))
assert Iter.equal(RORPIter.CollateIterators(makeiter1(),
makeiter2(),
makeiter3()),
iter([(indicies[0], indicies[0], None),
(indicies[1], indicies[1], indicies[1]),
(indicies[2], None, indicies[2]),
(indicies[3], indicies[3], None)]))
assert Iter.equal(RORPIter.CollateIterators(makeiter1(), iter([])),
iter(map(lambda i: (i, None),
indicies)))
assert Iter.equal(iter(map(lambda i: (i, None), indicies)),
RORPIter.CollateIterators(makeiter1(), iter([])))
def testCombinedPatching(self):
"""Combined signature, patch, and diff operations"""
if self.output.lstat(): self.output.delete()
def turninto(final_rp):
sigfile = RORPIter.ToFile(RORPIter.GetSignatureIter(self.output))
diff_file = RORPIter.ToFile(
RORPIter.GetDiffIter(RORPIter.FromFile(sigfile),
RORPIter.IterateRPaths(final_rp)))
RORPIter.PatchIter(self.output, RORPIter.FromFile(diff_file))
turninto(self.inc1rp)
RPath.copy_attribs(self.inc1rp, self.output) # Update time
assert self.compare_no_times(self.inc1rp, self.output)
turninto(self.inc2rp)
RPath.copy_attribs(self.inc2rp, self.output)
assert self.compare_no_times(self.inc2rp, self.output)
def compare_no_times(self, src_rp, dest_rp):
"""Compare but disregard directories attributes"""
dsiter1, dsiter2 = map(DestructiveStepping.Iterate_with_Finalizer,
[src_rp, dest_rp], [1, None])
def equal(src_rorp, dest_rorp):
return ((src_rorp.isdir() and dest_rorp.isdir()) or
src_rorp == dest_rorp)
result = Iter.equal(dsiter1, dsiter2, 1, equal)
for i in dsiter1: pass # make sure all files processed anyway
for i in dsiter2: pass
return result
class IndexedTupleTest(unittest.TestCase):
def testTuple(self):
"""Test indexed tuple"""
i = IndexedTuple((1,2,3), ("a", "b"))
i2 = IndexedTuple((), ("hello", "there", "how are you"))
assert i[0] == "a"
assert i[1] == "b"
assert i2[1] == "there"
assert len(i) == 2 and len(i2) == 3
assert i2 < i, i2 < i
def testTupleAssignment(self):
a, b, c = IndexedTuple((), (1, 2, 3))
assert a == 1
assert b == 2
assert c == 3
if __name__ == "__main__": unittest.main()
|