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
|
import unittest, os, cStringIO, time
from rdiff_backup.metadata import *
from rdiff_backup import rpath, connection, Globals, selection
tempdir = rpath.RPath(Globals.local_connection, "testfiles/output")
class MetadataTest(unittest.TestCase):
def make_temp(self):
"""Make temp directory testfiles/output"""
global tempdir
if tempdir.lstat(): tempdir.delete()
tempdir.mkdir()
def testQuote(self):
"""Test quoting and unquoting"""
filenames = ["foo", ".", "hello\nthere", "\\", "\\\\\\",
"h\no\t\x87\n", " "]
for filename in filenames:
quoted = quote_path(filename)
assert not "\n" in quoted
result = unquote_path(quoted)
assert result == filename, (quoted, result, filename)
def get_rpaths(self):
"""Return list of rorps"""
vft = rpath.RPath(Globals.local_connection,
"testfiles/various_file_types")
rpaths = map(lambda x: vft.append(x), vft.listdir())
extra_rpaths = map(lambda x: rpath.RPath(Globals.local_connection, x),
['/bin/ls', '/dev/ttyS0', '/dev/hda', 'aoeuaou'])
return [vft] + rpaths + extra_rpaths
def testRORP2Record(self):
"""Test turning RORPs into records and back again"""
for rp in self.get_rpaths():
record = RORP2Record(rp)
#print record
new_rorp = Record2RORP(record)
assert new_rorp == rp, (new_rorp, rp, record)
def testIterator(self):
"""Test writing RORPs to file and iterating them back"""
def write_rorp_iter_to_file(rorp_iter, file):
for rorp in rorp_iter: file.write(RORP2Record(rorp))
l = self.get_rpaths()
fp = cStringIO.StringIO()
write_rorp_iter_to_file(iter(l), fp)
fp.seek(0)
cstring = fp.read()
fp.seek(0)
outlist = list(RorpExtractor(fp).iterate())
assert len(l) == len(outlist), (len(l), len(outlist))
for i in range(len(l)):
if not l[i].equal_verbose(outlist[i]):
#print cstring
assert 0, (i, str(l[i]), str(outlist[i]))
fp.close()
def write_metadata_to_temp(self):
"""If necessary, write metadata of bigdir to file metadata.gz"""
global tempdir
temprp = tempdir.append("metadata.gz")
if temprp.lstat(): return temprp
self.make_temp()
rootrp = rpath.RPath(Globals.local_connection, "testfiles/bigdir")
rpath_iter = selection.Select(rootrp).set_iter()
start_time = time.time()
MetadataFile.open_file(temprp)
for rp in rpath_iter: MetadataFile.write_object(rp)
MetadataFile.close_file()
print "Writing metadata took %s seconds" % (time.time() - start_time)
return temprp
def testSpeed(self):
"""Test testIterator on 10000 files"""
temprp = self.write_metadata_to_temp()
MetadataFile._rp = temprp
start_time = time.time(); i = 0
for rorp in MetadataFile.get_objects(): i += 1
print "Reading %s metadata entries took %s seconds." % \
(i, time.time() - start_time)
start_time = time.time()
blocksize = 32 * 1024
tempfp = temprp.open("rb", compress = 1)
while 1:
buf = tempfp.read(blocksize)
if not buf: break
assert not tempfp.close()
print "Simply decompressing metadata file took %s seconds" % \
(time.time() - start_time)
def testIterate_restricted(self):
"""Test getting rorps restricted to certain index
In this case, get assume subdir (subdir3, subdir10) has 50
files in it.
"""
temprp = self.write_metadata_to_temp()
MetadataFile._rp = temprp
start_time = time.time(); i = 0
for rorp in MetadataFile.get_objects(("subdir3", "subdir10")): i += 1
print "Reading %s metadata entries took %s seconds." % \
(i, time.time() - start_time)
assert i == 51
def test_write(self):
"""Test writing to metadata file, then reading back contents"""
global tempdir
temprp = tempdir.append("write_test.gz")
if temprp.lstat(): temprp.delete()
self.make_temp()
rootrp = rpath.RPath(Globals.local_connection,
"testfiles/various_file_types")
dirlisting = rootrp.listdir()
dirlisting.sort()
rps = map(rootrp.append, dirlisting)
assert not temprp.lstat()
MetadataFile.open_file(temprp)
for rp in rps: MetadataFile.write_object(rp)
MetadataFile.close_file()
assert temprp.lstat()
reread_rps = list(MetadataFile.get_objects())
assert len(reread_rps) == len(rps), (len(reread_rps), len(rps))
for i in range(len(reread_rps)):
assert reread_rps[i] == rps[i], i
if __name__ == "__main__": unittest.main()
|
