from wvtest import *
from bup import index, metadata
+from bup.compat import fsencode
from bup.helpers import mkdirp, resolve_parent
from buptest import no_lingering_errors, test_tempdir
import bup.xstat as xstat
-lib_t_dir = os.path.dirname(__file__)
+lib_t_dir = os.path.dirname(fsencode(__file__))
@wvtest
def index_basic():
with no_lingering_errors():
- cd = os.path.realpath('../../../t')
+ cd = os.path.realpath(b'../../../t')
WVPASS(cd)
- sd = os.path.realpath(cd + '/sampledata')
- WVPASSEQ(resolve_parent(cd + '/sampledata'), sd)
- WVPASSEQ(os.path.realpath(cd + '/sampledata/x'), sd + '/x')
- WVPASSEQ(os.path.realpath(cd + '/sampledata/var/abs-symlink'),
- sd + '/var/abs-symlink-target')
- WVPASSEQ(resolve_parent(cd + '/sampledata/var/abs-symlink'),
- sd + '/var/abs-symlink')
+ sd = os.path.realpath(cd + b'/sampledata')
+ WVPASSEQ(resolve_parent(cd + b'/sampledata'), sd)
+ WVPASSEQ(os.path.realpath(cd + b'/sampledata/x'), sd + b'/x')
+ WVPASSEQ(os.path.realpath(cd + b'/sampledata/var/abs-symlink'),
+ sd + b'/var/abs-symlink-target')
+ WVPASSEQ(resolve_parent(cd + b'/sampledata/var/abs-symlink'),
+ sd + b'/var/abs-symlink')
@wvtest
def index_writer():
with no_lingering_errors():
- with test_tempdir('bup-tindex-') as tmpdir:
+ with test_tempdir(b'bup-tindex-') as tmpdir:
orig_cwd = os.getcwd()
try:
os.chdir(tmpdir)
- ds = xstat.stat('.')
- fs = xstat.stat(lib_t_dir + '/tindex.py')
- ms = index.MetaStoreWriter('index.meta.tmp');
+ ds = xstat.stat(b'.')
+ fs = xstat.stat(lib_t_dir + b'/tindex.py')
+ ms = index.MetaStoreWriter(b'index.meta.tmp');
tmax = (time.time() - 1) * 10**9
- w = index.Writer('index.tmp', ms, tmax)
- w.add('/var/tmp/sporky', fs, 0)
- w.add('/etc/passwd', fs, 0)
- w.add('/etc/', ds, 0)
- w.add('/', ds, 0)
+ w = index.Writer(b'index.tmp', ms, tmax)
+ w.add(b'/var/tmp/sporky', fs, 0)
+ w.add(b'/etc/passwd', fs, 0)
+ w.add(b'/etc/', ds, 0)
+ w.add(b'/', ds, 0)
ms.close()
w.close()
finally:
@wvtest
def index_negative_timestamps():
with no_lingering_errors():
- with test_tempdir('bup-tindex-') as tmpdir:
+ with test_tempdir(b'bup-tindex-') as tmpdir:
# Makes 'foo' exist
- foopath = tmpdir + '/foo'
+ foopath = tmpdir + b'/foo'
f = open(foopath, 'wb')
f.close()
@wvtest
def index_dirty():
with no_lingering_errors():
- with test_tempdir('bup-tindex-') as tmpdir:
+ with test_tempdir(b'bup-tindex-') as tmpdir:
orig_cwd = os.getcwd()
try:
os.chdir(tmpdir)
default_meta = metadata.Metadata()
- ms1 = index.MetaStoreWriter('index.meta.tmp')
- ms2 = index.MetaStoreWriter('index2.meta.tmp')
- ms3 = index.MetaStoreWriter('index3.meta.tmp')
+ ms1 = index.MetaStoreWriter(b'index.meta.tmp')
+ ms2 = index.MetaStoreWriter(b'index2.meta.tmp')
+ ms3 = index.MetaStoreWriter(b'index3.meta.tmp')
meta_ofs1 = ms1.store(default_meta)
meta_ofs2 = ms2.store(default_meta)
meta_ofs3 = ms3.store(default_meta)
ds = xstat.stat(lib_t_dir)
- fs = xstat.stat(lib_t_dir + '/tindex.py')
+ fs = xstat.stat(lib_t_dir + b'/tindex.py')
tmax = (time.time() - 1) * 10**9
- w1 = index.Writer('index.tmp', ms1, tmax)
- w1.add('/a/b/x', fs, meta_ofs1)
- w1.add('/a/b/c', fs, meta_ofs1)
- w1.add('/a/b/', ds, meta_ofs1)
- w1.add('/a/', ds, meta_ofs1)
+ w1 = index.Writer(b'index.tmp', ms1, tmax)
+ w1.add(b'/a/b/x', fs, meta_ofs1)
+ w1.add(b'/a/b/c', fs, meta_ofs1)
+ w1.add(b'/a/b/', ds, meta_ofs1)
+ w1.add(b'/a/', ds, meta_ofs1)
#w1.close()
WVPASS()
- w2 = index.Writer('index2.tmp', ms2, tmax)
- w2.add('/a/b/n/2', fs, meta_ofs2)
+ w2 = index.Writer(b'index2.tmp', ms2, tmax)
+ w2.add(b'/a/b/n/2', fs, meta_ofs2)
#w2.close()
WVPASS()
- w3 = index.Writer('index3.tmp', ms3, tmax)
- w3.add('/a/c/n/3', fs, meta_ofs3)
+ w3 = index.Writer(b'index3.tmp', ms3, tmax)
+ w3.add(b'/a/c/n/3', fs, meta_ofs3)
#w3.close()
WVPASS()
r1all = [e.name for e in r1]
WVPASSEQ(r1all,
- ['/a/b/x', '/a/b/c', '/a/b/', '/a/', '/'])
+ [b'/a/b/x', b'/a/b/c', b'/a/b/', b'/a/', b'/'])
r2all = [e.name for e in r2]
WVPASSEQ(r2all,
- ['/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
+ [b'/a/b/n/2', b'/a/b/n/', b'/a/b/', b'/a/', b'/'])
r3all = [e.name for e in r3]
WVPASSEQ(r3all,
- ['/a/c/n/3', '/a/c/n/', '/a/c/', '/a/', '/'])
+ [b'/a/c/n/3', b'/a/c/n/', b'/a/c/', b'/a/', b'/'])
all = [e.name for e in index.merge(r2, r1, r3)]
WVPASSEQ(all,
- ['/a/c/n/3', '/a/c/n/', '/a/c/',
- '/a/b/x', '/a/b/n/2', '/a/b/n/', '/a/b/c',
- '/a/b/', '/a/', '/'])
+ [b'/a/c/n/3', b'/a/c/n/', b'/a/c/',
+ b'/a/b/x', b'/a/b/n/2', b'/a/b/n/', b'/a/b/c',
+ b'/a/b/', b'/a/', b'/'])
fake_validate(r1)
dump(r1)
WVPASSEQ([e.name for e in r1 if e.is_valid()], r1all)
WVPASSEQ([e.name for e in r1 if not e.is_valid()], [])
WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
- ['/a/c/n/3', '/a/c/n/', '/a/c/',
- '/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
+ [b'/a/c/n/3', b'/a/c/n/', b'/a/c/',
+ b'/a/b/n/2', b'/a/b/n/', b'/a/b/', b'/a/', b'/'])
- expect_invalid = ['/'] + r2all + r3all
+ expect_invalid = [b'/'] + r2all + r3all
expect_real = (set(r1all) - set(r2all) - set(r3all)) \
- | set(['/a/b/n/2', '/a/c/n/3'])
+ | set([b'/a/b/n/2', b'/a/c/n/3'])
dump(index.merge(r2, r1, r3))
for e in index.merge(r2, r1, r3):
print(e.name, hex(e.flags), e.ctime)
dump(index.merge(r2, r1, r3))
WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()], [])
- e = eget(index.merge(r2, r1, r3), '/a/b/c')
+ e = eget(index.merge(r2, r1, r3), b'/a/b/c')
e.invalidate()
e.repack()
dump(index.merge(r2, r1, r3))
WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
- ['/a/b/c', '/a/b/', '/a/', '/'])
+ [b'/a/b/c', b'/a/b/', b'/a/', b'/'])
w1.close()
w2.close()
w3.close()
projects / bup.git / blobdiff