6 from bup import index, metadata
7 from bup.helpers import mkdirp, resolve_parent
8 from buptest import no_lingering_errors, test_tempdir
9 import bup.xstat as xstat
12 lib_t_dir = os.path.dirname(__file__)
17 with no_lingering_errors():
18 cd = os.path.realpath('../../../t')
20 sd = os.path.realpath(cd + '/sampledata')
21 WVPASSEQ(resolve_parent(cd + '/sampledata'), sd)
22 WVPASSEQ(os.path.realpath(cd + '/sampledata/x'), sd + '/x')
23 WVPASSEQ(os.path.realpath(cd + '/sampledata/var/abs-symlink'),
24 sd + '/var/abs-symlink-target')
25 WVPASSEQ(resolve_parent(cd + '/sampledata/var/abs-symlink'),
26 sd + '/var/abs-symlink')
31 with no_lingering_errors():
32 with test_tempdir('bup-tindex-') as tmpdir:
33 orig_cwd = os.getcwd()
37 fs = xstat.stat(lib_t_dir + '/tindex.py')
38 ms = index.MetaStoreWriter('index.meta.tmp');
39 tmax = (time.time() - 1) * 10**9
40 w = index.Writer('index.tmp', ms, tmax)
41 w.add('/var/tmp/sporky', fs, 0)
42 w.add('/etc/passwd', fs, 0)
53 print '%s%s %s' % (e.is_valid() and ' ' or 'M',
54 e.is_fake() and 'F' or ' ',
57 def fake_validate(*l):
60 e.validate(0100644, index.FAKE_SHA)
69 def index_negative_timestamps():
70 with no_lingering_errors():
71 with test_tempdir('bup-tindex-') as tmpdir:
73 foopath = tmpdir + '/foo'
74 f = file(foopath, 'wb')
78 os.utime(foopath, (-86400, -86400))
80 tmax = (time.time() - 1) * ns_per_sec
81 e = index.BlankNewEntry(foopath, 0, tmax)
82 e.update_from_stat(xstat.stat(foopath), 0)
86 os.utime(foopath, (-0x80000000, -0x80000000))
87 e = index.BlankNewEntry(foopath, 0, tmax)
88 e.update_from_stat(xstat.stat(foopath), 0)
94 with no_lingering_errors():
95 with test_tempdir('bup-tindex-') as tmpdir:
96 orig_cwd = os.getcwd()
99 default_meta = metadata.Metadata()
100 ms1 = index.MetaStoreWriter('index.meta.tmp')
101 ms2 = index.MetaStoreWriter('index2.meta.tmp')
102 ms3 = index.MetaStoreWriter('index3.meta.tmp')
103 meta_ofs1 = ms1.store(default_meta)
104 meta_ofs2 = ms2.store(default_meta)
105 meta_ofs3 = ms3.store(default_meta)
107 ds = xstat.stat(lib_t_dir)
108 fs = xstat.stat(lib_t_dir + '/tindex.py')
109 tmax = (time.time() - 1) * 10**9
111 w1 = index.Writer('index.tmp', ms1, tmax)
112 w1.add('/a/b/x', fs, meta_ofs1)
113 w1.add('/a/b/c', fs, meta_ofs1)
114 w1.add('/a/b/', ds, meta_ofs1)
115 w1.add('/a/', ds, meta_ofs1)
119 w2 = index.Writer('index2.tmp', ms2, tmax)
120 w2.add('/a/b/n/2', fs, meta_ofs2)
124 w3 = index.Writer('index3.tmp', ms3, tmax)
125 w3.add('/a/c/n/3', fs, meta_ofs3)
134 r1all = [e.name for e in r1]
136 ['/a/b/x', '/a/b/c', '/a/b/', '/a/', '/'])
137 r2all = [e.name for e in r2]
139 ['/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
140 r3all = [e.name for e in r3]
142 ['/a/c/n/3', '/a/c/n/', '/a/c/', '/a/', '/'])
143 all = [e.name for e in index.merge(r2, r1, r3)]
145 ['/a/c/n/3', '/a/c/n/', '/a/c/',
146 '/a/b/x', '/a/b/n/2', '/a/b/n/', '/a/b/c',
147 '/a/b/', '/a/', '/'])
151 print [hex(e.flags) for e in r1]
152 WVPASSEQ([e.name for e in r1 if e.is_valid()], r1all)
153 WVPASSEQ([e.name for e in r1 if not e.is_valid()], [])
154 WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
155 ['/a/c/n/3', '/a/c/n/', '/a/c/',
156 '/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
158 expect_invalid = ['/'] + r2all + r3all
159 expect_real = (set(r1all) - set(r2all) - set(r3all)) \
160 | set(['/a/b/n/2', '/a/c/n/3'])
161 dump(index.merge(r2, r1, r3))
162 for e in index.merge(r2, r1, r3):
163 print e.name, hex(e.flags), e.ctime
164 eiv = e.name in expect_invalid
165 er = e.name in expect_real
166 WVPASSEQ(eiv, not e.is_valid())
167 WVPASSEQ(er, e.is_real())
168 fake_validate(r2, r3)
169 dump(index.merge(r2, r1, r3))
170 WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()], [])
172 e = eget(index.merge(r2, r1, r3), '/a/b/c')
175 dump(index.merge(r2, r1, r3))
176 WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
177 ['/a/b/c', '/a/b/', '/a/', '/'])