--- /dev/null
+
+from contextlib import contextmanager
+from os.path import basename, dirname, realpath
+from traceback import extract_stack
+import subprocess, sys, tempfile
+
+from wvtest import WVPASSEQ, wvfailure_count
+
+from bup import helpers
+
+
+@contextmanager
+def no_lingering_errors():
+ def fail_if_errors():
+ if helpers.saved_errors:
+ bt = extract_stack()
+ src_file, src_line, src_func, src_txt = bt[-4]
+ msg = 'saved_errors ' + repr(helpers.saved_errors)
+ print '! %-70s %s' % ('%s:%-4d %s' % (basename(src_file),
+ src_line,
+ msg),
+ 'FAILED')
+ sys.stdout.flush()
+ fail_if_errors()
+ helpers.clear_errors()
+ yield
+ fail_if_errors()
+ helpers.clear_errors()
+
+
+# Assumes (of course) this file is at the top-level of the source tree
+_bup_tmp = realpath(dirname(__file__) + '/t/tmp')
+helpers.mkdirp(_bup_tmp)
+
+
+@contextmanager
+def test_tempdir(prefix):
+ initial_failures = wvfailure_count()
+ tmpdir = tempfile.mkdtemp(dir=_bup_tmp, prefix=prefix)
+ yield tmpdir
+ if wvfailure_count() == initial_failures:
+ subprocess.call(['chmod', '-R', 'u+rwX', tmpdir])
+ subprocess.call(['rm', '-rf', tmpdir])
-import errno, platform, subprocess, tempfile
-
-from bup import bloom
-from bup.helpers import mkdirp
+import errno, platform, tempfile
from wvtest import *
+from bup import bloom
+from bup.helpers import mkdirp
+from buptest import no_lingering_errors, test_tempdir
-bup_tmp = os.path.realpath('../../../t/tmp')
-mkdirp(bup_tmp)
@wvtest
def test_bloom():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tbloom-')
- hashes = [os.urandom(20) for i in range(100)]
- class Idx:
- pass
- ix = Idx()
- ix.name='dummy.idx'
- ix.shatable = ''.join(hashes)
- for k in (4, 5):
- b = bloom.create(tmpdir + '/pybuptest.bloom', expected=100, k=k)
- b.add_idx(ix)
- WVPASSLT(b.pfalse_positive(), .1)
- b.close()
- b = bloom.ShaBloom(tmpdir + '/pybuptest.bloom')
- all_present = True
- for h in hashes:
- all_present &= b.exists(h)
- WVPASS(all_present)
- false_positives = 0
- for h in [os.urandom(20) for i in range(1000)]:
- if b.exists(h):
- false_positives += 1
- WVPASSLT(false_positives, 5)
- os.unlink(tmpdir + '/pybuptest.bloom')
-
- tf = tempfile.TemporaryFile()
- b = bloom.create('bup.bloom', f=tf, expected=100)
- WVPASSEQ(b.rwfile, tf)
- WVPASSEQ(b.k, 5)
-
- # Test large (~1GiB) filter. This may fail on s390 (31-bit
- # architecture), and anywhere else where the address space is
- # sufficiently limited.
- tf = tempfile.TemporaryFile()
- skip_test = False
- try:
- b = bloom.create('bup.bloom', f=tf, expected=2**28, delaywrite=False)
- except EnvironmentError as ex:
- (ptr_width, linkage) = platform.architecture()
- if ptr_width == '32bit' and ex.errno == errno.ENOMEM:
- WVMSG('skipping large bloom filter test (mmap probably failed) '
- + str(ex))
- skip_test = True
- else:
- raise
- if not skip_test:
- WVPASSEQ(b.k, 4)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tbloom-') as tmpdir:
+ hashes = [os.urandom(20) for i in range(100)]
+ class Idx:
+ pass
+ ix = Idx()
+ ix.name='dummy.idx'
+ ix.shatable = ''.join(hashes)
+ for k in (4, 5):
+ b = bloom.create(tmpdir + '/pybuptest.bloom', expected=100, k=k)
+ b.add_idx(ix)
+ WVPASSLT(b.pfalse_positive(), .1)
+ b.close()
+ b = bloom.ShaBloom(tmpdir + '/pybuptest.bloom')
+ all_present = True
+ for h in hashes:
+ all_present &= b.exists(h)
+ WVPASS(all_present)
+ false_positives = 0
+ for h in [os.urandom(20) for i in range(1000)]:
+ if b.exists(h):
+ false_positives += 1
+ WVPASSLT(false_positives, 5)
+ os.unlink(tmpdir + '/pybuptest.bloom')
+
+ tf = tempfile.TemporaryFile(dir=tmpdir)
+ b = bloom.create('bup.bloom', f=tf, expected=100)
+ WVPASSEQ(b.rwfile, tf)
+ WVPASSEQ(b.k, 5)
+
+ # Test large (~1GiB) filter. This may fail on s390 (31-bit
+ # architecture), and anywhere else where the address space is
+ # sufficiently limited.
+ tf = tempfile.TemporaryFile(dir=tmpdir)
+ skip_test = False
+ try:
+ b = bloom.create('bup.bloom', f=tf, expected=2**28,
+ delaywrite=False)
+ except EnvironmentError as ex:
+ (ptr_width, linkage) = platform.architecture()
+ if ptr_width == '32bit' and ex.errno == errno.ENOMEM:
+ WVMSG('skipping large bloom filter test (mmap probably failed) '
+ + str(ex))
+ skip_test = True
+ else:
+ raise
+ if not skip_test:
+ WVPASSEQ(b.k, 4)
-import sys, os, stat, time, random, subprocess, glob, tempfile
+
+import sys, os, stat, time, random, subprocess, glob
+
+from wvtest import *
+
from bup import client, git
from bup.helpers import mkdirp
-from wvtest import *
+from buptest import no_lingering_errors, test_tempdir
-bup_tmp = os.path.realpath('../../../t/tmp')
-mkdirp(bup_tmp)
def randbytes(sz):
s = ''
s += chr(random.randrange(0,256))
return s
+
s1 = randbytes(10000)
s2 = randbytes(10000)
s3 = randbytes(10000)
IDX_PAT = '/*.idx'
+
@wvtest
def test_server_split_with_indexes():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tclient-')
- os.environ['BUP_MAIN_EXE'] = '../../../bup'
- os.environ['BUP_DIR'] = bupdir = tmpdir
- git.init_repo(bupdir)
- lw = git.PackWriter()
- c = client.Client(bupdir, create=True)
- rw = c.new_packwriter()
-
- lw.new_blob(s1)
- lw.close()
-
- rw.new_blob(s2)
- rw.breakpoint()
- rw.new_blob(s1)
- rw.close()
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tclient-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = '../../../bup'
+ os.environ['BUP_DIR'] = bupdir = tmpdir
+ git.init_repo(bupdir)
+ lw = git.PackWriter()
+ c = client.Client(bupdir, create=True)
+ rw = c.new_packwriter()
+
+ lw.new_blob(s1)
+ lw.close()
+
+ rw.new_blob(s2)
+ rw.breakpoint()
+ rw.new_blob(s1)
+ rw.close()
@wvtest
def test_multiple_suggestions():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tclient-')
- os.environ['BUP_MAIN_EXE'] = '../../../bup'
- os.environ['BUP_DIR'] = bupdir = tmpdir
- git.init_repo(bupdir)
-
- lw = git.PackWriter()
- lw.new_blob(s1)
- lw.close()
- lw = git.PackWriter()
- lw.new_blob(s2)
- lw.close()
- WVPASSEQ(len(glob.glob(git.repo('objects/pack'+IDX_PAT))), 2)
-
- c = client.Client(bupdir, create=True)
- WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 0)
- rw = c.new_packwriter()
- s1sha = rw.new_blob(s1)
- WVPASS(rw.exists(s1sha))
- s2sha = rw.new_blob(s2)
- # This is a little hacky, but ensures that we test the code under test
- while (len(glob.glob(c.cachedir+IDX_PAT)) < 2 and
- not c.conn.has_input()):
- pass
- rw.new_blob(s2)
- WVPASS(rw.objcache.exists(s1sha))
- WVPASS(rw.objcache.exists(s2sha))
- rw.new_blob(s3)
- WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 2)
- rw.close()
- WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 3)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tclient-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = '../../../bup'
+ os.environ['BUP_DIR'] = bupdir = tmpdir
+ git.init_repo(bupdir)
+
+ lw = git.PackWriter()
+ lw.new_blob(s1)
+ lw.close()
+ lw = git.PackWriter()
+ lw.new_blob(s2)
+ lw.close()
+ WVPASSEQ(len(glob.glob(git.repo('objects/pack'+IDX_PAT))), 2)
+
+ c = client.Client(bupdir, create=True)
+ WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 0)
+ rw = c.new_packwriter()
+ s1sha = rw.new_blob(s1)
+ WVPASS(rw.exists(s1sha))
+ s2sha = rw.new_blob(s2)
+ # This is a little hacky, but ensures that we test the code under test
+ while (len(glob.glob(c.cachedir+IDX_PAT)) < 2 and
+ not c.conn.has_input()):
+ pass
+ rw.new_blob(s2)
+ WVPASS(rw.objcache.exists(s1sha))
+ WVPASS(rw.objcache.exists(s2sha))
+ rw.new_blob(s3)
+ WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 2)
+ rw.close()
+ WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 3)
@wvtest
def test_dumb_client_server():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tclient-')
- os.environ['BUP_MAIN_EXE'] = '../../../bup'
- os.environ['BUP_DIR'] = bupdir = tmpdir
- git.init_repo(bupdir)
- open(git.repo('bup-dumb-server'), 'w').close()
-
- lw = git.PackWriter()
- lw.new_blob(s1)
- lw.close()
-
- c = client.Client(bupdir, create=True)
- rw = c.new_packwriter()
- WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 1)
- rw.new_blob(s1)
- WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 1)
- rw.new_blob(s2)
- rw.close()
- WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 2)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tclient-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = '../../../bup'
+ os.environ['BUP_DIR'] = bupdir = tmpdir
+ git.init_repo(bupdir)
+ open(git.repo('bup-dumb-server'), 'w').close()
+
+ lw = git.PackWriter()
+ lw.new_blob(s1)
+ lw.close()
+
+ c = client.Client(bupdir, create=True)
+ rw = c.new_packwriter()
+ WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 1)
+ rw.new_blob(s1)
+ WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 1)
+ rw.new_blob(s2)
+ rw.close()
+ WVPASSEQ(len(glob.glob(c.cachedir+IDX_PAT)), 2)
@wvtest
def test_midx_refreshing():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tclient-')
- os.environ['BUP_MAIN_EXE'] = bupmain = '../../../bup'
- os.environ['BUP_DIR'] = bupdir = tmpdir
- git.init_repo(bupdir)
- c = client.Client(bupdir, create=True)
- rw = c.new_packwriter()
- rw.new_blob(s1)
- p1base = rw.breakpoint()
- p1name = os.path.join(c.cachedir, p1base)
- s1sha = rw.new_blob(s1) # should not be written; it's already in p1
- s2sha = rw.new_blob(s2)
- p2base = rw.close()
- p2name = os.path.join(c.cachedir, p2base)
- del rw
-
- pi = git.PackIdxList(bupdir + '/objects/pack')
- WVPASSEQ(len(pi.packs), 2)
- pi.refresh()
- WVPASSEQ(len(pi.packs), 2)
- WVPASSEQ(sorted([os.path.basename(i.name) for i in pi.packs]),
- sorted([p1base, p2base]))
-
- p1 = git.open_idx(p1name)
- WVPASS(p1.exists(s1sha))
- p2 = git.open_idx(p2name)
- WVFAIL(p2.exists(s1sha))
- WVPASS(p2.exists(s2sha))
-
- subprocess.call([bupmain, 'midx', '-f'])
- pi.refresh()
- WVPASSEQ(len(pi.packs), 1)
- pi.refresh(skip_midx=True)
- WVPASSEQ(len(pi.packs), 2)
- pi.refresh(skip_midx=False)
- WVPASSEQ(len(pi.packs), 1)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tclient-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = bupmain = '../../../bup'
+ os.environ['BUP_DIR'] = bupdir = tmpdir
+ git.init_repo(bupdir)
+ c = client.Client(bupdir, create=True)
+ rw = c.new_packwriter()
+ rw.new_blob(s1)
+ p1base = rw.breakpoint()
+ p1name = os.path.join(c.cachedir, p1base)
+ s1sha = rw.new_blob(s1) # should not be written; it's already in p1
+ s2sha = rw.new_blob(s2)
+ p2base = rw.close()
+ p2name = os.path.join(c.cachedir, p2base)
+ del rw
+
+ pi = git.PackIdxList(bupdir + '/objects/pack')
+ WVPASSEQ(len(pi.packs), 2)
+ pi.refresh()
+ WVPASSEQ(len(pi.packs), 2)
+ WVPASSEQ(sorted([os.path.basename(i.name) for i in pi.packs]),
+ sorted([p1base, p2base]))
+
+ p1 = git.open_idx(p1name)
+ WVPASS(p1.exists(s1sha))
+ p2 = git.open_idx(p2name)
+ WVFAIL(p2.exists(s1sha))
+ WVPASS(p2.exists(s2sha))
+
+ subprocess.call([bupmain, 'midx', '-f'])
+ pi.refresh()
+ WVPASSEQ(len(pi.packs), 1)
+ pi.refresh(skip_midx=True)
+ WVPASSEQ(len(pi.packs), 2)
+ pi.refresh(skip_midx=False)
+ WVPASSEQ(len(pi.packs), 1)
@wvtest
def test_remote_parsing():
- tests = (
- (':/bup', ('file', None, None, '/bup')),
- ('file:///bup', ('file', None, None, '/bup')),
- ('192.168.1.1:/bup', ('ssh', '192.168.1.1', None, '/bup')),
- ('ssh://192.168.1.1:2222/bup', ('ssh', '192.168.1.1', '2222', '/bup')),
- ('ssh://[ff:fe::1]:2222/bup', ('ssh', 'ff:fe::1', '2222', '/bup')),
- ('bup://foo.com:1950', ('bup', 'foo.com', '1950', None)),
- ('bup://foo.com:1950/bup', ('bup', 'foo.com', '1950', '/bup')),
- ('bup://[ff:fe::1]/bup', ('bup', 'ff:fe::1', None, '/bup')),
- )
- for remote, values in tests:
- WVPASSEQ(client.parse_remote(remote), values)
- try:
- client.parse_remote('http://asdf.com/bup')
- WVFAIL()
- except client.ClientError:
- WVPASS()
+ with no_lingering_errors():
+ tests = (
+ (':/bup', ('file', None, None, '/bup')),
+ ('file:///bup', ('file', None, None, '/bup')),
+ ('192.168.1.1:/bup', ('ssh', '192.168.1.1', None, '/bup')),
+ ('ssh://192.168.1.1:2222/bup', ('ssh', '192.168.1.1', '2222', '/bup')),
+ ('ssh://[ff:fe::1]:2222/bup', ('ssh', 'ff:fe::1', '2222', '/bup')),
+ ('bup://foo.com:1950', ('bup', 'foo.com', '1950', None)),
+ ('bup://foo.com:1950/bup', ('bup', 'foo.com', '1950', '/bup')),
+ ('bup://[ff:fe::1]/bup', ('bup', 'ff:fe::1', None, '/bup')),)
+ for remote, values in tests:
+ WVPASSEQ(client.parse_remote(remote), values)
+ try:
+ client.parse_remote('http://asdf.com/bup')
+ WVFAIL()
+ except client.ClientError:
+ WVPASS()
+
from subprocess import check_call
-import struct, os, subprocess, tempfile, time
+import struct, os, time
+
+from wvtest import *
from bup import git
from bup.helpers import localtime, log, mkdirp, readpipe
-
-from wvtest import *
+from buptest import no_lingering_errors, test_tempdir
top_dir = os.path.realpath('../../..')
bup_exe = top_dir + '/bup'
-bup_tmp = top_dir + '/t/tmp'
def exc(*cmd):
@wvtest
def testmangle():
- afile = 0100644
- afile2 = 0100770
- alink = 0120000
- adir = 0040000
- adir2 = 0040777
- WVPASSEQ(git.mangle_name("a", adir2, adir), "a")
- WVPASSEQ(git.mangle_name(".bup", adir2, adir), ".bup.bupl")
- WVPASSEQ(git.mangle_name("a.bupa", adir2, adir), "a.bupa.bupl")
- WVPASSEQ(git.mangle_name("b.bup", alink, alink), "b.bup.bupl")
- WVPASSEQ(git.mangle_name("b.bu", alink, alink), "b.bu")
- WVPASSEQ(git.mangle_name("f", afile, afile2), "f")
- WVPASSEQ(git.mangle_name("f.bup", afile, afile2), "f.bup.bupl")
- WVPASSEQ(git.mangle_name("f.bup", afile, adir), "f.bup.bup")
- WVPASSEQ(git.mangle_name("f", afile, adir), "f.bup")
-
- WVPASSEQ(git.demangle_name("f.bup", afile), ("f", git.BUP_CHUNKED))
- WVPASSEQ(git.demangle_name("f.bupl", afile), ("f", git.BUP_NORMAL))
- WVPASSEQ(git.demangle_name("f.bup.bupl", afile), ("f.bup", git.BUP_NORMAL))
-
- WVPASSEQ(git.demangle_name(".bupm", afile), ('', git.BUP_NORMAL))
- WVPASSEQ(git.demangle_name(".bupm", adir), ('', git.BUP_CHUNKED))
-
- # for safety, we ignore .bup? suffixes we don't recognize. Future
- # versions might implement a .bup[a-z] extension as something other
- # than BUP_NORMAL.
- WVPASSEQ(git.demangle_name("f.bupa", afile), ("f.bupa", git.BUP_NORMAL))
+ with no_lingering_errors():
+ afile = 0100644
+ afile2 = 0100770
+ alink = 0120000
+ adir = 0040000
+ adir2 = 0040777
+ WVPASSEQ(git.mangle_name("a", adir2, adir), "a")
+ WVPASSEQ(git.mangle_name(".bup", adir2, adir), ".bup.bupl")
+ WVPASSEQ(git.mangle_name("a.bupa", adir2, adir), "a.bupa.bupl")
+ WVPASSEQ(git.mangle_name("b.bup", alink, alink), "b.bup.bupl")
+ WVPASSEQ(git.mangle_name("b.bu", alink, alink), "b.bu")
+ WVPASSEQ(git.mangle_name("f", afile, afile2), "f")
+ WVPASSEQ(git.mangle_name("f.bup", afile, afile2), "f.bup.bupl")
+ WVPASSEQ(git.mangle_name("f.bup", afile, adir), "f.bup.bup")
+ WVPASSEQ(git.mangle_name("f", afile, adir), "f.bup")
+
+ WVPASSEQ(git.demangle_name("f.bup", afile), ("f", git.BUP_CHUNKED))
+ WVPASSEQ(git.demangle_name("f.bupl", afile), ("f", git.BUP_NORMAL))
+ WVPASSEQ(git.demangle_name("f.bup.bupl", afile), ("f.bup", git.BUP_NORMAL))
+
+ WVPASSEQ(git.demangle_name(".bupm", afile), ('', git.BUP_NORMAL))
+ WVPASSEQ(git.demangle_name(".bupm", adir), ('', git.BUP_CHUNKED))
+
+ # for safety, we ignore .bup? suffixes we don't recognize. Future
+ # versions might implement a .bup[a-z] extension as something other
+ # than BUP_NORMAL.
+ WVPASSEQ(git.demangle_name("f.bupa", afile), ("f.bupa", git.BUP_NORMAL))
@wvtest
def testencode():
- s = 'hello world'
- looseb = ''.join(git._encode_looseobj('blob', s))
- looset = ''.join(git._encode_looseobj('tree', s))
- loosec = ''.join(git._encode_looseobj('commit', s))
- packb = ''.join(git._encode_packobj('blob', s))
- packt = ''.join(git._encode_packobj('tree', s))
- packc = ''.join(git._encode_packobj('commit', s))
- WVPASSEQ(git._decode_looseobj(looseb), ('blob', s))
- WVPASSEQ(git._decode_looseobj(looset), ('tree', s))
- WVPASSEQ(git._decode_looseobj(loosec), ('commit', s))
- WVPASSEQ(git._decode_packobj(packb), ('blob', s))
- WVPASSEQ(git._decode_packobj(packt), ('tree', s))
- WVPASSEQ(git._decode_packobj(packc), ('commit', s))
+ with no_lingering_errors():
+ s = 'hello world'
+ looseb = ''.join(git._encode_looseobj('blob', s))
+ looset = ''.join(git._encode_looseobj('tree', s))
+ loosec = ''.join(git._encode_looseobj('commit', s))
+ packb = ''.join(git._encode_packobj('blob', s))
+ packt = ''.join(git._encode_packobj('tree', s))
+ packc = ''.join(git._encode_packobj('commit', s))
+ WVPASSEQ(git._decode_looseobj(looseb), ('blob', s))
+ WVPASSEQ(git._decode_looseobj(looset), ('tree', s))
+ WVPASSEQ(git._decode_looseobj(loosec), ('commit', s))
+ WVPASSEQ(git._decode_packobj(packb), ('blob', s))
+ WVPASSEQ(git._decode_packobj(packt), ('tree', s))
+ WVPASSEQ(git._decode_packobj(packc), ('commit', s))
@wvtest
def testpacks():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tgit-')
- os.environ['BUP_MAIN_EXE'] = bup_exe
- os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
- git.init_repo(bupdir)
- git.verbose = 1
-
- w = git.PackWriter()
- w.new_blob(os.urandom(100))
- w.new_blob(os.urandom(100))
- w.abort()
-
- w = git.PackWriter()
- hashes = []
- nobj = 1000
- for i in range(nobj):
- hashes.append(w.new_blob(str(i)))
- log('\n')
- nameprefix = w.close()
- print repr(nameprefix)
- WVPASS(os.path.exists(nameprefix + '.pack'))
- WVPASS(os.path.exists(nameprefix + '.idx'))
-
- r = git.open_idx(nameprefix + '.idx')
- print repr(r.fanout)
-
- for i in range(nobj):
- WVPASS(r.find_offset(hashes[i]) > 0)
- WVPASS(r.exists(hashes[99]))
- WVFAIL(r.exists('\0'*20))
-
- pi = iter(r)
- for h in sorted(hashes):
- WVPASSEQ(str(pi.next()).encode('hex'), h.encode('hex'))
-
- WVFAIL(r.find_offset('\0'*20))
-
- r = git.PackIdxList(bupdir + '/objects/pack')
- WVPASS(r.exists(hashes[5]))
- WVPASS(r.exists(hashes[6]))
- WVFAIL(r.exists('\0'*20))
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tgit-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = bup_exe
+ os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
+ git.init_repo(bupdir)
+ git.verbose = 1
+
+ w = git.PackWriter()
+ w.new_blob(os.urandom(100))
+ w.new_blob(os.urandom(100))
+ w.abort()
-@wvtest
-def test_pack_name_lookup():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tgit-')
- os.environ['BUP_MAIN_EXE'] = bup_exe
- os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
- git.init_repo(bupdir)
- git.verbose = 1
- packdir = git.repo('objects/pack')
-
- idxnames = []
- hashes = []
-
- for start in range(0,28,2):
w = git.PackWriter()
- for i in range(start, start+2):
+ hashes = []
+ nobj = 1000
+ for i in range(nobj):
hashes.append(w.new_blob(str(i)))
log('\n')
- idxnames.append(os.path.basename(w.close() + '.idx'))
+ nameprefix = w.close()
+ print repr(nameprefix)
+ WVPASS(os.path.exists(nameprefix + '.pack'))
+ WVPASS(os.path.exists(nameprefix + '.idx'))
- r = git.PackIdxList(packdir)
- WVPASSEQ(len(r.packs), 2)
- for e,idxname in enumerate(idxnames):
- for i in range(e*2, (e+1)*2):
- WVPASSEQ(r.exists(hashes[i], want_source=True), idxname)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ r = git.open_idx(nameprefix + '.idx')
+ print repr(r.fanout)
+
+ for i in range(nobj):
+ WVPASS(r.find_offset(hashes[i]) > 0)
+ WVPASS(r.exists(hashes[99]))
+ WVFAIL(r.exists('\0'*20))
+
+ pi = iter(r)
+ for h in sorted(hashes):
+ WVPASSEQ(str(pi.next()).encode('hex'), h.encode('hex'))
+
+ WVFAIL(r.find_offset('\0'*20))
+
+ r = git.PackIdxList(bupdir + '/objects/pack')
+ WVPASS(r.exists(hashes[5]))
+ WVPASS(r.exists(hashes[6]))
+ WVFAIL(r.exists('\0'*20))
@wvtest
-def test_long_index():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tgit-')
- os.environ['BUP_MAIN_EXE'] = bup_exe
- os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
- git.init_repo(bupdir)
- w = git.PackWriter()
- obj_bin = struct.pack('!IIIII',
- 0x00112233, 0x44556677, 0x88990011, 0x22334455, 0x66778899)
- obj2_bin = struct.pack('!IIIII',
- 0x11223344, 0x55667788, 0x99001122, 0x33445566, 0x77889900)
- obj3_bin = struct.pack('!IIIII',
- 0x22334455, 0x66778899, 0x00112233, 0x44556677, 0x88990011)
- pack_bin = struct.pack('!IIIII',
- 0x99887766, 0x55443322, 0x11009988, 0x77665544, 0x33221100)
- idx = list(list() for i in xrange(256))
- idx[0].append((obj_bin, 1, 0xfffffffff))
- idx[0x11].append((obj2_bin, 2, 0xffffffffff))
- idx[0x22].append((obj3_bin, 3, 0xff))
- (fd,name) = tempfile.mkstemp(suffix='.idx', dir=git.repo('objects'))
- os.close(fd)
- w.count = 3
- r = w._write_pack_idx_v2(name, idx, pack_bin)
- i = git.PackIdxV2(name, open(name, 'rb'))
- WVPASSEQ(i.find_offset(obj_bin), 0xfffffffff)
- WVPASSEQ(i.find_offset(obj2_bin), 0xffffffffff)
- WVPASSEQ(i.find_offset(obj3_bin), 0xff)
- if wvfailure_count() == initial_failures:
- os.remove(name)
- subprocess.call(['rm', '-rf', tmpdir])
+def test_pack_name_lookup():
+ with no_lingering_errors(), test_tempdir('bup-tgit-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = bup_exe
+ os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
+ git.init_repo(bupdir)
+ git.verbose = 1
+ packdir = git.repo('objects/pack')
+
+ idxnames = []
+ hashes = []
+
+ for start in range(0,28,2):
+ w = git.PackWriter()
+ for i in range(start, start+2):
+ hashes.append(w.new_blob(str(i)))
+ log('\n')
+ idxnames.append(os.path.basename(w.close() + '.idx'))
+
+ r = git.PackIdxList(packdir)
+ WVPASSEQ(len(r.packs), 2)
+ for e,idxname in enumerate(idxnames):
+ for i in range(e*2, (e+1)*2):
+ WVPASSEQ(r.exists(hashes[i], want_source=True), idxname)
@wvtest
-def test_check_repo_or_die():
- initial_failures = wvfailure_count()
- orig_cwd = os.getcwd()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tgit-')
- os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
- try:
- os.chdir(tmpdir)
+def test_long_index():
+ with no_lingering_errors(), test_tempdir('bup-tgit-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = bup_exe
+ os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
git.init_repo(bupdir)
- git.check_repo_or_die()
- WVPASS('check_repo_or_die') # if we reach this point the call above passed
+ w = git.PackWriter()
+ obj_bin = struct.pack('!IIIII',
+ 0x00112233, 0x44556677, 0x88990011, 0x22334455, 0x66778899)
+ obj2_bin = struct.pack('!IIIII',
+ 0x11223344, 0x55667788, 0x99001122, 0x33445566, 0x77889900)
+ obj3_bin = struct.pack('!IIIII',
+ 0x22334455, 0x66778899, 0x00112233, 0x44556677, 0x88990011)
+ pack_bin = struct.pack('!IIIII',
+ 0x99887766, 0x55443322, 0x11009988, 0x77665544, 0x33221100)
+ idx = list(list() for i in xrange(256))
+ idx[0].append((obj_bin, 1, 0xfffffffff))
+ idx[0x11].append((obj2_bin, 2, 0xffffffffff))
+ idx[0x22].append((obj3_bin, 3, 0xff))
+ w.count = 3
+ name = tmpdir + '/tmp.idx'
+ r = w._write_pack_idx_v2(name, idx, pack_bin)
+ i = git.PackIdxV2(name, open(name, 'rb'))
+ WVPASSEQ(i.find_offset(obj_bin), 0xfffffffff)
+ WVPASSEQ(i.find_offset(obj2_bin), 0xffffffffff)
+ WVPASSEQ(i.find_offset(obj3_bin), 0xff)
- os.rename(bupdir + '/objects/pack', bupdir + '/objects/pack.tmp')
- open(bupdir + '/objects/pack', 'w').close()
- try:
- git.check_repo_or_die()
- except SystemExit as e:
- WVPASSEQ(e.code, 14)
- else:
- WVFAIL()
- os.unlink(bupdir + '/objects/pack')
- os.rename(bupdir + '/objects/pack.tmp', bupdir + '/objects/pack')
+@wvtest
+def test_check_repo_or_die():
+ with no_lingering_errors(), test_tempdir('bup-tgit-') as tmpdir:
+ os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
+ orig_cwd = os.getcwd()
try:
- git.check_repo_or_die('nonexistantbup.tmp')
- except SystemExit as e:
- WVPASSEQ(e.code, 15)
- else:
- WVFAIL()
- finally:
- os.chdir(orig_cwd)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ os.chdir(tmpdir)
+ git.init_repo(bupdir)
+ git.check_repo_or_die()
+ WVPASS('check_repo_or_die') # if we reach this point the call above passed
+
+ os.rename(bupdir + '/objects/pack', bupdir + '/objects/pack.tmp')
+ open(bupdir + '/objects/pack', 'w').close()
+ try:
+ git.check_repo_or_die()
+ except SystemExit as e:
+ WVPASSEQ(e.code, 14)
+ else:
+ WVFAIL()
+ os.unlink(bupdir + '/objects/pack')
+ os.rename(bupdir + '/objects/pack.tmp', bupdir + '/objects/pack')
+
+ try:
+ git.check_repo_or_die('nonexistantbup.tmp')
+ except SystemExit as e:
+ WVPASSEQ(e.code, 15)
+ else:
+ WVFAIL()
+ finally:
+ os.chdir(orig_cwd)
@wvtest
def test_commit_parsing():
+
def restore_env_var(name, val):
if val is None:
del os.environ[name]
else:
os.environ[name] = val
+
def showval(commit, val):
return readpipe(['git', 'show', '-s',
'--pretty=format:%s' % val, commit]).strip()
- initial_failures = wvfailure_count()
- orig_cwd = os.getcwd()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tgit-')
- workdir = tmpdir + "/work"
- repodir = workdir + '/.git'
- orig_author_name = os.environ.get('GIT_AUTHOR_NAME')
- orig_author_email = os.environ.get('GIT_AUTHOR_EMAIL')
- orig_committer_name = os.environ.get('GIT_COMMITTER_NAME')
- orig_committer_email = os.environ.get('GIT_COMMITTER_EMAIL')
- os.environ['GIT_AUTHOR_NAME'] = 'bup test'
- os.environ['GIT_COMMITTER_NAME'] = os.environ['GIT_AUTHOR_NAME']
- os.environ['GIT_AUTHOR_EMAIL'] = 'bup@a425bc70a02811e49bdf73ee56450e6f'
- os.environ['GIT_COMMITTER_EMAIL'] = os.environ['GIT_AUTHOR_EMAIL']
- try:
- readpipe(['git', 'init', workdir])
- os.environ['GIT_DIR'] = os.environ['BUP_DIR'] = repodir
- git.check_repo_or_die(repodir)
- os.chdir(workdir)
- with open('foo', 'w') as f:
- print >> f, 'bar'
- readpipe(['git', 'add', '.'])
- readpipe(['git', 'commit', '-am', 'Do something',
- '--author', 'Someone <someone@somewhere>',
- '--date', 'Sat Oct 3 19:48:49 2009 -0400'])
- commit = readpipe(['git', 'show-ref', '-s', 'master']).strip()
- parents = showval(commit, '%P')
- tree = showval(commit, '%T')
- cname = showval(commit, '%cn')
- cmail = showval(commit, '%ce')
- cdate = showval(commit, '%ct')
- coffs = showval(commit, '%ci')
- coffs = coffs[-5:]
- coff = (int(coffs[-4:-2]) * 60 * 60) + (int(coffs[-2:]) * 60)
- if coffs[-5] == '-':
- coff = - coff
- commit_items = git.get_commit_items(commit, git.cp())
- WVPASSEQ(commit_items.parents, [])
- WVPASSEQ(commit_items.tree, tree)
- WVPASSEQ(commit_items.author_name, 'Someone')
- WVPASSEQ(commit_items.author_mail, 'someone@somewhere')
- WVPASSEQ(commit_items.author_sec, 1254613729)
- WVPASSEQ(commit_items.author_offset, -(4 * 60 * 60))
- WVPASSEQ(commit_items.committer_name, cname)
- WVPASSEQ(commit_items.committer_mail, cmail)
- WVPASSEQ(commit_items.committer_sec, int(cdate))
- WVPASSEQ(commit_items.committer_offset, coff)
- WVPASSEQ(commit_items.message, 'Do something\n')
- with open('bar', 'w') as f:
- print >> f, 'baz'
- readpipe(['git', 'add', '.'])
- readpipe(['git', 'commit', '-am', 'Do something else'])
- child = readpipe(['git', 'show-ref', '-s', 'master']).strip()
- parents = showval(child, '%P')
- commit_items = git.get_commit_items(child, git.cp())
- WVPASSEQ(commit_items.parents, [commit])
- finally:
- os.chdir(orig_cwd)
- restore_env_var('GIT_AUTHOR_NAME', orig_author_name)
- restore_env_var('GIT_AUTHOR_EMAIL', orig_author_email)
- restore_env_var('GIT_COMMITTER_NAME', orig_committer_name)
- restore_env_var('GIT_COMMITTER_EMAIL', orig_committer_email)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+
+ with no_lingering_errors(), test_tempdir('bup-tgit-') as tmpdir:
+ orig_cwd = os.getcwd()
+ workdir = tmpdir + "/work"
+ repodir = workdir + '/.git'
+ orig_author_name = os.environ.get('GIT_AUTHOR_NAME')
+ orig_author_email = os.environ.get('GIT_AUTHOR_EMAIL')
+ orig_committer_name = os.environ.get('GIT_COMMITTER_NAME')
+ orig_committer_email = os.environ.get('GIT_COMMITTER_EMAIL')
+ os.environ['GIT_AUTHOR_NAME'] = 'bup test'
+ os.environ['GIT_COMMITTER_NAME'] = os.environ['GIT_AUTHOR_NAME']
+ os.environ['GIT_AUTHOR_EMAIL'] = 'bup@a425bc70a02811e49bdf73ee56450e6f'
+ os.environ['GIT_COMMITTER_EMAIL'] = os.environ['GIT_AUTHOR_EMAIL']
+ try:
+ readpipe(['git', 'init', workdir])
+ os.environ['GIT_DIR'] = os.environ['BUP_DIR'] = repodir
+ git.check_repo_or_die(repodir)
+ os.chdir(workdir)
+ with open('foo', 'w') as f:
+ print >> f, 'bar'
+ readpipe(['git', 'add', '.'])
+ readpipe(['git', 'commit', '-am', 'Do something',
+ '--author', 'Someone <someone@somewhere>',
+ '--date', 'Sat Oct 3 19:48:49 2009 -0400'])
+ commit = readpipe(['git', 'show-ref', '-s', 'master']).strip()
+ parents = showval(commit, '%P')
+ tree = showval(commit, '%T')
+ cname = showval(commit, '%cn')
+ cmail = showval(commit, '%ce')
+ cdate = showval(commit, '%ct')
+ coffs = showval(commit, '%ci')
+ coffs = coffs[-5:]
+ coff = (int(coffs[-4:-2]) * 60 * 60) + (int(coffs[-2:]) * 60)
+ if coffs[-5] == '-':
+ coff = - coff
+ commit_items = git.get_commit_items(commit, git.cp())
+ WVPASSEQ(commit_items.parents, [])
+ WVPASSEQ(commit_items.tree, tree)
+ WVPASSEQ(commit_items.author_name, 'Someone')
+ WVPASSEQ(commit_items.author_mail, 'someone@somewhere')
+ WVPASSEQ(commit_items.author_sec, 1254613729)
+ WVPASSEQ(commit_items.author_offset, -(4 * 60 * 60))
+ WVPASSEQ(commit_items.committer_name, cname)
+ WVPASSEQ(commit_items.committer_mail, cmail)
+ WVPASSEQ(commit_items.committer_sec, int(cdate))
+ WVPASSEQ(commit_items.committer_offset, coff)
+ WVPASSEQ(commit_items.message, 'Do something\n')
+ with open('bar', 'w') as f:
+ print >> f, 'baz'
+ readpipe(['git', 'add', '.'])
+ readpipe(['git', 'commit', '-am', 'Do something else'])
+ child = readpipe(['git', 'show-ref', '-s', 'master']).strip()
+ parents = showval(child, '%P')
+ commit_items = git.get_commit_items(child, git.cp())
+ WVPASSEQ(commit_items.parents, [commit])
+ finally:
+ os.chdir(orig_cwd)
+ restore_env_var('GIT_AUTHOR_NAME', orig_author_name)
+ restore_env_var('GIT_AUTHOR_EMAIL', orig_author_email)
+ restore_env_var('GIT_COMMITTER_NAME', orig_committer_name)
+ restore_env_var('GIT_COMMITTER_EMAIL', orig_committer_email)
@wvtest
def test_new_commit():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tgit-')
- os.environ['BUP_MAIN_EXE'] = bup_exe
- os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
- git.init_repo(bupdir)
- git.verbose = 1
-
- w = git.PackWriter()
- tree = os.urandom(20)
- parent = os.urandom(20)
- author_name = 'Author'
- author_mail = 'author@somewhere'
- adate_sec = 1439657836
- cdate_sec = adate_sec + 1
- committer_name = 'Committer'
- committer_mail = 'committer@somewhere'
- adate_tz_sec = cdate_tz_sec = None
- commit = w.new_commit(tree, parent,
- '%s <%s>' % (author_name, author_mail),
- adate_sec, adate_tz_sec,
- '%s <%s>' % (committer_name, committer_mail),
- cdate_sec, cdate_tz_sec,
- 'There is a small mailbox here')
- adate_tz_sec = -60 * 60
- cdate_tz_sec = 120 * 60
- commit_off = w.new_commit(tree, parent,
+ with no_lingering_errors(), test_tempdir('bup-tgit-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = bup_exe
+ os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
+ git.init_repo(bupdir)
+ git.verbose = 1
+
+ w = git.PackWriter()
+ tree = os.urandom(20)
+ parent = os.urandom(20)
+ author_name = 'Author'
+ author_mail = 'author@somewhere'
+ adate_sec = 1439657836
+ cdate_sec = adate_sec + 1
+ committer_name = 'Committer'
+ committer_mail = 'committer@somewhere'
+ adate_tz_sec = cdate_tz_sec = None
+ commit = w.new_commit(tree, parent,
'%s <%s>' % (author_name, author_mail),
adate_sec, adate_tz_sec,
'%s <%s>' % (committer_name, committer_mail),
cdate_sec, cdate_tz_sec,
'There is a small mailbox here')
- w.close()
-
- commit_items = git.get_commit_items(commit.encode('hex'), git.cp())
- local_author_offset = localtime(adate_sec).tm_gmtoff
- local_committer_offset = localtime(cdate_sec).tm_gmtoff
- WVPASSEQ(tree, commit_items.tree.decode('hex'))
- WVPASSEQ(1, len(commit_items.parents))
- WVPASSEQ(parent, commit_items.parents[0].decode('hex'))
- WVPASSEQ(author_name, commit_items.author_name)
- WVPASSEQ(author_mail, commit_items.author_mail)
- WVPASSEQ(adate_sec, commit_items.author_sec)
- WVPASSEQ(local_author_offset, commit_items.author_offset)
- WVPASSEQ(committer_name, commit_items.committer_name)
- WVPASSEQ(committer_mail, commit_items.committer_mail)
- WVPASSEQ(cdate_sec, commit_items.committer_sec)
- WVPASSEQ(local_committer_offset, commit_items.committer_offset)
-
- commit_items = git.get_commit_items(commit_off.encode('hex'), git.cp())
- WVPASSEQ(tree, commit_items.tree.decode('hex'))
- WVPASSEQ(1, len(commit_items.parents))
- WVPASSEQ(parent, commit_items.parents[0].decode('hex'))
- WVPASSEQ(author_name, commit_items.author_name)
- WVPASSEQ(author_mail, commit_items.author_mail)
- WVPASSEQ(adate_sec, commit_items.author_sec)
- WVPASSEQ(adate_tz_sec, commit_items.author_offset)
- WVPASSEQ(committer_name, commit_items.committer_name)
- WVPASSEQ(committer_mail, commit_items.committer_mail)
- WVPASSEQ(cdate_sec, commit_items.committer_sec)
- WVPASSEQ(cdate_tz_sec, commit_items.committer_offset)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ adate_tz_sec = -60 * 60
+ cdate_tz_sec = 120 * 60
+ commit_off = w.new_commit(tree, parent,
+ '%s <%s>' % (author_name, author_mail),
+ adate_sec, adate_tz_sec,
+ '%s <%s>' % (committer_name, committer_mail),
+ cdate_sec, cdate_tz_sec,
+ 'There is a small mailbox here')
+ w.close()
+
+ commit_items = git.get_commit_items(commit.encode('hex'), git.cp())
+ local_author_offset = localtime(adate_sec).tm_gmtoff
+ local_committer_offset = localtime(cdate_sec).tm_gmtoff
+ WVPASSEQ(tree, commit_items.tree.decode('hex'))
+ WVPASSEQ(1, len(commit_items.parents))
+ WVPASSEQ(parent, commit_items.parents[0].decode('hex'))
+ WVPASSEQ(author_name, commit_items.author_name)
+ WVPASSEQ(author_mail, commit_items.author_mail)
+ WVPASSEQ(adate_sec, commit_items.author_sec)
+ WVPASSEQ(local_author_offset, commit_items.author_offset)
+ WVPASSEQ(committer_name, commit_items.committer_name)
+ WVPASSEQ(committer_mail, commit_items.committer_mail)
+ WVPASSEQ(cdate_sec, commit_items.committer_sec)
+ WVPASSEQ(local_committer_offset, commit_items.committer_offset)
+
+ commit_items = git.get_commit_items(commit_off.encode('hex'), git.cp())
+ WVPASSEQ(tree, commit_items.tree.decode('hex'))
+ WVPASSEQ(1, len(commit_items.parents))
+ WVPASSEQ(parent, commit_items.parents[0].decode('hex'))
+ WVPASSEQ(author_name, commit_items.author_name)
+ WVPASSEQ(author_mail, commit_items.author_mail)
+ WVPASSEQ(adate_sec, commit_items.author_sec)
+ WVPASSEQ(adate_tz_sec, commit_items.author_offset)
+ WVPASSEQ(committer_name, commit_items.committer_name)
+ WVPASSEQ(committer_mail, commit_items.committer_mail)
+ WVPASSEQ(cdate_sec, commit_items.committer_sec)
+ WVPASSEQ(cdate_tz_sec, commit_items.committer_offset)
@wvtest
def test_list_refs():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tgit-')
- os.environ['BUP_MAIN_EXE'] = bup_exe
- os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
- src = tmpdir + '/src'
- mkdirp(src)
- with open(src + '/1', 'w+') as f:
- print f, 'something'
- with open(src + '/2', 'w+') as f:
- print f, 'something else'
- git.init_repo(bupdir)
- emptyset = frozenset()
- WVPASSEQ(frozenset(git.list_refs()), emptyset)
- WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)), emptyset)
- WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)), emptyset)
- exc(bup_exe, 'index', src)
- exc(bup_exe, 'save', '-n', 'src', '--strip', src)
- src_hash = exo('git', '--git-dir', bupdir,
- 'rev-parse', 'src').strip().split('\n')
- assert(len(src_hash) == 1)
- src_hash = src_hash[0].decode('hex')
- tree_hash = exo('git', '--git-dir', bupdir,
- 'rev-parse', 'src:').strip().split('\n')[0].decode('hex')
- blob_hash = exo('git', '--git-dir', bupdir,
- 'rev-parse', 'src:1').strip().split('\n')[0].decode('hex')
- WVPASSEQ(frozenset(git.list_refs()),
- frozenset([('refs/heads/src', src_hash)]))
- WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)), emptyset)
- WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)),
- frozenset([('refs/heads/src', src_hash)]))
- exc('git', '--git-dir', bupdir, 'tag', 'commit-tag', 'src')
- WVPASSEQ(frozenset(git.list_refs()),
- frozenset([('refs/heads/src', src_hash),
- ('refs/tags/commit-tag', src_hash)]))
- WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)),
- frozenset([('refs/tags/commit-tag', src_hash)]))
- WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)),
- frozenset([('refs/heads/src', src_hash)]))
- exc('git', '--git-dir', bupdir, 'tag', 'tree-tag', 'src:')
- exc('git', '--git-dir', bupdir, 'tag', 'blob-tag', 'src:1')
- os.unlink(bupdir + '/refs/heads/src')
- expected_tags = frozenset([('refs/tags/commit-tag', src_hash),
- ('refs/tags/tree-tag', tree_hash),
- ('refs/tags/blob-tag', blob_hash)])
- WVPASSEQ(frozenset(git.list_refs()), expected_tags)
- WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)), frozenset([]))
- WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)), expected_tags)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tgit-') as tmpdir:
+ os.environ['BUP_MAIN_EXE'] = bup_exe
+ os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
+ src = tmpdir + '/src'
+ mkdirp(src)
+ with open(src + '/1', 'w+') as f:
+ print f, 'something'
+ with open(src + '/2', 'w+') as f:
+ print f, 'something else'
+ git.init_repo(bupdir)
+ emptyset = frozenset()
+ WVPASSEQ(frozenset(git.list_refs()), emptyset)
+ WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)), emptyset)
+ WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)), emptyset)
+ exc(bup_exe, 'index', src)
+ exc(bup_exe, 'save', '-n', 'src', '--strip', src)
+ src_hash = exo('git', '--git-dir', bupdir,
+ 'rev-parse', 'src').strip().split('\n')
+ assert(len(src_hash) == 1)
+ src_hash = src_hash[0].decode('hex')
+ tree_hash = exo('git', '--git-dir', bupdir,
+ 'rev-parse', 'src:').strip().split('\n')[0].decode('hex')
+ blob_hash = exo('git', '--git-dir', bupdir,
+ 'rev-parse', 'src:1').strip().split('\n')[0].decode('hex')
+ WVPASSEQ(frozenset(git.list_refs()),
+ frozenset([('refs/heads/src', src_hash)]))
+ WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)), emptyset)
+ WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)),
+ frozenset([('refs/heads/src', src_hash)]))
+ exc('git', '--git-dir', bupdir, 'tag', 'commit-tag', 'src')
+ WVPASSEQ(frozenset(git.list_refs()),
+ frozenset([('refs/heads/src', src_hash),
+ ('refs/tags/commit-tag', src_hash)]))
+ WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)),
+ frozenset([('refs/tags/commit-tag', src_hash)]))
+ WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)),
+ frozenset([('refs/heads/src', src_hash)]))
+ exc('git', '--git-dir', bupdir, 'tag', 'tree-tag', 'src:')
+ exc('git', '--git-dir', bupdir, 'tag', 'blob-tag', 'src:1')
+ os.unlink(bupdir + '/refs/heads/src')
+ expected_tags = frozenset([('refs/tags/commit-tag', src_hash),
+ ('refs/tags/tree-tag', tree_hash),
+ ('refs/tags/blob-tag', blob_hash)])
+ WVPASSEQ(frozenset(git.list_refs()), expected_tags)
+ WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)), frozenset([]))
+ WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)), expected_tags)
+
def test__git_date_str():
- WVPASSEQ('0 +0000', git._git_date_str(0, 0))
- WVPASSEQ('0 -0130', git._git_date_str(0, -90 * 60))
- WVPASSEQ('0 +0130', git._git_date_str(0, 90 * 60))
+ with no_lingering_errors():
+ WVPASSEQ('0 +0000', git._git_date_str(0, 0))
+ WVPASSEQ('0 -0130', git._git_date_str(0, -90 * 60))
+ WVPASSEQ('0 +0130', git._git_date_str(0, 90 * 60))
from wvtest import *
from bup import hashsplit, _helpers, helpers
+from buptest import no_lingering_errors
def nr_regions(x, max_count=None):
@wvtest
def test_nonresident_page_regions():
- WVPASSEQ(nr_regions([]), [])
- WVPASSEQ(nr_regions([1]), [])
- WVPASSEQ(nr_regions([0]), [(0, 1)])
- WVPASSEQ(nr_regions([1, 0]), [(1, 1)])
- WVPASSEQ(nr_regions([0, 0]), [(0, 2)])
- WVPASSEQ(nr_regions([1, 0, 1]), [(1, 1)])
- WVPASSEQ(nr_regions([1, 0, 0]), [(1, 2)])
- WVPASSEQ(nr_regions([0, 1, 0]), [(0, 1), (2, 1)])
- WVPASSEQ(nr_regions([0, 0, 1, 1, 1, 0, 0, 0, 1, 0, 0]),
- [(0, 2), (5, 3), (9, 2)])
- WVPASSEQ(nr_regions([2, 42, 3, 101]), [(0, 2)])
- # Test limit
- WVPASSEQ(nr_regions([0, 0, 0], None), [(0, 3)])
- WVPASSEQ(nr_regions([0, 0, 0], 1), [(0, 1), (1, 1), (2, 1)])
- WVPASSEQ(nr_regions([0, 0, 0], 2), [(0, 2), (2, 1)])
- WVPASSEQ(nr_regions([0, 0, 0], 3), [(0, 3)])
- WVPASSEQ(nr_regions([0, 0, 0], 4), [(0, 3)])
- WVPASSEQ(nr_regions([0, 0, 1], None), [(0, 2)])
- WVPASSEQ(nr_regions([0, 0, 1], 1), [(0, 1), (1, 1)])
- WVPASSEQ(nr_regions([0, 0, 1], 2), [(0, 2)])
- WVPASSEQ(nr_regions([0, 0, 1], 3), [(0, 2)])
- WVPASSEQ(nr_regions([1, 0, 0], None), [(1, 2)])
- WVPASSEQ(nr_regions([1, 0, 0], 1), [(1, 1), (2, 1)])
- WVPASSEQ(nr_regions([1, 0, 0], 2), [(1, 2)])
- WVPASSEQ(nr_regions([1, 0, 0], 3), [(1, 2)])
- WVPASSEQ(nr_regions([1, 0, 0, 0, 1], None), [(1, 3)])
- WVPASSEQ(nr_regions([1, 0, 0, 0, 1], 1), [(1, 1), (2, 1), (3, 1)])
- WVPASSEQ(nr_regions([1, 0, 0, 0, 1], 2), [(1, 2), (3, 1)])
- WVPASSEQ(nr_regions([1, 0, 0, 0, 1], 3), [(1, 3)])
- WVPASSEQ(nr_regions([1, 0, 0, 0, 1], 4), [(1, 3)])
+ with no_lingering_errors():
+ WVPASSEQ(nr_regions([]), [])
+ WVPASSEQ(nr_regions([1]), [])
+ WVPASSEQ(nr_regions([0]), [(0, 1)])
+ WVPASSEQ(nr_regions([1, 0]), [(1, 1)])
+ WVPASSEQ(nr_regions([0, 0]), [(0, 2)])
+ WVPASSEQ(nr_regions([1, 0, 1]), [(1, 1)])
+ WVPASSEQ(nr_regions([1, 0, 0]), [(1, 2)])
+ WVPASSEQ(nr_regions([0, 1, 0]), [(0, 1), (2, 1)])
+ WVPASSEQ(nr_regions([0, 0, 1, 1, 1, 0, 0, 0, 1, 0, 0]),
+ [(0, 2), (5, 3), (9, 2)])
+ WVPASSEQ(nr_regions([2, 42, 3, 101]), [(0, 2)])
+ # Test limit
+ WVPASSEQ(nr_regions([0, 0, 0], None), [(0, 3)])
+ WVPASSEQ(nr_regions([0, 0, 0], 1), [(0, 1), (1, 1), (2, 1)])
+ WVPASSEQ(nr_regions([0, 0, 0], 2), [(0, 2), (2, 1)])
+ WVPASSEQ(nr_regions([0, 0, 0], 3), [(0, 3)])
+ WVPASSEQ(nr_regions([0, 0, 0], 4), [(0, 3)])
+ WVPASSEQ(nr_regions([0, 0, 1], None), [(0, 2)])
+ WVPASSEQ(nr_regions([0, 0, 1], 1), [(0, 1), (1, 1)])
+ WVPASSEQ(nr_regions([0, 0, 1], 2), [(0, 2)])
+ WVPASSEQ(nr_regions([0, 0, 1], 3), [(0, 2)])
+ WVPASSEQ(nr_regions([1, 0, 0], None), [(1, 2)])
+ WVPASSEQ(nr_regions([1, 0, 0], 1), [(1, 1), (2, 1)])
+ WVPASSEQ(nr_regions([1, 0, 0], 2), [(1, 2)])
+ WVPASSEQ(nr_regions([1, 0, 0], 3), [(1, 2)])
+ WVPASSEQ(nr_regions([1, 0, 0, 0, 1], None), [(1, 3)])
+ WVPASSEQ(nr_regions([1, 0, 0, 0, 1], 1), [(1, 1), (2, 1), (3, 1)])
+ WVPASSEQ(nr_regions([1, 0, 0, 0, 1], 2), [(1, 2), (3, 1)])
+ WVPASSEQ(nr_regions([1, 0, 0, 0, 1], 3), [(1, 3)])
+ WVPASSEQ(nr_regions([1, 0, 0, 0, 1], 4), [(1, 3)])
@wvtest
def mock_fadvise_pages_done(f, ofs, len):
history.append((f, ofs, len))
- uncache_upto = hashsplit._uncache_ours_upto
- page_size = helpers.sc_page_size
- orig_pages_done = hashsplit._fadvise_pages_done
- try:
- hashsplit._fadvise_pages_done = mock_fadvise_pages_done
- history = []
- uncache_upto(42, 0, (0, 1), iter([]))
- WVPASSEQ([], history)
- uncache_upto(42, page_size, (0, 1), iter([]))
- WVPASSEQ([(42, 0, 1)], history)
- history = []
- uncache_upto(42, page_size, (0, 3), iter([(5, 2)]))
- WVPASSEQ([], history)
- uncache_upto(42, 2 * page_size, (0, 3), iter([(5, 2)]))
- WVPASSEQ([], history)
- uncache_upto(42, 3 * page_size, (0, 3), iter([(5, 2)]))
- WVPASSEQ([(42, 0, 3)], history)
- history = []
- uncache_upto(42, 5 * page_size, (0, 3), iter([(5, 2)]))
- WVPASSEQ([(42, 0, 3)], history)
- history = []
- uncache_upto(42, 6 * page_size, (0, 3), iter([(5, 2)]))
- WVPASSEQ([(42, 0, 3)], history)
- history = []
- uncache_upto(42, 7 * page_size, (0, 3), iter([(5, 2)]))
- WVPASSEQ([(42, 0, 3), (42, 5, 2)], history)
- finally:
- hashsplit._fadvise_pages_done = orig_pages_done
+ with no_lingering_errors():
+ uncache_upto = hashsplit._uncache_ours_upto
+ page_size = helpers.sc_page_size
+ orig_pages_done = hashsplit._fadvise_pages_done
+ try:
+ hashsplit._fadvise_pages_done = mock_fadvise_pages_done
+ history = []
+ uncache_upto(42, 0, (0, 1), iter([]))
+ WVPASSEQ([], history)
+ uncache_upto(42, page_size, (0, 1), iter([]))
+ WVPASSEQ([(42, 0, 1)], history)
+ history = []
+ uncache_upto(42, page_size, (0, 3), iter([(5, 2)]))
+ WVPASSEQ([], history)
+ uncache_upto(42, 2 * page_size, (0, 3), iter([(5, 2)]))
+ WVPASSEQ([], history)
+ uncache_upto(42, 3 * page_size, (0, 3), iter([(5, 2)]))
+ WVPASSEQ([(42, 0, 3)], history)
+ history = []
+ uncache_upto(42, 5 * page_size, (0, 3), iter([(5, 2)]))
+ WVPASSEQ([(42, 0, 3)], history)
+ history = []
+ uncache_upto(42, 6 * page_size, (0, 3), iter([(5, 2)]))
+ WVPASSEQ([(42, 0, 3)], history)
+ history = []
+ uncache_upto(42, 7 * page_size, (0, 3), iter([(5, 2)]))
+ WVPASSEQ([(42, 0, 3), (42, 5, 2)], history)
+ finally:
+ hashsplit._fadvise_pages_done = orig_pages_done
@wvtest
def test_rolling_sums():
- WVPASS(_helpers.selftest())
+ with no_lingering_errors():
+ WVPASS(_helpers.selftest())
@wvtest
def test_fanout_behaviour():
return ofs, ord(c)
return 0, 0
- old_splitbuf = _helpers.splitbuf
- _helpers.splitbuf = splitbuf
- old_BLOB_MAX = hashsplit.BLOB_MAX
- hashsplit.BLOB_MAX = 4
- old_BLOB_READ_SIZE = hashsplit.BLOB_READ_SIZE
- hashsplit.BLOB_READ_SIZE = 10
- old_fanout = hashsplit.fanout
- hashsplit.fanout = 2
-
- levels = lambda f: [(len(b), l) for b, l in
- hashsplit.hashsplit_iter([f], True, None)]
- # Return a string of n null bytes
- z = lambda n: '\x00' * n
- # Return a byte which will be split with a level of n
- sb = lambda n: chr(basebits + n)
-
- split_never = BytesIO(z(16))
- split_first = BytesIO(z(1) + sb(3) + z(14))
- split_end = BytesIO(z(13) + sb(1) + z(2))
- split_many = BytesIO(sb(1) + z(3) + sb(2) + z(4) +
- sb(0) + z(4) + sb(5) + z(1))
- WVPASSEQ(levels(split_never), [(4, 0), (4, 0), (4, 0), (4, 0)])
- WVPASSEQ(levels(split_first), [(2, 3), (4, 0), (4, 0), (4, 0), (2, 0)])
- WVPASSEQ(levels(split_end), [(4, 0), (4, 0), (4, 0), (2, 1), (2, 0)])
- WVPASSEQ(levels(split_many),
- [(1, 1), (4, 2), (4, 0), (1, 0), (4, 0), (1, 5), (1, 0)])
-
- _helpers.splitbuf = old_splitbuf
- hashsplit.BLOB_MAX = old_BLOB_MAX
- hashsplit.BLOB_READ_SIZE = old_BLOB_READ_SIZE
- hashsplit.fanout = old_fanout
+ with no_lingering_errors():
+ old_splitbuf = _helpers.splitbuf
+ _helpers.splitbuf = splitbuf
+ old_BLOB_MAX = hashsplit.BLOB_MAX
+ hashsplit.BLOB_MAX = 4
+ old_BLOB_READ_SIZE = hashsplit.BLOB_READ_SIZE
+ hashsplit.BLOB_READ_SIZE = 10
+ old_fanout = hashsplit.fanout
+ hashsplit.fanout = 2
+
+ levels = lambda f: [(len(b), l) for b, l in
+ hashsplit.hashsplit_iter([f], True, None)]
+ # Return a string of n null bytes
+ z = lambda n: '\x00' * n
+ # Return a byte which will be split with a level of n
+ sb = lambda n: chr(basebits + n)
+
+ split_never = BytesIO(z(16))
+ split_first = BytesIO(z(1) + sb(3) + z(14))
+ split_end = BytesIO(z(13) + sb(1) + z(2))
+ split_many = BytesIO(sb(1) + z(3) + sb(2) + z(4) +
+ sb(0) + z(4) + sb(5) + z(1))
+ WVPASSEQ(levels(split_never), [(4, 0), (4, 0), (4, 0), (4, 0)])
+ WVPASSEQ(levels(split_first), [(2, 3), (4, 0), (4, 0), (4, 0), (2, 0)])
+ WVPASSEQ(levels(split_end), [(4, 0), (4, 0), (4, 0), (2, 1), (2, 0)])
+ WVPASSEQ(levels(split_many),
+ [(1, 1), (4, 2), (4, 0), (1, 0), (4, 0), (1, 5), (1, 0)])
+
+ _helpers.splitbuf = old_splitbuf
+ hashsplit.BLOB_MAX = old_BLOB_MAX
+ hashsplit.BLOB_READ_SIZE = old_BLOB_READ_SIZE
+ hashsplit.fanout = old_fanout
-import helpers
-import math
-import os
-import os.path
-import tempfile
-import stat
-import subprocess
+
+import helpers, math, os, os.path, stat, subprocess
+
+from wvtest import *
from bup.helpers import (atomically_replaced_file, batchpipe, detect_fakeroot,
grafted_path_components, mkdirp, parse_num,
path_components, readpipe, stripped_path_components,
utc_offset_str)
+from buptest import no_lingering_errors, test_tempdir
import bup._helpers as _helpers
-from wvtest import *
-
bup_tmp = os.path.realpath('../../../t/tmp')
mkdirp(bup_tmp)
@wvtest
def test_next():
- # Test whatever you end up with for next() after import '*'.
- WVPASSEQ(next(iter([]), None), None)
- x = iter([1])
- WVPASSEQ(next(x, None), 1)
- WVPASSEQ(next(x, None), None)
- x = iter([1])
- WVPASSEQ(next(x, 'x'), 1)
- WVPASSEQ(next(x, 'x'), 'x')
- WVEXCEPT(StopIteration, next, iter([]))
- x = iter([1])
- WVPASSEQ(next(x), 1)
- WVEXCEPT(StopIteration, next, x)
+ with no_lingering_errors():
+ # Test whatever you end up with for next() after import '*'.
+ WVPASSEQ(next(iter([]), None), None)
+ x = iter([1])
+ WVPASSEQ(next(x, None), 1)
+ WVPASSEQ(next(x, None), None)
+ x = iter([1])
+ WVPASSEQ(next(x, 'x'), 1)
+ WVPASSEQ(next(x, 'x'), 'x')
+ WVEXCEPT(StopIteration, next, iter([]))
+ x = iter([1])
+ WVPASSEQ(next(x), 1)
+ WVEXCEPT(StopIteration, next, x)
@wvtest
def test_fallback_next():
- global next
- orig = next
- next = helpers._fallback_next
- try:
- test_next()
- finally:
- next = orig
+ with no_lingering_errors():
+ global next
+ orig = next
+ next = helpers._fallback_next
+ try:
+ test_next()
+ finally:
+ next = orig
@wvtest
def test_parse_num():
- pn = parse_num
- WVPASSEQ(pn('1'), 1)
- WVPASSEQ(pn('0'), 0)
- WVPASSEQ(pn('1.5k'), 1536)
- WVPASSEQ(pn('2 gb'), 2*1024*1024*1024)
- WVPASSEQ(pn('1e+9 k'), 1000000000 * 1024)
- WVPASSEQ(pn('-3e-3mb'), int(-0.003 * 1024 * 1024))
+ with no_lingering_errors():
+ pn = parse_num
+ WVPASSEQ(pn('1'), 1)
+ WVPASSEQ(pn('0'), 0)
+ WVPASSEQ(pn('1.5k'), 1536)
+ WVPASSEQ(pn('2 gb'), 2*1024*1024*1024)
+ WVPASSEQ(pn('1e+9 k'), 1000000000 * 1024)
+ WVPASSEQ(pn('-3e-3mb'), int(-0.003 * 1024 * 1024))
@wvtest
def test_detect_fakeroot():
- if os.getenv('FAKEROOTKEY'):
- WVPASS(detect_fakeroot())
- else:
- WVPASS(not detect_fakeroot())
+ with no_lingering_errors():
+ if os.getenv('FAKEROOTKEY'):
+ WVPASS(detect_fakeroot())
+ else:
+ WVPASS(not detect_fakeroot())
@wvtest
def test_path_components():
- WVPASSEQ(path_components('/'), [('', '/')])
- WVPASSEQ(path_components('/foo'), [('', '/'), ('foo', '/foo')])
- WVPASSEQ(path_components('/foo/'), [('', '/'), ('foo', '/foo')])
- WVPASSEQ(path_components('/foo/bar'),
- [('', '/'), ('foo', '/foo'), ('bar', '/foo/bar')])
- WVEXCEPT(Exception, path_components, 'foo')
+ with no_lingering_errors():
+ WVPASSEQ(path_components('/'), [('', '/')])
+ WVPASSEQ(path_components('/foo'), [('', '/'), ('foo', '/foo')])
+ WVPASSEQ(path_components('/foo/'), [('', '/'), ('foo', '/foo')])
+ WVPASSEQ(path_components('/foo/bar'),
+ [('', '/'), ('foo', '/foo'), ('bar', '/foo/bar')])
+ WVEXCEPT(Exception, path_components, 'foo')
@wvtest
def test_stripped_path_components():
- WVPASSEQ(stripped_path_components('/', []), [('', '/')])
- WVPASSEQ(stripped_path_components('/', ['']), [('', '/')])
- WVPASSEQ(stripped_path_components('/', ['/']), [('', '/')])
- WVPASSEQ(stripped_path_components('/foo', ['/']),
- [('', '/'), ('foo', '/foo')])
- WVPASSEQ(stripped_path_components('/', ['/foo']), [('', '/')])
- WVPASSEQ(stripped_path_components('/foo', ['/bar']),
- [('', '/'), ('foo', '/foo')])
- WVPASSEQ(stripped_path_components('/foo', ['/foo']), [('', '/foo')])
- WVPASSEQ(stripped_path_components('/foo/bar', ['/foo']),
- [('', '/foo'), ('bar', '/foo/bar')])
- WVPASSEQ(stripped_path_components('/foo/bar', ['/bar', '/foo', '/baz']),
- [('', '/foo'), ('bar', '/foo/bar')])
- WVPASSEQ(stripped_path_components('/foo/bar/baz', ['/foo/bar/baz']),
- [('', '/foo/bar/baz')])
- WVEXCEPT(Exception, stripped_path_components, 'foo', [])
+ with no_lingering_errors():
+ WVPASSEQ(stripped_path_components('/', []), [('', '/')])
+ WVPASSEQ(stripped_path_components('/', ['']), [('', '/')])
+ WVPASSEQ(stripped_path_components('/', ['/']), [('', '/')])
+ WVPASSEQ(stripped_path_components('/foo', ['/']),
+ [('', '/'), ('foo', '/foo')])
+ WVPASSEQ(stripped_path_components('/', ['/foo']), [('', '/')])
+ WVPASSEQ(stripped_path_components('/foo', ['/bar']),
+ [('', '/'), ('foo', '/foo')])
+ WVPASSEQ(stripped_path_components('/foo', ['/foo']), [('', '/foo')])
+ WVPASSEQ(stripped_path_components('/foo/bar', ['/foo']),
+ [('', '/foo'), ('bar', '/foo/bar')])
+ WVPASSEQ(stripped_path_components('/foo/bar', ['/bar', '/foo', '/baz']),
+ [('', '/foo'), ('bar', '/foo/bar')])
+ WVPASSEQ(stripped_path_components('/foo/bar/baz', ['/foo/bar/baz']),
+ [('', '/foo/bar/baz')])
+ WVEXCEPT(Exception, stripped_path_components, 'foo', [])
@wvtest
def test_grafted_path_components():
- WVPASSEQ(grafted_path_components([('/chroot', '/')], '/foo'),
- [('', '/'), ('foo', '/foo')])
- WVPASSEQ(grafted_path_components([('/foo/bar', '/')], '/foo/bar/baz/bax'),
- [('', '/foo/bar'),
- ('baz', '/foo/bar/baz'),
- ('bax', '/foo/bar/baz/bax')])
- WVPASSEQ(grafted_path_components([('/foo/bar/baz', '/bax')],
- '/foo/bar/baz/1/2'),
- [('', None),
- ('bax', '/foo/bar/baz'),
- ('1', '/foo/bar/baz/1'),
- ('2', '/foo/bar/baz/1/2')])
- WVPASSEQ(grafted_path_components([('/foo', '/bar/baz/bax')],
- '/foo/bar'),
- [('', None),
- ('bar', None),
- ('baz', None),
- ('bax', '/foo'),
- ('bar', '/foo/bar')])
- WVPASSEQ(grafted_path_components([('/foo/bar/baz', '/a/b/c')],
- '/foo/bar/baz'),
- [('', None), ('a', None), ('b', None), ('c', '/foo/bar/baz')])
- WVPASSEQ(grafted_path_components([('/', '/a/b/c/')], '/foo/bar'),
- [('', None), ('a', None), ('b', None), ('c', '/'),
- ('foo', '/foo'), ('bar', '/foo/bar')])
- WVEXCEPT(Exception, grafted_path_components, 'foo', [])
+ with no_lingering_errors():
+ WVPASSEQ(grafted_path_components([('/chroot', '/')], '/foo'),
+ [('', '/'), ('foo', '/foo')])
+ WVPASSEQ(grafted_path_components([('/foo/bar', '/')],
+ '/foo/bar/baz/bax'),
+ [('', '/foo/bar'),
+ ('baz', '/foo/bar/baz'),
+ ('bax', '/foo/bar/baz/bax')])
+ WVPASSEQ(grafted_path_components([('/foo/bar/baz', '/bax')],
+ '/foo/bar/baz/1/2'),
+ [('', None),
+ ('bax', '/foo/bar/baz'),
+ ('1', '/foo/bar/baz/1'),
+ ('2', '/foo/bar/baz/1/2')])
+ WVPASSEQ(grafted_path_components([('/foo', '/bar/baz/bax')],
+ '/foo/bar'),
+ [('', None),
+ ('bar', None),
+ ('baz', None),
+ ('bax', '/foo'),
+ ('bar', '/foo/bar')])
+ WVPASSEQ(grafted_path_components([('/foo/bar/baz', '/a/b/c')],
+ '/foo/bar/baz'),
+ [('', None), ('a', None), ('b', None), ('c', '/foo/bar/baz')])
+ WVPASSEQ(grafted_path_components([('/', '/a/b/c/')], '/foo/bar'),
+ [('', None), ('a', None), ('b', None), ('c', '/'),
+ ('foo', '/foo'), ('bar', '/foo/bar')])
+ WVEXCEPT(Exception, grafted_path_components, 'foo', [])
@wvtest
def test_readpipe():
- x = readpipe(['echo', '42'])
- WVPASSEQ(x, '42\n')
- try:
- readpipe(['bash', '-c', 'exit 42'])
- except Exception as ex:
- WVPASSEQ(str(ex), "subprocess 'bash -c exit 42' failed with status 42")
+ with no_lingering_errors():
+ x = readpipe(['echo', '42'])
+ WVPASSEQ(x, '42\n')
+ try:
+ readpipe(['bash', '-c', 'exit 42'])
+ except Exception as ex:
+ WVPASSEQ(str(ex),
+ "subprocess 'bash -c exit 42' failed with status 42")
@wvtest
def test_batchpipe():
- for chunk in batchpipe(['echo'], []):
- WVPASS(False)
- out = ''
- for chunk in batchpipe(['echo'], ['42']):
- out += chunk
- WVPASSEQ(out, '42\n')
- try:
- batchpipe(['bash', '-c'], ['exit 42'])
- except Exception as ex:
- WVPASSEQ(str(ex), "subprocess 'bash -c exit 42' failed with status 42")
- args = [str(x) for x in range(6)]
- # Force batchpipe to break the args into batches of 3. This
- # approach assumes all args are the same length.
- arg_max = \
- helpers._argmax_base(['echo']) + helpers._argmax_args_size(args[:3])
- batches = batchpipe(['echo'], args, arg_max=arg_max)
- WVPASSEQ(next(batches), '0 1 2\n')
- WVPASSEQ(next(batches), '3 4 5\n')
- WVPASSEQ(next(batches, None), None)
- batches = batchpipe(['echo'], [str(x) for x in range(5)], arg_max=arg_max)
- WVPASSEQ(next(batches), '0 1 2\n')
- WVPASSEQ(next(batches), '3 4\n')
- WVPASSEQ(next(batches, None), None)
+ with no_lingering_errors():
+ for chunk in batchpipe(['echo'], []):
+ WVPASS(False)
+ out = ''
+ for chunk in batchpipe(['echo'], ['42']):
+ out += chunk
+ WVPASSEQ(out, '42\n')
+ try:
+ batchpipe(['bash', '-c'], ['exit 42'])
+ except Exception as ex:
+ WVPASSEQ(str(ex),
+ "subprocess 'bash -c exit 42' failed with status 42")
+ args = [str(x) for x in range(6)]
+ # Force batchpipe to break the args into batches of 3. This
+ # approach assumes all args are the same length.
+ arg_max = \
+ helpers._argmax_base(['echo']) + helpers._argmax_args_size(args[:3])
+ batches = batchpipe(['echo'], args, arg_max=arg_max)
+ WVPASSEQ(next(batches), '0 1 2\n')
+ WVPASSEQ(next(batches), '3 4 5\n')
+ WVPASSEQ(next(batches, None), None)
+ batches = batchpipe(['echo'], [str(x) for x in range(5)], arg_max=arg_max)
+ WVPASSEQ(next(batches), '0 1 2\n')
+ WVPASSEQ(next(batches), '3 4\n')
+ WVPASSEQ(next(batches, None), None)
@wvtest
def test_atomically_replaced_file():
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-thelper-')
- target_file = os.path.join(tmpdir, 'test-atomic-write')
- initial_failures = wvfailure_count()
+ with no_lingering_errors(), test_tempdir('bup-thelper-') as tmpdir:
+ target_file = os.path.join(tmpdir, 'test-atomic-write')
- with atomically_replaced_file(target_file, mode='w') as f:
- f.write('asdf')
- WVPASSEQ(f.mode, 'w')
- f = open(target_file, 'r')
- WVPASSEQ(f.read(), 'asdf')
-
- try:
with atomically_replaced_file(target_file, mode='w') as f:
- f.write('wxyz')
- raise Exception()
- except:
- pass
- with open(target_file) as f:
+ f.write('asdf')
+ WVPASSEQ(f.mode, 'w')
+ f = open(target_file, 'r')
WVPASSEQ(f.read(), 'asdf')
- with atomically_replaced_file(target_file, mode='wb') as f:
- f.write(os.urandom(20))
- WVPASSEQ(f.mode, 'wb')
+ try:
+ with atomically_replaced_file(target_file, mode='w') as f:
+ f.write('wxyz')
+ raise Exception()
+ except:
+ pass
+ with open(target_file) as f:
+ WVPASSEQ(f.read(), 'asdf')
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with atomically_replaced_file(target_file, mode='wb') as f:
+ f.write(os.urandom(20))
+ WVPASSEQ(f.mode, 'wb')
@wvtest
def test_utc_offset_str():
- tz = os.environ.get('TZ')
- try:
- os.environ['TZ'] = 'FOO+0:00'
- WVPASSEQ(utc_offset_str(0), '+0000')
- os.environ['TZ'] = 'FOO+1:00'
- WVPASSEQ(utc_offset_str(0), '-0100')
- os.environ['TZ'] = 'FOO-1:00'
- WVPASSEQ(utc_offset_str(0), '+0100')
- os.environ['TZ'] = 'FOO+3:3'
- WVPASSEQ(utc_offset_str(0), '-0303')
- os.environ['TZ'] = 'FOO-3:3'
- WVPASSEQ(utc_offset_str(0), '+0303')
- # Offset is not an integer number of minutes
- os.environ['TZ'] = 'FOO+3:3:3'
- WVPASSEQ(utc_offset_str(1), '-0303')
- os.environ['TZ'] = 'FOO-3:3:3'
- WVPASSEQ(utc_offset_str(1), '+0303')
- WVPASSEQ(utc_offset_str(314159), '+0303')
- finally:
- if tz:
- os.environ['TZ'] = tz
- else:
- try:
- del os.environ['TZ']
- except KeyError:
- pass
+ with no_lingering_errors():
+ tz = os.environ.get('TZ')
+ try:
+ os.environ['TZ'] = 'FOO+0:00'
+ WVPASSEQ(utc_offset_str(0), '+0000')
+ os.environ['TZ'] = 'FOO+1:00'
+ WVPASSEQ(utc_offset_str(0), '-0100')
+ os.environ['TZ'] = 'FOO-1:00'
+ WVPASSEQ(utc_offset_str(0), '+0100')
+ os.environ['TZ'] = 'FOO+3:3'
+ WVPASSEQ(utc_offset_str(0), '-0303')
+ os.environ['TZ'] = 'FOO-3:3'
+ WVPASSEQ(utc_offset_str(0), '+0303')
+ # Offset is not an integer number of minutes
+ os.environ['TZ'] = 'FOO+3:3:3'
+ WVPASSEQ(utc_offset_str(1), '-0303')
+ os.environ['TZ'] = 'FOO-3:3:3'
+ WVPASSEQ(utc_offset_str(1), '+0303')
+ WVPASSEQ(utc_offset_str(314159), '+0303')
+ finally:
+ if tz:
+ os.environ['TZ'] = tz
+ else:
+ try:
+ del os.environ['TZ']
+ except KeyError:
+ pass
-import os, subprocess, time, tempfile
+import os, time
+
+from wvtest import *
from bup import index, metadata
from bup.helpers import mkdirp, resolve_parent
+from buptest import no_lingering_errors, test_tempdir
import bup.xstat as xstat
-from wvtest import *
+lib_t_dir = os.path.dirname(__file__)
-lib_t_dir = os.getcwd()
-bup_tmp = os.path.realpath('../../../t/tmp')
-mkdirp(bup_tmp)
@wvtest
def index_basic():
- cd = os.path.realpath('../../../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')
+ with no_lingering_errors():
+ cd = os.path.realpath('../../../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')
@wvtest
def index_writer():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tindex-')
- 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');
- 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)
- ms.close()
- w.close()
- finally:
- os.chdir(orig_cwd)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('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');
+ 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)
+ ms.close()
+ w.close()
+ finally:
+ os.chdir(orig_cwd)
def dump(m):
@wvtest
def index_negative_timestamps():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tindex-')
- # Makes 'foo' exist
- foopath = tmpdir + '/foo'
- f = file(foopath, 'wb')
- f.close()
-
- # Dec 31, 1969
- os.utime(foopath, (-86400, -86400))
- ns_per_sec = 10**9
- tstart = time.time() * ns_per_sec
- tmax = tstart - ns_per_sec
- e = index.BlankNewEntry(foopath, 0, tmax)
- e.from_stat(xstat.stat(foopath), 0, tstart)
- assert len(e.packed())
- WVPASS()
-
- # Jun 10, 1893
- os.utime(foopath, (-0x80000000, -0x80000000))
- e = index.BlankNewEntry(foopath, 0, tmax)
- e.from_stat(xstat.stat(foopath), 0, tstart)
- assert len(e.packed())
- WVPASS()
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
-
-
-@wvtest
-def index_dirty():
- initial_failures = wvfailure_count()
- orig_cwd = os.getcwd()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tindex-')
- 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')
- 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')
- 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.close()
+ with no_lingering_errors(), test_tempdir('bup-tindex-') as tmpdir:
+ # Makes 'foo' exist
+ foopath = tmpdir + '/foo'
+ f = file(foopath, 'wb')
+ f.close()
+
+ # Dec 31, 1969
+ os.utime(foopath, (-86400, -86400))
+ ns_per_sec = 10**9
+ tstart = time.time() * ns_per_sec
+ tmax = tstart - ns_per_sec
+ e = index.BlankNewEntry(foopath, 0, tmax)
+ e.from_stat(xstat.stat(foopath), 0, tstart)
+ assert len(e.packed())
WVPASS()
- w2 = index.Writer('index2.tmp', ms2, tmax)
- w2.add('/a/b/n/2', fs, meta_ofs2)
- #w2.close()
+ # Jun 10, 1893
+ os.utime(foopath, (-0x80000000, -0x80000000))
+ e = index.BlankNewEntry(foopath, 0, tmax)
+ e.from_stat(xstat.stat(foopath), 0, tstart)
+ assert len(e.packed())
WVPASS()
- w3 = index.Writer('index3.tmp', ms3, tmax)
- w3.add('/a/c/n/3', fs, meta_ofs3)
- #w3.close()
- WVPASS()
-
- r1 = w1.new_reader()
- r2 = w2.new_reader()
- r3 = w3.new_reader()
- WVPASS()
- r1all = [e.name for e in r1]
- WVPASSEQ(r1all,
- ['/a/b/x', '/a/b/c', '/a/b/', '/a/', '/'])
- r2all = [e.name for e in r2]
- WVPASSEQ(r2all,
- ['/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
- r3all = [e.name for e in r3]
- WVPASSEQ(r3all,
- ['/a/c/n/3', '/a/c/n/', '/a/c/', '/a/', '/'])
- 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/', '/'])
- fake_validate(r1)
- dump(r1)
-
- print [hex(e.flags) for e in 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/', '/'])
-
- expect_invalid = ['/'] + r2all + r3all
- expect_real = (set(r1all) - set(r2all) - set(r3all)) \
- | set(['/a/b/n/2', '/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
- eiv = e.name in expect_invalid
- er = e.name in expect_real
- WVPASSEQ(eiv, not e.is_valid())
- WVPASSEQ(er, e.is_real())
- fake_validate(r2, r3)
- 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.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/', '/'])
- w1.close()
- w2.close()
- w3.close()
- finally:
- os.chdir(orig_cwd)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+@wvtest
+def index_dirty():
+ with no_lingering_errors(), test_tempdir('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')
+ 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')
+ 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.close()
+ WVPASS()
+
+ w2 = index.Writer('index2.tmp', ms2, tmax)
+ w2.add('/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.close()
+ WVPASS()
+
+ r1 = w1.new_reader()
+ r2 = w2.new_reader()
+ r3 = w3.new_reader()
+ WVPASS()
+
+ r1all = [e.name for e in r1]
+ WVPASSEQ(r1all,
+ ['/a/b/x', '/a/b/c', '/a/b/', '/a/', '/'])
+ r2all = [e.name for e in r2]
+ WVPASSEQ(r2all,
+ ['/a/b/n/2', '/a/b/n/', '/a/b/', '/a/', '/'])
+ r3all = [e.name for e in r3]
+ WVPASSEQ(r3all,
+ ['/a/c/n/3', '/a/c/n/', '/a/c/', '/a/', '/'])
+ 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/', '/'])
+ fake_validate(r1)
+ dump(r1)
+
+ print [hex(e.flags) for e in 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/', '/'])
+
+ expect_invalid = ['/'] + r2all + r3all
+ expect_real = (set(r1all) - set(r2all) - set(r3all)) \
+ | set(['/a/b/n/2', '/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
+ eiv = e.name in expect_invalid
+ er = e.name in expect_real
+ WVPASSEQ(eiv, not e.is_valid())
+ WVPASSEQ(er, e.is_real())
+ fake_validate(r2, r3)
+ 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.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/', '/'])
+ w1.close()
+ w2.close()
+ w3.close()
+ finally:
+ os.chdir(orig_cwd)
+
import errno, glob, grp, pwd, stat, tempfile, subprocess
-import bup.helpers as helpers
+
+from wvtest import *
+
from bup import git, metadata, vfs
from bup.helpers import clear_errors, detect_fakeroot, is_superuser, resolve_parent
-from wvtest import *
from bup.xstat import utime, lutime
+from buptest import no_lingering_errors, test_tempdir
+import bup.helpers as helpers
top_dir = '../../..'
@wvtest
def test_clean_up_archive_path():
- cleanup = metadata._clean_up_path_for_archive
- WVPASSEQ(cleanup('foo'), 'foo')
- WVPASSEQ(cleanup('/foo'), 'foo')
- WVPASSEQ(cleanup('///foo'), 'foo')
- WVPASSEQ(cleanup('/foo/bar'), 'foo/bar')
- WVPASSEQ(cleanup('foo/./bar'), 'foo/bar')
- WVPASSEQ(cleanup('/foo/./bar'), 'foo/bar')
- WVPASSEQ(cleanup('/foo/./bar/././baz'), 'foo/bar/baz')
- WVPASSEQ(cleanup('/foo/./bar///././baz'), 'foo/bar/baz')
- WVPASSEQ(cleanup('//./foo/./bar///././baz/.///'), 'foo/bar/baz/')
- WVPASSEQ(cleanup('./foo/./.bar'), 'foo/.bar')
- WVPASSEQ(cleanup('./foo/.'), 'foo')
- WVPASSEQ(cleanup('./foo/..'), '.')
- WVPASSEQ(cleanup('//./..//.../..//.'), '.')
- WVPASSEQ(cleanup('//./..//..././/.'), '...')
- WVPASSEQ(cleanup('/////.'), '.')
- WVPASSEQ(cleanup('/../'), '.')
- WVPASSEQ(cleanup(''), '.')
+ with no_lingering_errors():
+ cleanup = metadata._clean_up_path_for_archive
+ WVPASSEQ(cleanup('foo'), 'foo')
+ WVPASSEQ(cleanup('/foo'), 'foo')
+ WVPASSEQ(cleanup('///foo'), 'foo')
+ WVPASSEQ(cleanup('/foo/bar'), 'foo/bar')
+ WVPASSEQ(cleanup('foo/./bar'), 'foo/bar')
+ WVPASSEQ(cleanup('/foo/./bar'), 'foo/bar')
+ WVPASSEQ(cleanup('/foo/./bar/././baz'), 'foo/bar/baz')
+ WVPASSEQ(cleanup('/foo/./bar///././baz'), 'foo/bar/baz')
+ WVPASSEQ(cleanup('//./foo/./bar///././baz/.///'), 'foo/bar/baz/')
+ WVPASSEQ(cleanup('./foo/./.bar'), 'foo/.bar')
+ WVPASSEQ(cleanup('./foo/.'), 'foo')
+ WVPASSEQ(cleanup('./foo/..'), '.')
+ WVPASSEQ(cleanup('//./..//.../..//.'), '.')
+ WVPASSEQ(cleanup('//./..//..././/.'), '...')
+ WVPASSEQ(cleanup('/////.'), '.')
+ WVPASSEQ(cleanup('/../'), '.')
+ WVPASSEQ(cleanup(''), '.')
@wvtest
def test_risky_path():
- risky = metadata._risky_path
- WVPASS(risky('/foo'))
- WVPASS(risky('///foo'))
- WVPASS(risky('/../foo'))
- WVPASS(risky('../foo'))
- WVPASS(risky('foo/..'))
- WVPASS(risky('foo/../'))
- WVPASS(risky('foo/../bar'))
- WVFAIL(risky('foo'))
- WVFAIL(risky('foo/'))
- WVFAIL(risky('foo///'))
- WVFAIL(risky('./foo'))
- WVFAIL(risky('foo/.'))
- WVFAIL(risky('./foo/.'))
- WVFAIL(risky('foo/bar'))
- WVFAIL(risky('foo/./bar'))
+ with no_lingering_errors():
+ risky = metadata._risky_path
+ WVPASS(risky('/foo'))
+ WVPASS(risky('///foo'))
+ WVPASS(risky('/../foo'))
+ WVPASS(risky('../foo'))
+ WVPASS(risky('foo/..'))
+ WVPASS(risky('foo/../'))
+ WVPASS(risky('foo/../bar'))
+ WVFAIL(risky('foo'))
+ WVFAIL(risky('foo/'))
+ WVFAIL(risky('foo///'))
+ WVFAIL(risky('./foo'))
+ WVFAIL(risky('foo/.'))
+ WVFAIL(risky('./foo/.'))
+ WVFAIL(risky('foo/bar'))
+ WVFAIL(risky('foo/./bar'))
@wvtest
def test_clean_up_extract_path():
- cleanup = metadata._clean_up_extract_path
- WVPASSEQ(cleanup('/foo'), 'foo')
- WVPASSEQ(cleanup('///foo'), 'foo')
- WVFAIL(cleanup('/../foo'))
- WVFAIL(cleanup('../foo'))
- WVFAIL(cleanup('foo/..'))
- WVFAIL(cleanup('foo/../'))
- WVFAIL(cleanup('foo/../bar'))
- WVPASSEQ(cleanup('foo'), 'foo')
- WVPASSEQ(cleanup('foo/'), 'foo/')
- WVPASSEQ(cleanup('foo///'), 'foo///')
- WVPASSEQ(cleanup('./foo'), './foo')
- WVPASSEQ(cleanup('foo/.'), 'foo/.')
- WVPASSEQ(cleanup('./foo/.'), './foo/.')
- WVPASSEQ(cleanup('foo/bar'), 'foo/bar')
- WVPASSEQ(cleanup('foo/./bar'), 'foo/./bar')
- WVPASSEQ(cleanup('/'), '.')
- WVPASSEQ(cleanup('./'), './')
- WVPASSEQ(cleanup('///foo/bar'), 'foo/bar')
- WVPASSEQ(cleanup('///foo/bar'), 'foo/bar')
+ with no_lingering_errors():
+ cleanup = metadata._clean_up_extract_path
+ WVPASSEQ(cleanup('/foo'), 'foo')
+ WVPASSEQ(cleanup('///foo'), 'foo')
+ WVFAIL(cleanup('/../foo'))
+ WVFAIL(cleanup('../foo'))
+ WVFAIL(cleanup('foo/..'))
+ WVFAIL(cleanup('foo/../'))
+ WVFAIL(cleanup('foo/../bar'))
+ WVPASSEQ(cleanup('foo'), 'foo')
+ WVPASSEQ(cleanup('foo/'), 'foo/')
+ WVPASSEQ(cleanup('foo///'), 'foo///')
+ WVPASSEQ(cleanup('./foo'), './foo')
+ WVPASSEQ(cleanup('foo/.'), 'foo/.')
+ WVPASSEQ(cleanup('./foo/.'), './foo/.')
+ WVPASSEQ(cleanup('foo/bar'), 'foo/bar')
+ WVPASSEQ(cleanup('foo/./bar'), 'foo/./bar')
+ WVPASSEQ(cleanup('/'), '.')
+ WVPASSEQ(cleanup('./'), './')
+ WVPASSEQ(cleanup('///foo/bar'), 'foo/bar')
+ WVPASSEQ(cleanup('///foo/bar'), 'foo/bar')
@wvtest
def test_metadata_method():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tmetadata-')
- bup_dir = tmpdir + '/bup'
- data_path = tmpdir + '/foo'
- os.mkdir(data_path)
- ex('touch', data_path + '/file')
- ex('ln', '-s', 'file', data_path + '/symlink')
- test_time1 = 13 * 1000000000
- test_time2 = 42 * 1000000000
- utime(data_path + '/file', (0, test_time1))
- lutime(data_path + '/symlink', (0, 0))
- utime(data_path, (0, test_time2))
- ex(bup_path, '-d', bup_dir, 'init')
- ex(bup_path, '-d', bup_dir, 'index', '-v', data_path)
- ex(bup_path, '-d', bup_dir, 'save', '-tvvn', 'test', data_path)
- git.check_repo_or_die(bup_dir)
- top = vfs.RefList(None)
- n = top.lresolve('/test/latest' + resolve_parent(data_path))
- m = n.metadata()
- WVPASS(m.mtime == test_time2)
- WVPASS(len(n.subs()) == 2)
- WVPASS(n.name == 'foo')
- WVPASS(set([x.name for x in n.subs()]) == set(['file', 'symlink']))
- for sub in n:
- if sub.name == 'file':
- m = sub.metadata()
- WVPASS(m.mtime == test_time1)
- elif sub.name == 'symlink':
- m = sub.metadata()
- WVPASS(m.mtime == 0)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tmetadata-') as tmpdir:
+ bup_dir = tmpdir + '/bup'
+ data_path = tmpdir + '/foo'
+ os.mkdir(data_path)
+ ex('touch', data_path + '/file')
+ ex('ln', '-s', 'file', data_path + '/symlink')
+ test_time1 = 13 * 1000000000
+ test_time2 = 42 * 1000000000
+ utime(data_path + '/file', (0, test_time1))
+ lutime(data_path + '/symlink', (0, 0))
+ utime(data_path, (0, test_time2))
+ ex(bup_path, '-d', bup_dir, 'init')
+ ex(bup_path, '-d', bup_dir, 'index', '-v', data_path)
+ ex(bup_path, '-d', bup_dir, 'save', '-tvvn', 'test', data_path)
+ git.check_repo_or_die(bup_dir)
+ top = vfs.RefList(None)
+ n = top.lresolve('/test/latest' + resolve_parent(data_path))
+ m = n.metadata()
+ WVPASS(m.mtime == test_time2)
+ WVPASS(len(n.subs()) == 2)
+ WVPASS(n.name == 'foo')
+ WVPASS(set([x.name for x in n.subs()]) == set(['file', 'symlink']))
+ for sub in n:
+ if sub.name == 'file':
+ m = sub.metadata()
+ WVPASS(m.mtime == test_time1)
+ elif sub.name == 'symlink':
+ m = sub.metadata()
+ WVPASS(m.mtime == 0)
def _first_err():
@wvtest
def test_from_path_error():
- initial_failures = wvfailure_count()
if is_superuser() or detect_fakeroot():
return
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tmetadata-')
- path = tmpdir + '/foo'
- os.mkdir(path)
- m = metadata.from_path(path, archive_path=path, save_symlinks=True)
- WVPASSEQ(m.path, path)
- os.chmod(path, 000)
- metadata.from_path(path, archive_path=path, save_symlinks=True)
- if metadata.get_linux_file_attr:
- print >> sys.stderr, 'saved_errors:', helpers.saved_errors
- WVPASS(len(helpers.saved_errors) == 1)
- errmsg = _first_err()
- WVPASS(errmsg.startswith('read Linux attr'))
- clear_errors()
- if wvfailure_count() == initial_failures:
- subprocess.call(['chmod', '-R', 'u+rwX', tmpdir])
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tmetadata-') as tmpdir:
+ path = tmpdir + '/foo'
+ os.mkdir(path)
+ m = metadata.from_path(path, archive_path=path, save_symlinks=True)
+ WVPASSEQ(m.path, path)
+ os.chmod(path, 000)
+ metadata.from_path(path, archive_path=path, save_symlinks=True)
+ if metadata.get_linux_file_attr:
+ print >> sys.stderr, 'saved_errors:', helpers.saved_errors
+ WVPASS(len(helpers.saved_errors) == 1)
+ errmsg = _first_err()
+ WVPASS(errmsg.startswith('read Linux attr'))
+ clear_errors()
def _linux_attr_supported(path):
@wvtest
def test_apply_to_path_restricted_access():
- initial_failures = wvfailure_count()
if is_superuser() or detect_fakeroot():
return
if sys.platform.startswith('cygwin'):
return # chmod 000 isn't effective.
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tmetadata-')
- parent = tmpdir + '/foo'
- path = parent + '/bar'
- os.mkdir(parent)
- os.mkdir(path)
- clear_errors()
- m = metadata.from_path(path, archive_path=path, save_symlinks=True)
- WVPASSEQ(m.path, path)
- os.chmod(parent, 000)
- m.apply_to_path(path)
- print >> sys.stderr, 'saved_errors:', helpers.saved_errors
- expected_errors = ['utime: ']
- if m.linux_attr and _linux_attr_supported(tmpdir):
- expected_errors.append('Linux chattr: ')
- if metadata.xattr and m.linux_xattr:
- expected_errors.append("xattr.set '")
- WVPASS(len(helpers.saved_errors) == len(expected_errors))
- for i in xrange(len(expected_errors)):
- WVPASS(str(helpers.saved_errors[i]).startswith(expected_errors[i]))
- clear_errors()
- if wvfailure_count() == initial_failures:
- subprocess.call(['chmod', '-R', 'u+rwX', tmpdir])
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tmetadata-') as tmpdir:
+ parent = tmpdir + '/foo'
+ path = parent + '/bar'
+ os.mkdir(parent)
+ os.mkdir(path)
+ clear_errors()
+ m = metadata.from_path(path, archive_path=path, save_symlinks=True)
+ WVPASSEQ(m.path, path)
+ os.chmod(parent, 000)
+ m.apply_to_path(path)
+ print >> sys.stderr, 'saved_errors:', helpers.saved_errors
+ expected_errors = ['utime: ']
+ if m.linux_attr and _linux_attr_supported(tmpdir):
+ expected_errors.append('Linux chattr: ')
+ if metadata.xattr and m.linux_xattr:
+ expected_errors.append("xattr.set '")
+ WVPASS(len(helpers.saved_errors) == len(expected_errors))
+ for i in xrange(len(expected_errors)):
+ WVPASS(str(helpers.saved_errors[i]).startswith(expected_errors[i]))
+ clear_errors()
@wvtest
def test_restore_over_existing_target():
- initial_failures = wvfailure_count()
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-tmetadata-')
- path = tmpdir + '/foo'
- os.mkdir(path)
- dir_m = metadata.from_path(path, archive_path=path, save_symlinks=True)
- os.rmdir(path)
- open(path, 'w').close()
- file_m = metadata.from_path(path, archive_path=path, save_symlinks=True)
- # Restore dir over file.
- WVPASSEQ(dir_m.create_path(path, create_symlinks=True), None)
- WVPASS(stat.S_ISDIR(os.stat(path).st_mode))
- # Restore dir over dir.
- WVPASSEQ(dir_m.create_path(path, create_symlinks=True), None)
- WVPASS(stat.S_ISDIR(os.stat(path).st_mode))
- # Restore file over dir.
- WVPASSEQ(file_m.create_path(path, create_symlinks=True), None)
- WVPASS(stat.S_ISREG(os.stat(path).st_mode))
- # Restore file over file.
- WVPASSEQ(file_m.create_path(path, create_symlinks=True), None)
- WVPASS(stat.S_ISREG(os.stat(path).st_mode))
- # Restore file over non-empty dir.
- os.remove(path)
- os.mkdir(path)
- open(path + '/bar', 'w').close()
- WVEXCEPT(Exception, file_m.create_path, path, create_symlinks=True)
- # Restore dir over non-empty dir.
- os.remove(path + '/bar')
- os.mkdir(path + '/bar')
- WVEXCEPT(Exception, dir_m.create_path, path, create_symlinks=True)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-tmetadata-') as tmpdir:
+ path = tmpdir + '/foo'
+ os.mkdir(path)
+ dir_m = metadata.from_path(path, archive_path=path, save_symlinks=True)
+ os.rmdir(path)
+ open(path, 'w').close()
+ file_m = metadata.from_path(path, archive_path=path, save_symlinks=True)
+ # Restore dir over file.
+ WVPASSEQ(dir_m.create_path(path, create_symlinks=True), None)
+ WVPASS(stat.S_ISDIR(os.stat(path).st_mode))
+ # Restore dir over dir.
+ WVPASSEQ(dir_m.create_path(path, create_symlinks=True), None)
+ WVPASS(stat.S_ISDIR(os.stat(path).st_mode))
+ # Restore file over dir.
+ WVPASSEQ(file_m.create_path(path, create_symlinks=True), None)
+ WVPASS(stat.S_ISREG(os.stat(path).st_mode))
+ # Restore file over file.
+ WVPASSEQ(file_m.create_path(path, create_symlinks=True), None)
+ WVPASS(stat.S_ISREG(os.stat(path).st_mode))
+ # Restore file over non-empty dir.
+ os.remove(path)
+ os.mkdir(path)
+ open(path + '/bar', 'w').close()
+ WVEXCEPT(Exception, file_m.create_path, path, create_symlinks=True)
+ # Restore dir over non-empty dir.
+ os.remove(path + '/bar')
+ os.mkdir(path + '/bar')
+ WVEXCEPT(Exception, dir_m.create_path, path, create_symlinks=True)
from bup.metadata import posix1e
+
from bup import options
+
from wvtest import *
+from buptest import no_lingering_errors
+
@wvtest
def test_optdict():
- d = options.OptDict({
- 'x': ('x', False),
- 'y': ('y', False),
- 'z': ('z', False),
- 'other_thing': ('other_thing', False),
- 'no_other_thing': ('other_thing', True),
- 'no_z': ('z', True),
- 'no_smart': ('smart', True),
- 'smart': ('smart', False),
- 'stupid': ('smart', True),
- 'no_smart': ('smart', False),
- })
- WVPASS('foo')
- d['x'] = 5
- d['y'] = 4
- d['z'] = 99
- d['no_other_thing'] = 5
- WVPASSEQ(d.x, 5)
- WVPASSEQ(d.y, 4)
- WVPASSEQ(d.z, 99)
- WVPASSEQ(d.no_z, False)
- WVPASSEQ(d.no_other_thing, True)
- WVEXCEPT(KeyError, lambda: d.p)
+ with no_lingering_errors():
+ d = options.OptDict({
+ 'x': ('x', False),
+ 'y': ('y', False),
+ 'z': ('z', False),
+ 'other_thing': ('other_thing', False),
+ 'no_other_thing': ('other_thing', True),
+ 'no_z': ('z', True),
+ 'no_smart': ('smart', True),
+ 'smart': ('smart', False),
+ 'stupid': ('smart', True),
+ 'no_smart': ('smart', False),
+ })
+ WVPASS('foo')
+ d['x'] = 5
+ d['y'] = 4
+ d['z'] = 99
+ d['no_other_thing'] = 5
+ WVPASSEQ(d.x, 5)
+ WVPASSEQ(d.y, 4)
+ WVPASSEQ(d.z, 99)
+ WVPASSEQ(d.no_z, False)
+ WVPASSEQ(d.no_other_thing, True)
+ WVEXCEPT(KeyError, lambda: d.p)
invalid_optspec0 = """
@wvtest
def test_invalid_optspec():
- WVPASS(options.Options(invalid_optspec0).parse([]))
- WVPASS(options.Options(invalid_optspec1).parse([]))
- WVPASS(options.Options(invalid_optspec2).parse([]))
+ with no_lingering_errors():
+ WVPASS(options.Options(invalid_optspec0).parse([]))
+ WVPASS(options.Options(invalid_optspec1).parse([]))
+ WVPASS(options.Options(invalid_optspec2).parse([]))
optspec = """
@wvtest
def test_options():
- o = options.Options(optspec)
- (opt,flags,extra) = o.parse(['-tttqp', 7, '--longoption', '19',
- 'hanky', '--onlylong', '-7'])
- WVPASSEQ(flags[0], ('-t', ''))
- WVPASSEQ(flags[1], ('-t', ''))
- WVPASSEQ(flags[2], ('-t', ''))
- WVPASSEQ(flags[3], ('-q', ''))
- WVPASSEQ(flags[4], ('-p', 7))
- WVPASSEQ(flags[5], ('--longoption', '19'))
- WVPASSEQ(extra, ['hanky'])
- WVPASSEQ((opt.t, opt.q, opt.p, opt.l, opt.onlylong,
- opt.neveropt), (3,1,7,19,1,None))
- WVPASSEQ((opt.deftest1, opt.deftest2, opt.deftest3, opt.deftest4,
- opt.deftest5), (1,2,None,None,'[square'))
- WVPASSEQ((opt.stupid, opt.no_stupid), (True, None))
- WVPASSEQ((opt.smart, opt.no_smart), (None, True))
- WVPASSEQ((opt.x, opt.extended, opt.no_simple), (2,2,2))
- WVPASSEQ((opt.no_x, opt.no_extended, opt.simple), (False,False,False))
- WVPASSEQ(opt['#'], 7)
- WVPASSEQ(opt.compress, 7)
-
- (opt,flags,extra) = o.parse(['--onlylong', '-t', '--no-onlylong',
- '--smart', '--simple'])
- WVPASSEQ((opt.t, opt.q, opt.onlylong), (1, None, 0))
- WVPASSEQ((opt.stupid, opt.no_stupid), (False, True))
- WVPASSEQ((opt.smart, opt.no_smart), (True, False))
- WVPASSEQ((opt.x, opt.extended, opt.no_simple), (0,0,0))
- WVPASSEQ((opt.no_x, opt.no_extended, opt.simple), (True,True,True))
+ with no_lingering_errors():
+ o = options.Options(optspec)
+ (opt,flags,extra) = o.parse(['-tttqp', 7, '--longoption', '19',
+ 'hanky', '--onlylong', '-7'])
+ WVPASSEQ(flags[0], ('-t', ''))
+ WVPASSEQ(flags[1], ('-t', ''))
+ WVPASSEQ(flags[2], ('-t', ''))
+ WVPASSEQ(flags[3], ('-q', ''))
+ WVPASSEQ(flags[4], ('-p', 7))
+ WVPASSEQ(flags[5], ('--longoption', '19'))
+ WVPASSEQ(extra, ['hanky'])
+ WVPASSEQ((opt.t, opt.q, opt.p, opt.l, opt.onlylong,
+ opt.neveropt), (3,1,7,19,1,None))
+ WVPASSEQ((opt.deftest1, opt.deftest2, opt.deftest3, opt.deftest4,
+ opt.deftest5), (1,2,None,None,'[square'))
+ WVPASSEQ((opt.stupid, opt.no_stupid), (True, None))
+ WVPASSEQ((opt.smart, opt.no_smart), (None, True))
+ WVPASSEQ((opt.x, opt.extended, opt.no_simple), (2,2,2))
+ WVPASSEQ((opt.no_x, opt.no_extended, opt.simple), (False,False,False))
+ WVPASSEQ(opt['#'], 7)
+ WVPASSEQ(opt.compress, 7)
+
+ (opt,flags,extra) = o.parse(['--onlylong', '-t', '--no-onlylong',
+ '--smart', '--simple'])
+ WVPASSEQ((opt.t, opt.q, opt.onlylong), (1, None, 0))
+ WVPASSEQ((opt.stupid, opt.no_stupid), (False, True))
+ WVPASSEQ((opt.smart, opt.no_smart), (True, False))
+ WVPASSEQ((opt.x, opt.extended, opt.no_simple), (0,0,0))
+ WVPASSEQ((opt.no_x, opt.no_extended, opt.simple), (True,True,True))
-from bup import shquote
+
from wvtest import *
+from bup import shquote
+from buptest import no_lingering_errors
+
+
def qst(line):
return [word for offset,word in shquote.quotesplit(line)]
@wvtest
def test_shquote():
- WVPASSEQ(qst(""" this is basic \t\n\r text """),
- ['this', 'is', 'basic', 'text'])
- WVPASSEQ(qst(r""" \"x\" "help" 'yelp' """), ['"x"', 'help', 'yelp'])
- WVPASSEQ(qst(r""" "'\"\"'" '\"\'' """), ["'\"\"'", '\\"\''])
-
- WVPASSEQ(shquote.quotesplit(' this is "unfinished'),
- [(2,'this'), (7,'is'), (10,'unfinished')])
-
- WVPASSEQ(shquote.quotesplit('"silly"\'will'),
- [(0,'silly'), (7,'will')])
-
- WVPASSEQ(shquote.unfinished_word('this is a "billy" "goat'),
- ('"', 'goat'))
- WVPASSEQ(shquote.unfinished_word("'x"),
- ("'", 'x'))
- WVPASSEQ(shquote.unfinished_word("abra cadabra "),
- (None, ''))
- WVPASSEQ(shquote.unfinished_word("abra cadabra"),
- (None, 'cadabra'))
-
- (qtype, word) = shquote.unfinished_word("this is /usr/loc")
- WVPASSEQ(shquote.what_to_add(qtype, word, "/usr/local", True),
- "al")
- (qtype, word) = shquote.unfinished_word("this is '/usr/loc")
- WVPASSEQ(shquote.what_to_add(qtype, word, "/usr/local", True),
- "al'")
- (qtype, word) = shquote.unfinished_word("this is \"/usr/loc")
- WVPASSEQ(shquote.what_to_add(qtype, word, "/usr/local", True),
- "al\"")
- (qtype, word) = shquote.unfinished_word("this is \"/usr/loc")
- WVPASSEQ(shquote.what_to_add(qtype, word, "/usr/local", False),
- "al")
- (qtype, word) = shquote.unfinished_word("this is \\ hammer\\ \"")
- WVPASSEQ(word, ' hammer "')
- WVPASSEQ(shquote.what_to_add(qtype, word, " hammer \"time\"", True),
- "time\\\"")
-
- WVPASSEQ(shquote.quotify_list(['a', '', '"word"', "'third'", "'", "x y"]),
- "a '' '\"word\"' \"'third'\" \"'\" 'x y'")
+ with no_lingering_errors():
+ WVPASSEQ(qst(""" this is basic \t\n\r text """),
+ ['this', 'is', 'basic', 'text'])
+ WVPASSEQ(qst(r""" \"x\" "help" 'yelp' """), ['"x"', 'help', 'yelp'])
+ WVPASSEQ(qst(r""" "'\"\"'" '\"\'' """), ["'\"\"'", '\\"\''])
+
+ WVPASSEQ(shquote.quotesplit(' this is "unfinished'),
+ [(2,'this'), (7,'is'), (10,'unfinished')])
+
+ WVPASSEQ(shquote.quotesplit('"silly"\'will'),
+ [(0,'silly'), (7,'will')])
+
+ WVPASSEQ(shquote.unfinished_word('this is a "billy" "goat'),
+ ('"', 'goat'))
+ WVPASSEQ(shquote.unfinished_word("'x"),
+ ("'", 'x'))
+ WVPASSEQ(shquote.unfinished_word("abra cadabra "),
+ (None, ''))
+ WVPASSEQ(shquote.unfinished_word("abra cadabra"),
+ (None, 'cadabra'))
+
+ (qtype, word) = shquote.unfinished_word("this is /usr/loc")
+ WVPASSEQ(shquote.what_to_add(qtype, word, "/usr/local", True),
+ "al")
+ (qtype, word) = shquote.unfinished_word("this is '/usr/loc")
+ WVPASSEQ(shquote.what_to_add(qtype, word, "/usr/local", True),
+ "al'")
+ (qtype, word) = shquote.unfinished_word("this is \"/usr/loc")
+ WVPASSEQ(shquote.what_to_add(qtype, word, "/usr/local", True),
+ "al\"")
+ (qtype, word) = shquote.unfinished_word("this is \"/usr/loc")
+ WVPASSEQ(shquote.what_to_add(qtype, word, "/usr/local", False),
+ "al")
+ (qtype, word) = shquote.unfinished_word("this is \\ hammer\\ \"")
+ WVPASSEQ(word, ' hammer "')
+ WVPASSEQ(shquote.what_to_add(qtype, word, " hammer \"time\"", True),
+ "time\\\"")
+
+ WVPASSEQ(shquote.quotify_list(['a', '', '"word"', "'third'", "'", "x y"]),
+ "a '' '\"word\"' \"'third'\" \"'\" 'x y'")
from wvtest import *
from bup import vint
+from buptest import no_lingering_errors
def encode_and_decode_vuint(x):
@wvtest
def test_vuint():
- for x in (0, 1, 42, 128, 10**16):
- WVPASSEQ(encode_and_decode_vuint(x), x)
- WVEXCEPT(Exception, vint.write_vuint, BytesIO(), -1)
- WVEXCEPT(EOFError, vint.read_vuint, BytesIO())
+ with no_lingering_errors():
+ for x in (0, 1, 42, 128, 10**16):
+ WVPASSEQ(encode_and_decode_vuint(x), x)
+ WVEXCEPT(Exception, vint.write_vuint, BytesIO(), -1)
+ WVEXCEPT(EOFError, vint.read_vuint, BytesIO())
def encode_and_decode_vint(x):
@wvtest
def test_vint():
- values = (0, 1, 42, 64, 10**16)
- for x in values:
- WVPASSEQ(encode_and_decode_vint(x), x)
- for x in [-x for x in values]:
- WVPASSEQ(encode_and_decode_vint(x), x)
- WVEXCEPT(EOFError, vint.read_vint, BytesIO())
+ with no_lingering_errors():
+ values = (0, 1, 42, 64, 10**16)
+ for x in values:
+ WVPASSEQ(encode_and_decode_vint(x), x)
+ for x in [-x for x in values]:
+ WVPASSEQ(encode_and_decode_vint(x), x)
+ WVEXCEPT(EOFError, vint.read_vint, BytesIO())
def encode_and_decode_bvec(x):
@wvtest
def test_bvec():
- values = ('', 'x', 'foo', '\0', '\0foo', 'foo\0bar\0')
- for x in values:
- WVPASSEQ(encode_and_decode_bvec(x), x)
- WVEXCEPT(EOFError, vint.read_bvec, BytesIO())
- outf = BytesIO()
- for x in ('foo', 'bar', 'baz', 'bax'):
- vint.write_bvec(outf, x)
- inf = BytesIO(outf.getvalue())
- WVPASSEQ(vint.read_bvec(inf), 'foo')
- WVPASSEQ(vint.read_bvec(inf), 'bar')
- vint.skip_bvec(inf)
- WVPASSEQ(vint.read_bvec(inf), 'bax')
+ with no_lingering_errors():
+ values = ('', 'x', 'foo', '\0', '\0foo', 'foo\0bar\0')
+ for x in values:
+ WVPASSEQ(encode_and_decode_bvec(x), x)
+ WVEXCEPT(EOFError, vint.read_bvec, BytesIO())
+ outf = BytesIO()
+ for x in ('foo', 'bar', 'baz', 'bax'):
+ vint.write_bvec(outf, x)
+ inf = BytesIO(outf.getvalue())
+ WVPASSEQ(vint.read_bvec(inf), 'foo')
+ WVPASSEQ(vint.read_bvec(inf), 'bar')
+ vint.skip_bvec(inf)
+ WVPASSEQ(vint.read_bvec(inf), 'bax')
def pack_and_unpack(types, *values):
@wvtest
def test_pack_and_unpack():
- tests = [('', []),
- ('s', ['foo']),
- ('ss', ['foo', 'bar']),
- ('sV', ['foo', 0]),
- ('sv', ['foo', -1]),
- ('V', [0]),
- ('Vs', [0, 'foo']),
- ('VV', [0, 1]),
- ('Vv', [0, -1]),
- ('v', [0]),
- ('vs', [0, 'foo']),
- ('vV', [0, 1]),
- ('vv', [0, -1])]
- for test in tests:
- (types, values) = test
- WVPASSEQ(pack_and_unpack(types, *values), values)
- WVEXCEPT(Exception, vint.pack, 's')
- WVEXCEPT(Exception, vint.pack, 's', 'foo', 'bar')
- WVEXCEPT(Exception, vint.pack, 'x', 1)
- WVEXCEPT(Exception, vint.unpack, 's', '')
- WVEXCEPT(Exception, vint.unpack, 'x', '')
+ with no_lingering_errors():
+ tests = [('', []),
+ ('s', ['foo']),
+ ('ss', ['foo', 'bar']),
+ ('sV', ['foo', 0]),
+ ('sv', ['foo', -1]),
+ ('V', [0]),
+ ('Vs', [0, 'foo']),
+ ('VV', [0, 1]),
+ ('Vv', [0, -1]),
+ ('v', [0]),
+ ('vs', [0, 'foo']),
+ ('vV', [0, 1]),
+ ('vv', [0, -1])]
+ for test in tests:
+ (types, values) = test
+ WVPASSEQ(pack_and_unpack(types, *values), values)
+ WVEXCEPT(Exception, vint.pack, 's')
+ WVEXCEPT(Exception, vint.pack, 's', 'foo', 'bar')
+ WVEXCEPT(Exception, vint.pack, 'x', 1)
+ WVEXCEPT(Exception, vint.unpack, 's', '')
+ WVEXCEPT(Exception, vint.unpack, 'x', '')
import math, tempfile, subprocess
+
from wvtest import *
+
import bup._helpers as _helpers
from bup import xstat
+from buptest import no_lingering_errors, test_tempdir
-bup_tmp = os.path.realpath('../../../t/tmp')
@wvtest
def test_fstime():
- WVPASSEQ(xstat.timespec_to_nsecs((0, 0)), 0)
- WVPASSEQ(xstat.timespec_to_nsecs((1, 0)), 10**9)
- WVPASSEQ(xstat.timespec_to_nsecs((0, 10**9 / 2)), 500000000)
- WVPASSEQ(xstat.timespec_to_nsecs((1, 10**9 / 2)), 1500000000)
- WVPASSEQ(xstat.timespec_to_nsecs((-1, 0)), -10**9)
- WVPASSEQ(xstat.timespec_to_nsecs((-1, 10**9 / 2)), -500000000)
- WVPASSEQ(xstat.timespec_to_nsecs((-2, 10**9 / 2)), -1500000000)
- WVPASSEQ(xstat.timespec_to_nsecs((0, -1)), -1)
- WVPASSEQ(type(xstat.timespec_to_nsecs((2, 22222222))), type(0))
- WVPASSEQ(type(xstat.timespec_to_nsecs((-2, 22222222))), type(0))
+ with no_lingering_errors():
+ WVPASSEQ(xstat.timespec_to_nsecs((0, 0)), 0)
+ WVPASSEQ(xstat.timespec_to_nsecs((1, 0)), 10**9)
+ WVPASSEQ(xstat.timespec_to_nsecs((0, 10**9 / 2)), 500000000)
+ WVPASSEQ(xstat.timespec_to_nsecs((1, 10**9 / 2)), 1500000000)
+ WVPASSEQ(xstat.timespec_to_nsecs((-1, 0)), -10**9)
+ WVPASSEQ(xstat.timespec_to_nsecs((-1, 10**9 / 2)), -500000000)
+ WVPASSEQ(xstat.timespec_to_nsecs((-2, 10**9 / 2)), -1500000000)
+ WVPASSEQ(xstat.timespec_to_nsecs((0, -1)), -1)
+ WVPASSEQ(type(xstat.timespec_to_nsecs((2, 22222222))), type(0))
+ WVPASSEQ(type(xstat.timespec_to_nsecs((-2, 22222222))), type(0))
- WVPASSEQ(xstat.nsecs_to_timespec(0), (0, 0))
- WVPASSEQ(xstat.nsecs_to_timespec(10**9), (1, 0))
- WVPASSEQ(xstat.nsecs_to_timespec(500000000), (0, 10**9 / 2))
- WVPASSEQ(xstat.nsecs_to_timespec(1500000000), (1, 10**9 / 2))
- WVPASSEQ(xstat.nsecs_to_timespec(-10**9), (-1, 0))
- WVPASSEQ(xstat.nsecs_to_timespec(-500000000), (-1, 10**9 / 2))
- WVPASSEQ(xstat.nsecs_to_timespec(-1500000000), (-2, 10**9 / 2))
- x = xstat.nsecs_to_timespec(1977777778)
- WVPASSEQ(type(x[0]), type(0))
- WVPASSEQ(type(x[1]), type(0))
- x = xstat.nsecs_to_timespec(-1977777778)
- WVPASSEQ(type(x[0]), type(0))
- WVPASSEQ(type(x[1]), type(0))
+ WVPASSEQ(xstat.nsecs_to_timespec(0), (0, 0))
+ WVPASSEQ(xstat.nsecs_to_timespec(10**9), (1, 0))
+ WVPASSEQ(xstat.nsecs_to_timespec(500000000), (0, 10**9 / 2))
+ WVPASSEQ(xstat.nsecs_to_timespec(1500000000), (1, 10**9 / 2))
+ WVPASSEQ(xstat.nsecs_to_timespec(-10**9), (-1, 0))
+ WVPASSEQ(xstat.nsecs_to_timespec(-500000000), (-1, 10**9 / 2))
+ WVPASSEQ(xstat.nsecs_to_timespec(-1500000000), (-2, 10**9 / 2))
+ x = xstat.nsecs_to_timespec(1977777778)
+ WVPASSEQ(type(x[0]), type(0))
+ WVPASSEQ(type(x[1]), type(0))
+ x = xstat.nsecs_to_timespec(-1977777778)
+ WVPASSEQ(type(x[0]), type(0))
+ WVPASSEQ(type(x[1]), type(0))
- WVPASSEQ(xstat.nsecs_to_timeval(0), (0, 0))
- WVPASSEQ(xstat.nsecs_to_timeval(10**9), (1, 0))
- WVPASSEQ(xstat.nsecs_to_timeval(500000000), (0, (10**9 / 2) / 1000))
- WVPASSEQ(xstat.nsecs_to_timeval(1500000000), (1, (10**9 / 2) / 1000))
- WVPASSEQ(xstat.nsecs_to_timeval(-10**9), (-1, 0))
- WVPASSEQ(xstat.nsecs_to_timeval(-500000000), (-1, (10**9 / 2) / 1000))
- WVPASSEQ(xstat.nsecs_to_timeval(-1500000000), (-2, (10**9 / 2) / 1000))
- x = xstat.nsecs_to_timeval(1977777778)
- WVPASSEQ(type(x[0]), type(0))
- WVPASSEQ(type(x[1]), type(0))
- x = xstat.nsecs_to_timeval(-1977777778)
- WVPASSEQ(type(x[0]), type(0))
- WVPASSEQ(type(x[1]), type(0))
+ WVPASSEQ(xstat.nsecs_to_timeval(0), (0, 0))
+ WVPASSEQ(xstat.nsecs_to_timeval(10**9), (1, 0))
+ WVPASSEQ(xstat.nsecs_to_timeval(500000000), (0, (10**9 / 2) / 1000))
+ WVPASSEQ(xstat.nsecs_to_timeval(1500000000), (1, (10**9 / 2) / 1000))
+ WVPASSEQ(xstat.nsecs_to_timeval(-10**9), (-1, 0))
+ WVPASSEQ(xstat.nsecs_to_timeval(-500000000), (-1, (10**9 / 2) / 1000))
+ WVPASSEQ(xstat.nsecs_to_timeval(-1500000000), (-2, (10**9 / 2) / 1000))
+ x = xstat.nsecs_to_timeval(1977777778)
+ WVPASSEQ(type(x[0]), type(0))
+ WVPASSEQ(type(x[1]), type(0))
+ x = xstat.nsecs_to_timeval(-1977777778)
+ WVPASSEQ(type(x[0]), type(0))
+ WVPASSEQ(type(x[1]), type(0))
- WVPASSEQ(xstat.fstime_floor_secs(0), 0)
- WVPASSEQ(xstat.fstime_floor_secs(10**9 / 2), 0)
- WVPASSEQ(xstat.fstime_floor_secs(10**9), 1)
- WVPASSEQ(xstat.fstime_floor_secs(-10**9 / 2), -1)
- WVPASSEQ(xstat.fstime_floor_secs(-10**9), -1)
- WVPASSEQ(type(xstat.fstime_floor_secs(10**9 / 2)), type(0))
- WVPASSEQ(type(xstat.fstime_floor_secs(-10**9 / 2)), type(0))
+ WVPASSEQ(xstat.fstime_floor_secs(0), 0)
+ WVPASSEQ(xstat.fstime_floor_secs(10**9 / 2), 0)
+ WVPASSEQ(xstat.fstime_floor_secs(10**9), 1)
+ WVPASSEQ(xstat.fstime_floor_secs(-10**9 / 2), -1)
+ WVPASSEQ(xstat.fstime_floor_secs(-10**9), -1)
+ WVPASSEQ(type(xstat.fstime_floor_secs(10**9 / 2)), type(0))
+ WVPASSEQ(type(xstat.fstime_floor_secs(-10**9 / 2)), type(0))
@wvtest
def test_bup_utimensat():
- initial_failures = wvfailure_count()
if not xstat._bup_utimensat:
return
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-txstat-')
- path = tmpdir + '/foo'
- open(path, 'w').close()
- frac_ts = (0, 10**9 / 2)
- xstat._bup_utimensat(_helpers.AT_FDCWD, path, (frac_ts, frac_ts), 0)
- st = _helpers.stat(path)
- atime_ts = st[8]
- mtime_ts = st[9]
- WVPASSEQ(atime_ts[0], 0)
- WVPASS(atime_ts[1] == 0 or atime_ts[1] == frac_ts[1])
- WVPASSEQ(mtime_ts[0], 0)
- WVPASS(mtime_ts[1] == 0 or mtime_ts[1] == frac_ts[1])
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-txstat-') as tmpdir:
+ path = tmpdir + '/foo'
+ open(path, 'w').close()
+ frac_ts = (0, 10**9 / 2)
+ xstat._bup_utimensat(_helpers.AT_FDCWD, path, (frac_ts, frac_ts), 0)
+ st = _helpers.stat(path)
+ atime_ts = st[8]
+ mtime_ts = st[9]
+ WVPASSEQ(atime_ts[0], 0)
+ WVPASS(atime_ts[1] == 0 or atime_ts[1] == frac_ts[1])
+ WVPASSEQ(mtime_ts[0], 0)
+ WVPASS(mtime_ts[1] == 0 or mtime_ts[1] == frac_ts[1])
@wvtest
def test_bup_utimes():
- initial_failures = wvfailure_count()
if not xstat._bup_utimes:
return
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-txstat-')
- path = tmpdir + '/foo'
- open(path, 'w').close()
- frac_ts = (0, 10**6 / 2)
- xstat._bup_utimes(path, (frac_ts, frac_ts))
- st = _helpers.stat(path)
- atime_ts = st[8]
- mtime_ts = st[9]
- WVPASSEQ(atime_ts[0], 0)
- WVPASS(atime_ts[1] == 0 or atime_ts[1] == frac_ts[1] * 1000)
- WVPASSEQ(mtime_ts[0], 0)
- WVPASS(mtime_ts[1] == 0 or mtime_ts[1] == frac_ts[1] * 1000)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-txstat-') as tmpdir:
+ path = tmpdir + '/foo'
+ open(path, 'w').close()
+ frac_ts = (0, 10**6 / 2)
+ xstat._bup_utimes(path, (frac_ts, frac_ts))
+ st = _helpers.stat(path)
+ atime_ts = st[8]
+ mtime_ts = st[9]
+ WVPASSEQ(atime_ts[0], 0)
+ WVPASS(atime_ts[1] == 0 or atime_ts[1] == frac_ts[1] * 1000)
+ WVPASSEQ(mtime_ts[0], 0)
+ WVPASS(mtime_ts[1] == 0 or mtime_ts[1] == frac_ts[1] * 1000)
@wvtest
def test_bup_lutimes():
- initial_failures = wvfailure_count()
if not xstat._bup_lutimes:
return
- tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-txstat-')
- path = tmpdir + '/foo'
- open(path, 'w').close()
- frac_ts = (0, 10**6 / 2)
- xstat._bup_lutimes(path, (frac_ts, frac_ts))
- st = _helpers.stat(path)
- atime_ts = st[8]
- mtime_ts = st[9]
- WVPASSEQ(atime_ts[0], 0)
- WVPASS(atime_ts[1] == 0 or atime_ts[1] == frac_ts[1] * 1000)
- WVPASSEQ(mtime_ts[0], 0)
- WVPASS(mtime_ts[1] == 0 or mtime_ts[1] == frac_ts[1] * 1000)
- if wvfailure_count() == initial_failures:
- subprocess.call(['rm', '-rf', tmpdir])
+ with no_lingering_errors(), test_tempdir('bup-txstat-') as tmpdir:
+ path = tmpdir + '/foo'
+ open(path, 'w').close()
+ frac_ts = (0, 10**6 / 2)
+ xstat._bup_lutimes(path, (frac_ts, frac_ts))
+ st = _helpers.stat(path)
+ atime_ts = st[8]
+ mtime_ts = st[9]
+ WVPASSEQ(atime_ts[0], 0)
+ WVPASS(atime_ts[1] == 0 or atime_ts[1] == frac_ts[1] * 1000)
+ WVPASSEQ(mtime_ts[0], 0)
+ WVPASS(mtime_ts[1] == 0 or mtime_ts[1] == frac_ts[1] * 1000)
projects / bup.git / commitdiff