Python 2.6 only supports a single with clause.
Signed-off-by: Rob Browning <rlb@defaultvalue.org>
Tested-by: Rob Browning <rlb@defaultvalue.org>
@wvtest
def test_bloom():
- 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')
+ with no_lingering_errors():
+ with 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)
+ 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)
+ # 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)
@wvtest
def test_server_split_with_indexes():
- 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()
+ with no_lingering_errors():
+ with 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():
- 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)
+ with no_lingering_errors():
+ with 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():
- 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)
+ with no_lingering_errors():
+ with 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():
- 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)
+ with no_lingering_errors():
+ with 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
@wvtest
def testpacks():
- 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()
-
- 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))
+ with no_lingering_errors():
+ with 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()
+
+ 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))
@wvtest
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')
+ with no_lingering_errors():
+ with 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 = []
+ 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'))
+ 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)
+ 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_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)
- 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)
+ with no_lingering_errors():
+ with test_tempdir('bup-tgit-') as tmpdir:
+ 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))
+ 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)
@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:
- 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()
+ with no_lingering_errors():
+ with test_tempdir('bup-tgit-') as tmpdir:
+ os.environ['BUP_DIR'] = bupdir = tmpdir + "/bup"
+ orig_cwd = os.getcwd()
try:
+ os.chdir(tmpdir)
+ git.init_repo(bupdir)
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)
+ # if we reach this point the call above passed
+ WVPASS('check_repo_or_die')
+
+ 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
return readpipe(['git', 'show', '-s',
'--pretty=format:%s' % val, commit]).strip()
- 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)
+ with no_lingering_errors():
+ with 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():
- 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')
- adate_tz_sec = -60 * 60
- cdate_tz_sec = 120 * 60
- commit_off = w.new_commit(tree, parent,
+ with no_lingering_errors():
+ with 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)
+ 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():
- 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)
+ with no_lingering_errors():
+ with 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():
@wvtest
def test_atomically_replaced_file():
- 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')
+ with no_lingering_errors():
+ with test_tempdir('bup-thelper-') as tmpdir:
+ target_file = os.path.join(tmpdir, 'test-atomic-write')
- 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')
+
+ with atomically_replaced_file(target_file, mode='wb') as f:
+ f.write(os.urandom(20))
+ WVPASSEQ(f.mode, 'wb')
@wvtest
@wvtest
def index_writer():
- 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)
+ with no_lingering_errors():
+ with 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():
- 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()
-
- # 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()
-
-
-@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()
+ with no_lingering_errors():
+ with 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()
- w3 = index.Writer('index3.tmp', ms3, tmax)
- w3.add('/a/c/n/3', fs, meta_ofs3)
- #w3.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()
- 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)
+@wvtest
+def index_dirty():
+ with no_lingering_errors():
+ with 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)
@wvtest
def test_metadata_method():
- 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)
+ with no_lingering_errors():
+ with 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():
def test_from_path_error():
if is_superuser() or detect_fakeroot():
return
- 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()
+ with no_lingering_errors():
+ with 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):
return
if sys.platform.startswith('cygwin'):
return # chmod 000 isn't effective.
- 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()
+ with no_lingering_errors():
+ with 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():
- 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)
+ with no_lingering_errors():
+ with 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
def test_bup_utimensat():
if not xstat._bup_utimensat:
return
- 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])
+ with no_lingering_errors():
+ with 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():
if not xstat._bup_utimes:
return
- 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)
+ with no_lingering_errors():
+ with 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():
if not xstat._bup_lutimes:
return
- 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)
+ with no_lingering_errors():
+ with 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