success/failure counts, but the final counts displayed should be
correct.
-Individual non-Python tests can be run via "./wvtest run test/TEST" and
-if you'd like to see all of the test output, you can omit the wvtest
-run wrapper: "test/TEST"
+Individual non-Python tests can be run via
-Individual Python tests can be run via "./wvtest run ./wvtest.py
-test/int/TEST", and as above, you can see all the output by omitting
-the wvtest run wrapper like this: "./wvtest.py test/TEST"
+ ./wvtest run test/ext/TEST
+
+and if you'd like to see all of the test output, you can omit the
+wvtest run wrapper: `test/ext/TEST`. Individual Python tests can be
+run via
+
+ ./pytest test/int/test_something.py
Internal tests that test bup's code directly are located in test/int,
and external tests that test bup from the outside, typically by
runtests: runtests-python runtests-cmdline
-python_tests := \
- test/int/test_bloom.py \
- test/int/test_client.py \
- test/int/test_compat.py \
- test/int/test_git.py \
- test/int/test_hashsplit.py \
- test/int/test_helpers.py \
- test/int/test_index.py \
- test/int/test_metadata.py \
- test/int/test_options.py \
- test/int/test_resolve.py \
- test/int/test_shquote.py \
- test/int/test_vfs.py \
- test/int/test_vint.py \
- test/int/test_xstat.py
-
-
-# The "pwd -P" here may not be appropriate in the long run, but we
-# need it until we settle the relevant drecurse/exclusion questions:
-# https://groups.google.com/forum/#!topic/bup-list/9ke-Mbp10Q0
runtests-python: all test/tmp
- mkdir -p test/tmp/test-log
- $(pf); cd $$(pwd -P); TMPDIR="$(test_tmp)" \
- ./wvtest.py $(python_tests) 2>&1 \
- | tee -a test/tmp/test-log/$$$$.log
+ ./pytest
cmdline_tests := \
test/ext/test.sh \
test: all
if test -e test/tmp/test-log; then rm -r test/tmp/test-log; fi
mkdir -p test/tmp/test-log
+ $(MAKE) runtests-python
./wvtest watch --no-counts \
- $(MAKE) runtests 2>test/tmp/test-log/$$$$.log
+ $(MAKE) runtests-cmdline 2>test/tmp/test-log/$$$$.log
./wvtest report test/tmp/test-log/*.log
check: test
from __future__ import absolute_import
-from os.path import dirname, realpath
+from os.path import basename, dirname, realpath
from time import tzset
+from traceback import extract_stack
import os
import pytest
+import subprocess
import sys
sys.path[:0] = ['lib']
# https://groups.google.com/forum/#!topic/bup-list/9ke-Mbp10Q0
os.chdir(realpath(os.getcwd()))
+@pytest.fixture(autouse=True)
+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)
+ assert False, '%s:%-4d %s' % (basename(src_file),
+ src_line, msg)
+
+ fail_if_errors()
+ helpers.clear_errors()
+ yield None
+ fail_if_errors()
+ helpers.clear_errors()
+
@pytest.fixture(autouse=True)
def ephemeral_env_changes():
orig_env = environ.copy()
if k == b'TZ':
tzset()
os.chdir(_bup_src_top)
+
+@pytest.fixture()
+def tmpdir(tmp_path):
+ try:
+ yield bytes(tmp_path)
+ finally:
+ subprocess.call([b'chmod', b'-R', b'u+rwX', bytes(tmp_path)])
+ # FIXME: delete only if there are no errors
+ #subprocess.call(['rm', '-rf', tmpdir])
apt-get install -y \
$common_debs \
python2.7-dev python-fuse \
- python-"$xattr" python-tornado
+ python-"$xattr" python-tornado python-pytest
;;
python3)
apt-get install -y \
$common_debs \
python3-dev python3-distutils python3-fuse \
- python3-"$xattr" python3-tornado
+ python3-"$xattr" python3-tornado python3-pytest
;;
*)
usage 1>&2
case "$pyver" in
python2)
- pkg install $common_pkgs python2 py27-tornado
+ pkg install $common_pkgs python2 py27-tornado pytest
;;
python3)
- pkg install $common_pkgs python3 py37-tornado
+ pkg install $common_pkgs python3 py37-tornado pytest3
;;
*)
usage 1>&2
# "brew unlink readline" will undo this hack
case "$pyver" in
- python2) ;;
- python3) brew install python ;;
+ python2)
+ easy_install-2.7 --user pip
+ /Users/anka/Library/Python/2.7/bin/pip install --user pytest
+ ;;
+ python3)
+ brew install python
+ easy_install --user pip
+ pip3 install --user pytest
+ ;;
*)
usage 1>&2
exit 2
--- /dev/null
+#!/bin/sh
+
+set -eu
+
+script_home="$(cd "$(dirname "$0")" && pwd -P)"
+testlibdir="$script_home/test/lib"
+
+export PYTHONPATH="$testlibdir${PYTHONPATH:+:$PYTHONPATH}"
+exec dev/bup-python -m pytest -v "$@"
--- /dev/null
+[pytest]
+testpaths = test/int test/ext
--- /dev/null
+../../dev/sort-z
\ No newline at end of file
+++ /dev/null
-../../../dev/sort-z
\ No newline at end of file
root_status="$(dev/root-status)" || exit $?
TOP="$(WVPASS pwd)" || exit $?
-export PATH="$TOP/test/ext/bin:$PATH"
+export PATH="$TOP/test/bin:$PATH"
tmpdir="$(WVPASS wvmktempdir)" || exit $?
export BUP_DIR="$tmpdir/bup"
-
-from __future__ import absolute_import
-import sys
-
-sys.path[:0] = ['../..']
from __future__ import absolute_import, print_function
+import os
import errno, platform, tempfile
-
-from wvtest import *
+import logging
from bup import bloom
from bup.helpers import mkdirp
-from buptest import no_lingering_errors, test_tempdir
-
-@wvtest
-def test_bloom():
- with no_lingering_errors():
- with test_tempdir(b'bup-tbloom-') as tmpdir:
- hashes = [os.urandom(20) for i in range(100)]
- class Idx:
- pass
- ix = Idx()
- ix.name = b'dummy.idx'
- ix.shatable = b''.join(hashes)
- for k in (4, 5):
- b = bloom.create(tmpdir + b'/pybuptest.bloom', expected=100, k=k)
- b.add_idx(ix)
- WVPASSLT(b.pfalse_positive(), .1)
- b.close()
- b = bloom.ShaBloom(tmpdir + b'/pybuptest.bloom')
- all_present = True
- for h in hashes:
- all_present &= (b.exists(h) or False)
- 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 + b'/pybuptest.bloom')
+def test_bloom(tmpdir):
+ hashes = [os.urandom(20) for i in range(100)]
+ class Idx:
+ pass
+ ix = Idx()
+ ix.name = b'dummy.idx'
+ ix.shatable = b''.join(hashes)
+ for k in (4, 5):
+ b = bloom.create(tmpdir + b'/pybuptest.bloom', expected=100, k=k)
+ b.add_idx(ix)
+ assert b.pfalse_positive() < .1
+ b.close()
+ b = bloom.ShaBloom(tmpdir + b'/pybuptest.bloom')
+ all_present = True
+ for h in hashes:
+ all_present &= (b.exists(h) or False)
+ assert all_present
+ false_positives = 0
+ for h in [os.urandom(20) for i in range(1000)]:
+ if b.exists(h):
+ false_positives += 1
+ assert false_positives < 5
+ os.unlink(tmpdir + b'/pybuptest.bloom')
- tf = tempfile.TemporaryFile(dir=tmpdir)
- b = bloom.create(b'bup.bloom', f=tf, expected=100)
- WVPASSEQ(b.rwfile, tf)
- WVPASSEQ(b.k, 5)
+ tf = tempfile.TemporaryFile(dir=tmpdir)
+ b = bloom.create(b'bup.bloom', f=tf, expected=100)
+ assert b.rwfile == tf
+ assert 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(b'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(b'bup.bloom', f=tf, expected=2**28,
+ delaywrite=False)
+ assert b.k == 4
+ except EnvironmentError as ex:
+ (ptr_width, linkage) = platform.architecture()
+ if ptr_width == '32bit' and ex.errno == errno.ENOMEM:
+ logging.getLogger().info('skipping large bloom filter test (mmap probably failed) '
+ + str(ex))
+ else:
+ raise
from __future__ import absolute_import
import sys, os, stat, time, random, subprocess, glob
-
-from wvtest import *
+import pytest
from bup import client, git, path
from bup.compat import bytes_from_uint, environ, range
from bup.helpers import mkdirp
-from buptest import no_lingering_errors, test_tempdir
-
def randbytes(sz):
s = b''
s3 = randbytes(10000)
IDX_PAT = b'/*.idx'
-
-
-@wvtest
-def test_server_split_with_indexes():
- with no_lingering_errors():
- with test_tempdir(b'bup-tclient-') as tmpdir:
- environ[b'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():
- with test_tempdir(b'bup-tclient-') as tmpdir:
- environ[b'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(b'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. First, flush to ensure that we've
- # actually sent all the command ('receive-objects-v2')
- # and their data to the server. This may be needed if
- # the output buffer size is bigger than the data (both
- # command and objects) we're writing. To see the need
- # for this, change the object sizes at the beginning
- # of this file to be very small (e.g. 10 instead of 10k)
- c.conn.outp.flush()
-
- # Then, check if we've already received the idx files.
- # This may happen if we're preempted just after writing
- # the data, then the server runs and suggests, and only
- # then we continue in PackWriter_Remote::_raw_write()
- # and check the has_input(), in that case we'll receive
- # the idx still in the rw.new_blob() calls above.
- #
- # In most cases though, that doesn't happen, and we'll
- # get past the has_input() check before the server has
- # a chance to respond - it has to actually hash the new
- # object here, so it takes some time. So also break out
- # of the loop if the server has sent something on the
- # connection.
- #
- # Finally, abort this after a little while (about one
- # second) just in case something's actually broken.
- n = 0
- while (len(glob.glob(c.cachedir+IDX_PAT)) < 2 and
- not c.conn.has_input() and n < 10):
- time.sleep(0.1)
- n += 1
- WVPASS(len(glob.glob(c.cachedir+IDX_PAT)) == 2 or c.conn.has_input())
- 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():
- with test_tempdir(b'bup-tclient-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir
- git.init_repo(bupdir)
- open(git.repo(b'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():
- with test_tempdir(b'bup-tclient-') as tmpdir:
- environ[b'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 + b'/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([path.exe(), b'midx', b'-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_server_split_with_indexes(tmpdir):
+ environ[b'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()
+
+
+def test_multiple_suggestions(tmpdir):
+ environ[b'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()
+ assert len(glob.glob(git.repo(b'objects/pack'+IDX_PAT))) == 2
+
+ c = client.Client(bupdir, create=True)
+ assert len(glob.glob(c.cachedir+IDX_PAT)) == 0
+ rw = c.new_packwriter()
+ s1sha = rw.new_blob(s1)
+ assert rw.exists(s1sha)
+ s2sha = rw.new_blob(s2)
+
+ # This is a little hacky, but ensures that we test the
+ # code under test. First, flush to ensure that we've
+ # actually sent all the command ('receive-objects-v2')
+ # and their data to the server. This may be needed if
+ # the output buffer size is bigger than the data (both
+ # command and objects) we're writing. To see the need
+ # for this, change the object sizes at the beginning
+ # of this file to be very small (e.g. 10 instead of 10k)
+ c.conn.outp.flush()
+
+ # Then, check if we've already received the idx files.
+ # This may happen if we're preempted just after writing
+ # the data, then the server runs and suggests, and only
+ # then we continue in PackWriter_Remote::_raw_write()
+ # and check the has_input(), in that case we'll receive
+ # the idx still in the rw.new_blob() calls above.
+ #
+ # In most cases though, that doesn't happen, and we'll
+ # get past the has_input() check before the server has
+ # a chance to respond - it has to actually hash the new
+ # object here, so it takes some time. So also break out
+ # of the loop if the server has sent something on the
+ # connection.
+ #
+ # Finally, abort this after a little while (about one
+ # second) just in case something's actually broken.
+ n = 0
+ while (len(glob.glob(c.cachedir+IDX_PAT)) < 2 and
+ not c.conn.has_input() and n < 10):
+ time.sleep(0.1)
+ n += 1
+ assert len(glob.glob(c.cachedir+IDX_PAT)) == 2 or c.conn.has_input()
+ rw.new_blob(s2)
+ assert rw.objcache.exists(s1sha)
+ assert rw.objcache.exists(s2sha)
+ rw.new_blob(s3)
+ assert len(glob.glob(c.cachedir+IDX_PAT)) == 2
+ rw.close()
+ assert len(glob.glob(c.cachedir+IDX_PAT)) == 3
+
+
+def test_dumb_client_server(tmpdir):
+ environ[b'BUP_DIR'] = bupdir = tmpdir
+ git.init_repo(bupdir)
+ open(git.repo(b'bup-dumb-server'), 'w').close()
+
+ lw = git.PackWriter()
+ lw.new_blob(s1)
+ lw.close()
+
+ c = client.Client(bupdir, create=True)
+ rw = c.new_packwriter()
+ assert len(glob.glob(c.cachedir+IDX_PAT)) == 1
+ rw.new_blob(s1)
+ assert len(glob.glob(c.cachedir+IDX_PAT)) == 1
+ rw.new_blob(s2)
+ rw.close()
+ assert len(glob.glob(c.cachedir+IDX_PAT)) == 2
+
+
+def test_midx_refreshing(tmpdir):
+ environ[b'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 + b'/objects/pack')
+ assert len(pi.packs) == 2
+ pi.refresh()
+ assert len(pi.packs) == 2
+ assert sorted([os.path.basename(i.name) for i in pi.packs]) == sorted([p1base, p2base])
+
+ p1 = git.open_idx(p1name)
+ assert p1.exists(s1sha)
+ p2 = git.open_idx(p2name)
+ assert not p2.exists(s1sha)
+ assert p2.exists(s2sha)
+
+ subprocess.call([path.exe(), b'midx', b'-f'])
+ pi.refresh()
+ assert len(pi.packs) == 1
+ pi.refresh(skip_midx=True)
+ assert len(pi.packs) == 2
+ pi.refresh(skip_midx=False)
+ assert len(pi.packs) == 1
+
+
def test_remote_parsing():
- with no_lingering_errors():
- tests = (
- (b':/bup', (b'file', None, None, b'/bup')),
- (b'file:///bup', (b'file', None, None, b'/bup')),
- (b'192.168.1.1:/bup', (b'ssh', b'192.168.1.1', None, b'/bup')),
- (b'ssh://192.168.1.1:2222/bup', (b'ssh', b'192.168.1.1', b'2222', b'/bup')),
- (b'ssh://[ff:fe::1]:2222/bup', (b'ssh', b'ff:fe::1', b'2222', b'/bup')),
- (b'bup://foo.com:1950', (b'bup', b'foo.com', b'1950', None)),
- (b'bup://foo.com:1950/bup', (b'bup', b'foo.com', b'1950', b'/bup')),
- (b'bup://[ff:fe::1]/bup', (b'bup', b'ff:fe::1', None, b'/bup')),)
- for remote, values in tests:
- WVPASSEQ(client.parse_remote(remote), values)
- try:
- client.parse_remote(b'http://asdf.com/bup')
- WVFAIL()
- except client.ClientError:
- WVPASS()
+ tests = (
+ (b':/bup', (b'file', None, None, b'/bup')),
+ (b'file:///bup', (b'file', None, None, b'/bup')),
+ (b'192.168.1.1:/bup', (b'ssh', b'192.168.1.1', None, b'/bup')),
+ (b'ssh://192.168.1.1:2222/bup', (b'ssh', b'192.168.1.1', b'2222', b'/bup')),
+ (b'ssh://[ff:fe::1]:2222/bup', (b'ssh', b'ff:fe::1', b'2222', b'/bup')),
+ (b'bup://foo.com:1950', (b'bup', b'foo.com', b'1950', None)),
+ (b'bup://foo.com:1950/bup', (b'bup', b'foo.com', b'1950', b'/bup')),
+ (b'bup://[ff:fe::1]/bup', (b'bup', b'ff:fe::1', None, b'/bup')),)
+ for remote, values in tests:
+ assert client.parse_remote(remote) == values
+
+ with pytest.raises(client.ClientError):
+ client.parse_remote(b'http://asdf.com/bup')
from __future__ import absolute_import, print_function
+import sys
from binascii import hexlify, unhexlify
from subprocess import check_call
import struct, os, time
+import pytest
-from wvtest import *
+from wvpytest import *
from bup import git, path
from bup.compat import bytes_from_byte, environ, range
from bup.helpers import localtime, log, mkdirp, readpipe
-from buptest import no_lingering_errors, test_tempdir
bup_exe = path.exe()
return readpipe(cmd)
-@wvtest
def test_git_version_detection():
- with no_lingering_errors():
- # Test version types from git's tag history
- for expected, ver in \
- (('insufficient', b'git version 0.99'),
- ('insufficient', b'git version 0.99.1'),
- ('insufficient', b'git version 0.99.7a'),
- ('insufficient', b'git version 1.0rc1'),
- ('insufficient', b'git version 1.0.1'),
- ('insufficient', b'git version 1.4.2.1'),
- ('insufficient', b'git version 1.5.5'),
- ('insufficient', b'git version 1.5.6-rc0'),
- ('suitable', b'git version 1.5.6'),
- ('suitable', b'git version 1.5.6.1'),
- ('suitable', b'git version 2.14.0-rc0'),
- ('suitable', b'git version 2.14.0 (something ...)'),
- ('suitable', b'git version 111.222.333.444-rc555'),
- ('unrecognized', b'huh?')):
- WVMSG('Checking version validation: %r' % ver)
- WVPASSEQ(expected, git.is_suitable_git(ver_str=ver))
- try:
- if expected == 'insufficient':
- WVEXCEPT(SystemExit, git.require_suitable_git, ver)
- elif expected == 'suitable':
- git.require_suitable_git(ver_str=ver)
- elif expected == 'unrecognized':
- WVEXCEPT(git.GitError, git.require_suitable_git, ver)
- else:
- WVPASS(False)
- finally:
- git._git_great = None
- try:
- environ[b'BUP_GIT_VERSION_IS_FINE'] = b'true'
+ # Test version types from git's tag history
+ for expected, ver in \
+ (('insufficient', b'git version 0.99'),
+ ('insufficient', b'git version 0.99.1'),
+ ('insufficient', b'git version 0.99.7a'),
+ ('insufficient', b'git version 1.0rc1'),
+ ('insufficient', b'git version 1.0.1'),
+ ('insufficient', b'git version 1.4.2.1'),
+ ('insufficient', b'git version 1.5.5'),
+ ('insufficient', b'git version 1.5.6-rc0'),
+ ('suitable', b'git version 1.5.6'),
+ ('suitable', b'git version 1.5.6.1'),
+ ('suitable', b'git version 2.14.0-rc0'),
+ ('suitable', b'git version 2.14.0 (something ...)'),
+ ('suitable', b'git version 111.222.333.444-rc555'),
+ ('unrecognized', b'huh?')):
+ assert expected == git.is_suitable_git(ver_str=ver)
+ try:
+ if expected == 'insufficient':
+ with pytest.raises(SystemExit):
+ git.require_suitable_git(ver)
+ elif expected == 'suitable':
git.require_suitable_git(ver_str=ver)
- finally:
- del environ[b'BUP_GIT_VERSION_IS_FINE']
- git._git_great = None
-
-
-@wvtest
-def testmangle():
- with no_lingering_errors():
- afile = 0o100644
- afile2 = 0o100770
- alink = 0o120000
- adir = 0o040000
- adir2 = 0o040777
- WVPASSEQ(git.mangle_name(b'a', adir2, adir), b'a')
- WVPASSEQ(git.mangle_name(b'.bup', adir2, adir), b'.bup.bupl')
- WVPASSEQ(git.mangle_name(b'a.bupa', adir2, adir), b'a.bupa.bupl')
- WVPASSEQ(git.mangle_name(b'b.bup', alink, alink), b'b.bup.bupl')
- WVPASSEQ(git.mangle_name(b'b.bu', alink, alink), b'b.bu')
- WVPASSEQ(git.mangle_name(b'f', afile, afile2), b'f')
- WVPASSEQ(git.mangle_name(b'f.bup', afile, afile2), b'f.bup.bupl')
- WVPASSEQ(git.mangle_name(b'f.bup', afile, adir), b'f.bup.bup')
- WVPASSEQ(git.mangle_name(b'f', afile, adir), b'f.bup')
-
- WVPASSEQ(git.demangle_name(b'f.bup', afile), (b'f', git.BUP_CHUNKED))
- WVPASSEQ(git.demangle_name(b'f.bupl', afile), (b'f', git.BUP_NORMAL))
- WVPASSEQ(git.demangle_name(b'f.bup.bupl', afile), (b'f.bup', git.BUP_NORMAL))
-
- WVPASSEQ(git.demangle_name(b'.bupm', afile), (b'', git.BUP_NORMAL))
- WVPASSEQ(git.demangle_name(b'.bupm', adir), (b'', 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(b'f.bupa', afile), (b'f.bupa', git.BUP_NORMAL))
-
-
-@wvtest
-def testencode():
- with no_lingering_errors():
- s = b'hello world'
- looseb = b''.join(git._encode_looseobj(b'blob', s))
- looset = b''.join(git._encode_looseobj(b'tree', s))
- loosec = b''.join(git._encode_looseobj(b'commit', s))
- packb = b''.join(git._encode_packobj(b'blob', s))
- packt = b''.join(git._encode_packobj(b'tree', s))
- packc = b''.join(git._encode_packobj(b'commit', s))
- packlb = b''.join(git._encode_packobj(b'blob', s * 200))
- WVPASSEQ(git._decode_looseobj(looseb), (b'blob', s))
- WVPASSEQ(git._decode_looseobj(looset), (b'tree', s))
- WVPASSEQ(git._decode_looseobj(loosec), (b'commit', s))
- WVPASSEQ(git._decode_packobj(packb), (b'blob', s))
- WVPASSEQ(git._decode_packobj(packt), (b'tree', s))
- WVPASSEQ(git._decode_packobj(packc), (b'commit', s))
- WVPASSEQ(git._decode_packobj(packlb), (b'blob', s * 200))
- for i in range(10):
- WVPASS(git._encode_looseobj(b'blob', s, compression_level=i))
- def encode_pobj(n):
- return b''.join(git._encode_packobj(b'blob', s, compression_level=n))
- WVEXCEPT(ValueError, encode_pobj, -1)
- WVEXCEPT(ValueError, encode_pobj, 10)
- WVEXCEPT(ValueError, encode_pobj, b'x')
-
-
-@wvtest
-def testpacks():
- with no_lingering_errors():
- with test_tempdir(b'bup-tgit-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir + b'/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(b'%d' % i))
- log('\n')
- nameprefix = w.close()
- print(repr(nameprefix))
- WVPASS(os.path.exists(nameprefix + b'.pack'))
- WVPASS(os.path.exists(nameprefix + b'.idx'))
-
- r = git.open_idx(nameprefix + b'.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(b'\0'*20))
-
- pi = iter(r)
- for h in sorted(hashes):
- WVPASSEQ(hexlify(next(pi)), hexlify(h))
-
- WVFAIL(r.find_offset(b'\0'*20))
-
- r = git.PackIdxList(bupdir + b'/objects/pack')
- WVPASS(r.exists(hashes[5]))
- WVPASS(r.exists(hashes[6]))
- WVFAIL(r.exists(b'\0'*20))
-
-
-@wvtest
-def test_pack_name_lookup():
- with no_lingering_errors():
- with test_tempdir(b'bup-tgit-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
- git.init_repo(bupdir)
- git.verbose = 1
- packdir = git.repo(b'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(b'%d' % i))
- log('\n')
- idxnames.append(os.path.basename(w.close() + b'.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(idxname, r.exists(hashes[i], want_source=True))
-
-
-@wvtest
-def test_long_index():
- with no_lingering_errors():
- with test_tempdir(b'bup-tgit-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
- git.init_repo(bupdir)
- idx = git.PackIdxV2Writer()
- 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.add(obj_bin, 1, 0xfffffffff)
- idx.add(obj2_bin, 2, 0xffffffffff)
- idx.add(obj3_bin, 3, 0xff)
- name = tmpdir + b'/tmp.idx'
- r = idx.write(name, 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():
- with test_tempdir(b'bup-tgit-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
- orig_cwd = os.getcwd()
- try:
- os.chdir(tmpdir)
- git.init_repo(bupdir)
- git.check_repo_or_die()
- # if we reach this point the call above passed
- WVPASS('check_repo_or_die')
-
- os.rename(bupdir + b'/objects/pack',
- bupdir + b'/objects/pack.tmp')
- open(bupdir + b'/objects/pack', 'w').close()
- try:
- git.check_repo_or_die()
- except SystemExit as e:
- WVPASSEQ(e.code, 14)
- else:
- WVFAIL()
- os.unlink(bupdir + b'/objects/pack')
- os.rename(bupdir + b'/objects/pack.tmp',
- bupdir + b'/objects/pack')
-
- try:
- git.check_repo_or_die(b'nonexistantbup.tmp')
- except SystemExit as e:
- WVPASSEQ(e.code, 15)
- else:
- WVFAIL()
- finally:
- os.chdir(orig_cwd)
-
-
-@wvtest
-def test_commit_parsing():
+ elif expected == 'unrecognized':
+ with pytest.raises(git.GitError):
+ git.require_suitable_git(ver)
+ else:
+ assert False
+ finally:
+ git._git_great = None
+ try:
+ environ[b'BUP_GIT_VERSION_IS_FINE'] = b'true'
+ git.require_suitable_git(ver_str=ver)
+ finally:
+ del environ[b'BUP_GIT_VERSION_IS_FINE']
+ git._git_great = None
+
+
+def test_mangle():
+ afile = 0o100644
+ afile2 = 0o100770
+ alink = 0o120000
+ adir = 0o040000
+ adir2 = 0o040777
+ assert git.mangle_name(b'a', adir2, adir) == b'a'
+ assert git.mangle_name(b'.bup', adir2, adir) == b'.bup.bupl'
+ assert git.mangle_name(b'a.bupa', adir2, adir) == b'a.bupa.bupl'
+ WVPASSEQ(git.mangle_name(b'b.bup', alink, alink), b'b.bup.bupl')
+ WVPASSEQ(git.mangle_name(b'b.bu', alink, alink), b'b.bu')
+ WVPASSEQ(git.mangle_name(b'f', afile, afile2), b'f')
+ WVPASSEQ(git.mangle_name(b'f.bup', afile, afile2), b'f.bup.bupl')
+ WVPASSEQ(git.mangle_name(b'f.bup', afile, adir), b'f.bup.bup')
+ WVPASSEQ(git.mangle_name(b'f', afile, adir), b'f.bup')
+
+ WVPASSEQ(git.demangle_name(b'f.bup', afile), (b'f', git.BUP_CHUNKED))
+ WVPASSEQ(git.demangle_name(b'f.bupl', afile), (b'f', git.BUP_NORMAL))
+ WVPASSEQ(git.demangle_name(b'f.bup.bupl', afile), (b'f.bup', git.BUP_NORMAL))
+
+ WVPASSEQ(git.demangle_name(b'.bupm', afile), (b'', git.BUP_NORMAL))
+ WVPASSEQ(git.demangle_name(b'.bupm', adir), (b'', 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(b'f.bupa', afile), (b'f.bupa', git.BUP_NORMAL))
+
+
+def test_encode():
+ s = b'hello world'
+ looseb = b''.join(git._encode_looseobj(b'blob', s))
+ looset = b''.join(git._encode_looseobj(b'tree', s))
+ loosec = b''.join(git._encode_looseobj(b'commit', s))
+ packb = b''.join(git._encode_packobj(b'blob', s))
+ packt = b''.join(git._encode_packobj(b'tree', s))
+ packc = b''.join(git._encode_packobj(b'commit', s))
+ packlb = b''.join(git._encode_packobj(b'blob', s * 200))
+ WVPASSEQ(git._decode_looseobj(looseb), (b'blob', s))
+ WVPASSEQ(git._decode_looseobj(looset), (b'tree', s))
+ WVPASSEQ(git._decode_looseobj(loosec), (b'commit', s))
+ WVPASSEQ(git._decode_packobj(packb), (b'blob', s))
+ WVPASSEQ(git._decode_packobj(packt), (b'tree', s))
+ WVPASSEQ(git._decode_packobj(packc), (b'commit', s))
+ WVPASSEQ(git._decode_packobj(packlb), (b'blob', s * 200))
+ for i in range(10):
+ WVPASS(git._encode_looseobj(b'blob', s, compression_level=i))
+ def encode_pobj(n):
+ return b''.join(git._encode_packobj(b'blob', s, compression_level=n))
+ WVEXCEPT(ValueError, encode_pobj, -1)
+ WVEXCEPT(ValueError, encode_pobj, 10)
+ WVEXCEPT(ValueError, encode_pobj, b'x')
+
+
+def test_packs(tmpdir):
+ environ[b'BUP_DIR'] = bupdir = tmpdir + b'/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(b'%d' % i))
+ log('\n')
+ nameprefix = w.close()
+ print(repr(nameprefix))
+ WVPASS(os.path.exists(nameprefix + b'.pack'))
+ WVPASS(os.path.exists(nameprefix + b'.idx'))
+
+ r = git.open_idx(nameprefix + b'.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(b'\0'*20))
+
+ pi = iter(r)
+ for h in sorted(hashes):
+ WVPASSEQ(hexlify(next(pi)), hexlify(h))
+
+ WVFAIL(r.find_offset(b'\0'*20))
+
+ r = git.PackIdxList(bupdir + b'/objects/pack')
+ WVPASS(r.exists(hashes[5]))
+ WVPASS(r.exists(hashes[6]))
+ WVFAIL(r.exists(b'\0'*20))
+
+
+def test_pack_name_lookup(tmpdir):
+ environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
+ git.init_repo(bupdir)
+ git.verbose = 1
+ packdir = git.repo(b'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(b'%d' % i))
+ log('\n')
+ idxnames.append(os.path.basename(w.close() + b'.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(idxname, r.exists(hashes[i], want_source=True))
+
+
+def test_long_index(tmpdir):
+ environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
+ git.init_repo(bupdir)
+ idx = git.PackIdxV2Writer()
+ 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.add(obj_bin, 1, 0xfffffffff)
+ idx.add(obj2_bin, 2, 0xffffffffff)
+ idx.add(obj3_bin, 3, 0xff)
+ name = tmpdir + b'/tmp.idx'
+ r = idx.write(name, 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)
+
+
+def test_check_repo_or_die(tmpdir):
+ environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
+ orig_cwd = os.getcwd()
+ try:
+ os.chdir(tmpdir)
+ git.init_repo(bupdir)
+ git.check_repo_or_die()
+ # if we reach this point the call above passed
+ WVPASS('check_repo_or_die')
+
+ os.rename(bupdir + b'/objects/pack',
+ bupdir + b'/objects/pack.tmp')
+ open(bupdir + b'/objects/pack', 'w').close()
+ try:
+ git.check_repo_or_die()
+ except SystemExit as e:
+ WVPASSEQ(e.code, 14)
+ else:
+ WVFAIL()
+ os.unlink(bupdir + b'/objects/pack')
+ os.rename(bupdir + b'/objects/pack.tmp',
+ bupdir + b'/objects/pack')
+
+ try:
+ git.check_repo_or_die(b'nonexistantbup.tmp')
+ except SystemExit as e:
+ WVPASSEQ(e.code, 15)
+ else:
+ WVFAIL()
+ finally:
+ os.chdir(orig_cwd)
+
+def test_commit_parsing(tmpdir):
def restore_env_var(name, val):
if val is None:
del environ[name]
return readpipe([b'git', b'show', b'-s',
b'--pretty=format:%s' % val, commit]).strip()
- with no_lingering_errors():
- with test_tempdir(b'bup-tgit-') as tmpdir:
- orig_cwd = os.getcwd()
- workdir = tmpdir + b'/work'
- repodir = workdir + b'/.git'
- orig_author_name = environ.get(b'GIT_AUTHOR_NAME')
- orig_author_email = environ.get(b'GIT_AUTHOR_EMAIL')
- orig_committer_name = environ.get(b'GIT_COMMITTER_NAME')
- orig_committer_email = environ.get(b'GIT_COMMITTER_EMAIL')
- environ[b'GIT_AUTHOR_NAME'] = b'bup test'
- environ[b'GIT_COMMITTER_NAME'] = environ[b'GIT_AUTHOR_NAME']
- environ[b'GIT_AUTHOR_EMAIL'] = b'bup@a425bc70a02811e49bdf73ee56450e6f'
- environ[b'GIT_COMMITTER_EMAIL'] = environ[b'GIT_AUTHOR_EMAIL']
- try:
- readpipe([b'git', b'init', workdir])
- environ[b'GIT_DIR'] = environ[b'BUP_DIR'] = repodir
- git.check_repo_or_die(repodir)
- os.chdir(workdir)
- with open('foo', 'w') as f:
- print('bar', file=f)
- readpipe([b'git', b'add', b'.'])
- readpipe([b'git', b'commit', b'-am', b'Do something',
- b'--author', b'Someone <someone@somewhere>',
- b'--date', b'Sat Oct 3 19:48:49 2009 -0400'])
- commit = readpipe([b'git', b'show-ref', b'-s', b'master']).strip()
- parents = showval(commit, b'%P')
- tree = showval(commit, b'%T')
- cname = showval(commit, b'%cn')
- cmail = showval(commit, b'%ce')
- cdate = showval(commit, b'%ct')
- coffs = showval(commit, b'%ci')
- coffs = coffs[-5:]
- coff = (int(coffs[-4:-2]) * 60 * 60) + (int(coffs[-2:]) * 60)
- if bytes_from_byte(coffs[-5]) == b'-':
- 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, b'Someone')
- WVPASSEQ(commit_items.author_mail, b'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, b'Do something\n')
- with open(b'bar', 'wb') as f:
- f.write(b'baz\n')
- readpipe([b'git', b'add', '.'])
- readpipe([b'git', b'commit', b'-am', b'Do something else'])
- child = readpipe([b'git', b'show-ref', b'-s', b'master']).strip()
- parents = showval(child, b'%P')
- commit_items = git.get_commit_items(child, git.cp())
- WVPASSEQ(commit_items.parents, [commit])
- finally:
- os.chdir(orig_cwd)
- restore_env_var(b'GIT_AUTHOR_NAME', orig_author_name)
- restore_env_var(b'GIT_AUTHOR_EMAIL', orig_author_email)
- restore_env_var(b'GIT_COMMITTER_NAME', orig_committer_name)
- restore_env_var(b'GIT_COMMITTER_EMAIL', orig_committer_email)
-
-
-@wvtest
-def test_new_commit():
- with no_lingering_errors():
- with test_tempdir(b'bup-tgit-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
- git.init_repo(bupdir)
- git.verbose = 1
-
- w = git.PackWriter()
- tree = os.urandom(20)
- parent = os.urandom(20)
- author_name = b'Author'
- author_mail = b'author@somewhere'
- adate_sec = 1439657836
- cdate_sec = adate_sec + 1
- committer_name = b'Committer'
- committer_mail = b'committer@somewhere'
- adate_tz_sec = cdate_tz_sec = None
- commit = w.new_commit(tree, parent,
- b'%s <%s>' % (author_name, author_mail),
- adate_sec, adate_tz_sec,
- b'%s <%s>' % (committer_name, committer_mail),
- cdate_sec, cdate_tz_sec,
- b'There is a small mailbox here')
- adate_tz_sec = -60 * 60
- cdate_tz_sec = 120 * 60
- commit_off = w.new_commit(tree, parent,
- b'%s <%s>' % (author_name, author_mail),
- adate_sec, adate_tz_sec,
- b'%s <%s>' % (committer_name, committer_mail),
- cdate_sec, cdate_tz_sec,
- b'There is a small mailbox here')
- w.close()
-
- commit_items = git.get_commit_items(hexlify(commit), git.cp())
- local_author_offset = localtime(adate_sec).tm_gmtoff
- local_committer_offset = localtime(cdate_sec).tm_gmtoff
- WVPASSEQ(tree, unhexlify(commit_items.tree))
- WVPASSEQ(1, len(commit_items.parents))
- WVPASSEQ(parent, unhexlify(commit_items.parents[0]))
- 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(hexlify(commit_off), git.cp())
- WVPASSEQ(tree, unhexlify(commit_items.tree))
- WVPASSEQ(1, len(commit_items.parents))
- WVPASSEQ(parent, unhexlify(commit_items.parents[0]))
- 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():
- with test_tempdir(b'bup-tgit-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
- src = tmpdir + b'/src'
- mkdirp(src)
- with open(src + b'/1', 'wb+') as f:
- f.write(b'something\n')
- with open(src + b'/2', 'wb+') as f:
- f.write(b'something else\n')
- 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, b'index', src)
- exc(bup_exe, b'save', b'-n', b'src', b'--strip', src)
- src_hash = exo(b'git', b'--git-dir', bupdir,
- b'rev-parse', b'src').strip().split(b'\n')
- assert(len(src_hash) == 1)
- src_hash = unhexlify(src_hash[0])
- tree_hash = unhexlify(exo(b'git', b'--git-dir', bupdir,
- b'rev-parse',
- b'src:').strip().split(b'\n')[0])
- blob_hash = unhexlify(exo(b'git', b'--git-dir', bupdir,
- b'rev-parse',
- b'src:1').strip().split(b'\n')[0])
- WVPASSEQ(frozenset(git.list_refs()),
- frozenset([(b'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([(b'refs/heads/src', src_hash)]))
- exc(b'git', b'--git-dir', bupdir, b'tag', b'commit-tag', b'src')
- WVPASSEQ(frozenset(git.list_refs()),
- frozenset([(b'refs/heads/src', src_hash),
- (b'refs/tags/commit-tag', src_hash)]))
- WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)),
- frozenset([(b'refs/tags/commit-tag', src_hash)]))
- WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)),
- frozenset([(b'refs/heads/src', src_hash)]))
- exc(b'git', b'--git-dir', bupdir, b'tag', b'tree-tag', b'src:')
- exc(b'git', b'--git-dir', bupdir, b'tag', b'blob-tag', b'src:1')
- os.unlink(bupdir + b'/refs/heads/src')
- expected_tags = frozenset([(b'refs/tags/commit-tag', src_hash),
- (b'refs/tags/tree-tag', tree_hash),
- (b'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)
-
-
-@wvtest
-def test__git_date_str():
- with no_lingering_errors():
- WVPASSEQ(b'0 +0000', git._git_date_str(0, 0))
- WVPASSEQ(b'0 -0130', git._git_date_str(0, -90 * 60))
- WVPASSEQ(b'0 +0130', git._git_date_str(0, 90 * 60))
-
-
-@wvtest
-def test_cat_pipe():
- with no_lingering_errors():
- with test_tempdir(b'bup-tgit-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
- src = tmpdir + b'/src'
- mkdirp(src)
- with open(src + b'/1', 'wb+') as f:
- f.write(b'something\n')
- with open(src + b'/2', 'wb+') as f:
- f.write(b'something else\n')
- git.init_repo(bupdir)
- exc(bup_exe, b'index', src)
- oidx = exo(bup_exe, b'save', b'-cn', b'src', b'--strip',
- src).strip()
- typ = exo(b'git', b'--git-dir', bupdir,
- b'cat-file', b'-t', b'src').strip()
- size = int(exo(b'git', b'--git-dir', bupdir,
- b'cat-file', b'-s', b'src'))
- it = git.cp().get(b'src')
- get_info = next(it)
- for buf in next(it):
- pass
- WVPASSEQ((oidx, typ, size), get_info)
+ orig_cwd = os.getcwd()
+ workdir = tmpdir + b'/work'
+ repodir = workdir + b'/.git'
+ orig_author_name = environ.get(b'GIT_AUTHOR_NAME')
+ orig_author_email = environ.get(b'GIT_AUTHOR_EMAIL')
+ orig_committer_name = environ.get(b'GIT_COMMITTER_NAME')
+ orig_committer_email = environ.get(b'GIT_COMMITTER_EMAIL')
+ environ[b'GIT_AUTHOR_NAME'] = b'bup test'
+ environ[b'GIT_COMMITTER_NAME'] = environ[b'GIT_AUTHOR_NAME']
+ environ[b'GIT_AUTHOR_EMAIL'] = b'bup@a425bc70a02811e49bdf73ee56450e6f'
+ environ[b'GIT_COMMITTER_EMAIL'] = environ[b'GIT_AUTHOR_EMAIL']
+ try:
+ readpipe([b'git', b'init', workdir])
+ environ[b'GIT_DIR'] = environ[b'BUP_DIR'] = repodir
+ git.check_repo_or_die(repodir)
+ os.chdir(workdir)
+ with open('foo', 'w') as f:
+ print('bar', file=f)
+ readpipe([b'git', b'add', b'.'])
+ readpipe([b'git', b'commit', b'-am', b'Do something',
+ b'--author', b'Someone <someone@somewhere>',
+ b'--date', b'Sat Oct 3 19:48:49 2009 -0400'])
+ commit = readpipe([b'git', b'show-ref', b'-s', b'master']).strip()
+ parents = showval(commit, b'%P')
+ tree = showval(commit, b'%T')
+ cname = showval(commit, b'%cn')
+ cmail = showval(commit, b'%ce')
+ cdate = showval(commit, b'%ct')
+ coffs = showval(commit, b'%ci')
+ coffs = coffs[-5:]
+ coff = (int(coffs[-4:-2]) * 60 * 60) + (int(coffs[-2:]) * 60)
+ if bytes_from_byte(coffs[-5]) == b'-':
+ 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, b'Someone')
+ WVPASSEQ(commit_items.author_mail, b'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, b'Do something\n')
+ with open(b'bar', 'wb') as f:
+ f.write(b'baz\n')
+ readpipe([b'git', b'add', '.'])
+ readpipe([b'git', b'commit', b'-am', b'Do something else'])
+ child = readpipe([b'git', b'show-ref', b'-s', b'master']).strip()
+ parents = showval(child, b'%P')
+ commit_items = git.get_commit_items(child, git.cp())
+ WVPASSEQ(commit_items.parents, [commit])
+ finally:
+ os.chdir(orig_cwd)
+ restore_env_var(b'GIT_AUTHOR_NAME', orig_author_name)
+ restore_env_var(b'GIT_AUTHOR_EMAIL', orig_author_email)
+ restore_env_var(b'GIT_COMMITTER_NAME', orig_committer_name)
+ restore_env_var(b'GIT_COMMITTER_EMAIL', orig_committer_email)
+
+
+def test_new_commit(tmpdir):
+ environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
+ git.init_repo(bupdir)
+ git.verbose = 1
+
+ w = git.PackWriter()
+ tree = os.urandom(20)
+ parent = os.urandom(20)
+ author_name = b'Author'
+ author_mail = b'author@somewhere'
+ adate_sec = 1439657836
+ cdate_sec = adate_sec + 1
+ committer_name = b'Committer'
+ committer_mail = b'committer@somewhere'
+ adate_tz_sec = cdate_tz_sec = None
+ commit = w.new_commit(tree, parent,
+ b'%s <%s>' % (author_name, author_mail),
+ adate_sec, adate_tz_sec,
+ b'%s <%s>' % (committer_name, committer_mail),
+ cdate_sec, cdate_tz_sec,
+ b'There is a small mailbox here')
+ adate_tz_sec = -60 * 60
+ cdate_tz_sec = 120 * 60
+ commit_off = w.new_commit(tree, parent,
+ b'%s <%s>' % (author_name, author_mail),
+ adate_sec, adate_tz_sec,
+ b'%s <%s>' % (committer_name, committer_mail),
+ cdate_sec, cdate_tz_sec,
+ b'There is a small mailbox here')
+ w.close()
+
+ commit_items = git.get_commit_items(hexlify(commit), git.cp())
+ local_author_offset = localtime(adate_sec).tm_gmtoff
+ local_committer_offset = localtime(cdate_sec).tm_gmtoff
+ WVPASSEQ(tree, unhexlify(commit_items.tree))
+ WVPASSEQ(1, len(commit_items.parents))
+ WVPASSEQ(parent, unhexlify(commit_items.parents[0]))
+ 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(hexlify(commit_off), git.cp())
+ WVPASSEQ(tree, unhexlify(commit_items.tree))
+ WVPASSEQ(1, len(commit_items.parents))
+ WVPASSEQ(parent, unhexlify(commit_items.parents[0]))
+ 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)
+
+
+def test_list_refs(tmpdir):
+ environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
+ src = tmpdir + b'/src'
+ mkdirp(src)
+ with open(src + b'/1', 'wb+') as f:
+ f.write(b'something\n')
+ with open(src + b'/2', 'wb+') as f:
+ f.write(b'something else\n')
+ 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, b'index', src)
+ exc(bup_exe, b'save', b'-n', b'src', b'--strip', src)
+ src_hash = exo(b'git', b'--git-dir', bupdir,
+ b'rev-parse', b'src').strip().split(b'\n')
+ assert(len(src_hash) == 1)
+ src_hash = unhexlify(src_hash[0])
+ tree_hash = unhexlify(exo(b'git', b'--git-dir', bupdir,
+ b'rev-parse',
+ b'src:').strip().split(b'\n')[0])
+ blob_hash = unhexlify(exo(b'git', b'--git-dir', bupdir,
+ b'rev-parse',
+ b'src:1').strip().split(b'\n')[0])
+ WVPASSEQ(frozenset(git.list_refs()),
+ frozenset([(b'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([(b'refs/heads/src', src_hash)]))
+ exc(b'git', b'--git-dir', bupdir, b'tag', b'commit-tag', b'src')
+ WVPASSEQ(frozenset(git.list_refs()),
+ frozenset([(b'refs/heads/src', src_hash),
+ (b'refs/tags/commit-tag', src_hash)]))
+ WVPASSEQ(frozenset(git.list_refs(limit_to_tags=True)),
+ frozenset([(b'refs/tags/commit-tag', src_hash)]))
+ WVPASSEQ(frozenset(git.list_refs(limit_to_heads=True)),
+ frozenset([(b'refs/heads/src', src_hash)]))
+ exc(b'git', b'--git-dir', bupdir, b'tag', b'tree-tag', b'src:')
+ exc(b'git', b'--git-dir', bupdir, b'tag', b'blob-tag', b'src:1')
+ os.unlink(bupdir + b'/refs/heads/src')
+ expected_tags = frozenset([(b'refs/tags/commit-tag', src_hash),
+ (b'refs/tags/tree-tag', tree_hash),
+ (b'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(b'0 +0000', git._git_date_str(0, 0))
+ WVPASSEQ(b'0 -0130', git._git_date_str(0, -90 * 60))
+ WVPASSEQ(b'0 +0130', git._git_date_str(0, 90 * 60))
+
+
+def test_cat_pipe(tmpdir):
+ environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
+ src = tmpdir + b'/src'
+ mkdirp(src)
+ with open(src + b'/1', 'wb+') as f:
+ f.write(b'something\n')
+ with open(src + b'/2', 'wb+') as f:
+ f.write(b'something else\n')
+ git.init_repo(bupdir)
+ exc(bup_exe, b'index', src)
+ oidx = exo(bup_exe, b'save', b'-cn', b'src', b'--strip',
+ src).strip()
+ typ = exo(b'git', b'--git-dir', bupdir,
+ b'cat-file', b'-t', b'src').strip()
+ size = int(exo(b'git', b'--git-dir', bupdir,
+ b'cat-file', b'-s', b'src'))
+ it = git.cp().get(b'src')
+ get_info = next(it)
+ for buf in next(it):
+ pass
+ WVPASSEQ((oidx, typ, size), get_info)
def _create_idx(d, i):
idx = git.PackIdxV2Writer()
packname = os.path.join(d, b'pack-%s.idx' % hexlify(packbin))
idx.write(packname, packbin)
-@wvtest
-def test_midx_close():
+def test_midx_close(tmpdir):
fddir = b'/proc/self/fd'
try:
os.listdir(fddir)
args = [path.exe(), b'midx', b'--auto', b'--dir', objdir]
check_call(args)
- with no_lingering_errors(), \
- test_tempdir(b'bup-tgit-') as tmpdir:
- environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
- git.init_repo(bupdir)
- # create a few dummy idxes
- for i in range(10):
- _create_idx(tmpdir, i)
- git.auto_midx(tmpdir)
- l = git.PackIdxList(tmpdir)
- # this doesn't exist (yet)
- WVPASSEQ(None, l.exists(struct.pack('18xBB', 10, 0)))
- for i in range(10, 15):
- _create_idx(tmpdir, i)
- # delete the midx ...
- # TODO: why do we need to? git.auto_midx() below doesn't?!
- for fn in os.listdir(tmpdir):
- if fn.endswith(b'.midx'):
- os.unlink(os.path.join(tmpdir, fn))
- # and make a new one
- git.auto_midx(tmpdir)
- # check it still doesn't exist - we haven't refreshed
- WVPASSEQ(None, l.exists(struct.pack('18xBB', 10, 0)))
- # check that we still have the midx open, this really
- # just checks more for the kernel API ('deleted' string)
- for fn in openfiles():
- if not b'midx-' in fn:
- continue
- WVPASSEQ(True, b'deleted' in fn)
- # refresh the PackIdxList
- l.refresh()
- # and check that an object in pack 10 exists now
- WVPASSEQ(True, l.exists(struct.pack('18xBB', 10, 0)))
- for fn in openfiles():
- if not b'midx-' in fn:
- continue
- # check that we don't have it open anymore
- WVPASSEQ(False, b'deleted' in fn)
+ environ[b'BUP_DIR'] = bupdir = tmpdir + b'/bup'
+ git.init_repo(bupdir)
+ # create a few dummy idxes
+ for i in range(10):
+ _create_idx(tmpdir, i)
+ git.auto_midx(tmpdir)
+ l = git.PackIdxList(tmpdir)
+ # this doesn't exist (yet)
+ WVPASSEQ(None, l.exists(struct.pack('18xBB', 10, 0)))
+ for i in range(10, 15):
+ _create_idx(tmpdir, i)
+ # delete the midx ...
+ # TODO: why do we need to? git.auto_midx() below doesn't?!
+ for fn in os.listdir(tmpdir):
+ if fn.endswith(b'.midx'):
+ os.unlink(os.path.join(tmpdir, fn))
+ # and make a new one
+ git.auto_midx(tmpdir)
+ # check it still doesn't exist - we haven't refreshed
+ WVPASSEQ(None, l.exists(struct.pack('18xBB', 10, 0)))
+ # check that we still have the midx open, this really
+ # just checks more for the kernel API ('deleted' string)
+ for fn in openfiles():
+ if not b'midx-' in fn:
+ continue
+ WVPASSEQ(True, b'deleted' in fn)
+ # refresh the PackIdxList
+ l.refresh()
+ # and check that an object in pack 10 exists now
+ WVPASSEQ(True, l.exists(struct.pack('18xBB', 10, 0)))
+ for fn in openfiles():
+ if not b'midx-' in fn:
+ continue
+ # check that we don't have it open anymore
+ WVPASSEQ(False, b'deleted' in fn)
from __future__ import absolute_import
from io import BytesIO
-from wvtest import *
+from wvpytest import *
from bup import hashsplit, _helpers, helpers
from bup.compat import byte_int, bytes_from_uint
-from buptest import no_lingering_errors
def nr_regions(x, max_count=None):
return list(hashsplit._nonresident_page_regions(bytearray(x), 1, max_count))
-@wvtest
def test_nonresident_page_regions():
- 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
+ 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)])
+
+
def test_uncache_ours_upto():
history = []
def mock_fadvise_pages_done(f, ofs, len):
history.append((f, ofs, len))
- 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
+ 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
+
+
def test_rolling_sums():
- with no_lingering_errors():
- WVPASS(_helpers.selftest())
+ WVPASS(_helpers.selftest())
-@wvtest
def test_fanout_behaviour():
-
# Drop in replacement for bupsplit, but splitting if the int value of a
# byte >= BUP_BLOBBITS
basebits = _helpers.blobbits()
return ofs, b
return 0, 0
- 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: b'\x00' * n
- # Return a byte which will be split with a level of n
- sb = lambda n: bytes_from_uint(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
+ 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: b'\x00' * n
+ # Return a byte which will be split with a level of n
+ sb = lambda n: bytes_from_uint(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 math, os, os.path, re, subprocess
from bup import helpers
-from wvtest import *
+from wvpytest import *
from bup.compat import bytes_from_byte, bytes_from_uint, environ
from bup.helpers import (atomically_replaced_file, batchpipe, detect_fakeroot,
path_components, readpipe, stripped_path_components,
shstr,
utc_offset_str)
-from buptest import no_lingering_errors, test_tempdir
import bup._helpers as _helpers
-@wvtest
def test_parse_num():
- with no_lingering_errors():
- pn = parse_num
- WVPASSEQ(pn(b'1'), 1)
- 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))
+ pn = parse_num
+ WVPASSEQ(pn(b'1'), 1)
+ 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():
- with no_lingering_errors():
- if b'FAKEROOTKEY' in environ:
- WVPASS(detect_fakeroot())
- else:
- WVPASS(not detect_fakeroot())
+ if b'FAKEROOTKEY' in environ:
+ WVPASS(detect_fakeroot())
+ else:
+ WVPASS(not detect_fakeroot())
-@wvtest
def test_path_components():
- with no_lingering_errors():
- WVPASSEQ(path_components(b'/'), [(b'', b'/')])
- WVPASSEQ(path_components(b'/foo'), [(b'', b'/'), (b'foo', b'/foo')])
- WVPASSEQ(path_components(b'/foo/'), [(b'', b'/'), (b'foo', b'/foo')])
- WVPASSEQ(path_components(b'/foo/bar'),
- [(b'', b'/'), (b'foo', b'/foo'), (b'bar', b'/foo/bar')])
- WVEXCEPT(Exception, path_components, b'foo')
+ WVPASSEQ(path_components(b'/'), [(b'', b'/')])
+ WVPASSEQ(path_components(b'/foo'), [(b'', b'/'), (b'foo', b'/foo')])
+ WVPASSEQ(path_components(b'/foo/'), [(b'', b'/'), (b'foo', b'/foo')])
+ WVPASSEQ(path_components(b'/foo/bar'),
+ [(b'', b'/'), (b'foo', b'/foo'), (b'bar', b'/foo/bar')])
+ WVEXCEPT(Exception, path_components, b'foo')
-@wvtest
def test_stripped_path_components():
- with no_lingering_errors():
- WVPASSEQ(stripped_path_components(b'/', []), [(b'', b'/')])
- WVPASSEQ(stripped_path_components(b'/', [b'']), [(b'', b'/')])
- WVPASSEQ(stripped_path_components(b'/', [b'/']), [(b'', b'/')])
- WVPASSEQ(stripped_path_components(b'/foo', [b'/']),
- [(b'', b'/'), (b'foo', b'/foo')])
- WVPASSEQ(stripped_path_components(b'/', [b'/foo']), [(b'', b'/')])
- WVPASSEQ(stripped_path_components(b'/foo', [b'/bar']),
- [(b'', b'/'), (b'foo', b'/foo')])
- WVPASSEQ(stripped_path_components(b'/foo', [b'/foo']), [(b'', b'/foo')])
- WVPASSEQ(stripped_path_components(b'/foo/bar', [b'/foo']),
- [(b'', b'/foo'), (b'bar', b'/foo/bar')])
- WVPASSEQ(stripped_path_components(b'/foo/bar', [b'/bar', b'/foo', b'/baz']),
- [(b'', b'/foo'), (b'bar', b'/foo/bar')])
- WVPASSEQ(stripped_path_components(b'/foo/bar/baz', [b'/foo/bar/baz']),
- [(b'', b'/foo/bar/baz')])
- WVEXCEPT(Exception, stripped_path_components, b'foo', [])
+ WVPASSEQ(stripped_path_components(b'/', []), [(b'', b'/')])
+ WVPASSEQ(stripped_path_components(b'/', [b'']), [(b'', b'/')])
+ WVPASSEQ(stripped_path_components(b'/', [b'/']), [(b'', b'/')])
+ WVPASSEQ(stripped_path_components(b'/foo', [b'/']),
+ [(b'', b'/'), (b'foo', b'/foo')])
+ WVPASSEQ(stripped_path_components(b'/', [b'/foo']), [(b'', b'/')])
+ WVPASSEQ(stripped_path_components(b'/foo', [b'/bar']),
+ [(b'', b'/'), (b'foo', b'/foo')])
+ WVPASSEQ(stripped_path_components(b'/foo', [b'/foo']), [(b'', b'/foo')])
+ WVPASSEQ(stripped_path_components(b'/foo/bar', [b'/foo']),
+ [(b'', b'/foo'), (b'bar', b'/foo/bar')])
+ WVPASSEQ(stripped_path_components(b'/foo/bar', [b'/bar', b'/foo', b'/baz']),
+ [(b'', b'/foo'), (b'bar', b'/foo/bar')])
+ WVPASSEQ(stripped_path_components(b'/foo/bar/baz', [b'/foo/bar/baz']),
+ [(b'', b'/foo/bar/baz')])
+ WVEXCEPT(Exception, stripped_path_components, b'foo', [])
-@wvtest
def test_grafted_path_components():
- with no_lingering_errors():
- WVPASSEQ(grafted_path_components([(b'/chroot', b'/')], b'/foo'),
- [(b'', b'/'), (b'foo', b'/foo')])
- WVPASSEQ(grafted_path_components([(b'/foo/bar', b'/')],
- b'/foo/bar/baz/bax'),
- [(b'', b'/foo/bar'),
- (b'baz', b'/foo/bar/baz'),
- (b'bax', b'/foo/bar/baz/bax')])
- WVPASSEQ(grafted_path_components([(b'/foo/bar/baz', b'/bax')],
- b'/foo/bar/baz/1/2'),
- [(b'', None),
- (b'bax', b'/foo/bar/baz'),
- (b'1', b'/foo/bar/baz/1'),
- (b'2', b'/foo/bar/baz/1/2')])
- WVPASSEQ(grafted_path_components([(b'/foo', b'/bar/baz/bax')],
- b'/foo/bar'),
- [(b'', None),
- (b'bar', None),
- (b'baz', None),
- (b'bax', b'/foo'),
- (b'bar', b'/foo/bar')])
- WVPASSEQ(grafted_path_components([(b'/foo/bar/baz', b'/a/b/c')],
- b'/foo/bar/baz'),
- [(b'', None), (b'a', None), (b'b', None), (b'c', b'/foo/bar/baz')])
- WVPASSEQ(grafted_path_components([(b'/', b'/a/b/c/')], b'/foo/bar'),
- [(b'', None), (b'a', None), (b'b', None), (b'c', b'/'),
- (b'foo', b'/foo'), (b'bar', b'/foo/bar')])
- WVEXCEPT(Exception, grafted_path_components, b'foo', [])
+ WVPASSEQ(grafted_path_components([(b'/chroot', b'/')], b'/foo'),
+ [(b'', b'/'), (b'foo', b'/foo')])
+ WVPASSEQ(grafted_path_components([(b'/foo/bar', b'/')],
+ b'/foo/bar/baz/bax'),
+ [(b'', b'/foo/bar'),
+ (b'baz', b'/foo/bar/baz'),
+ (b'bax', b'/foo/bar/baz/bax')])
+ WVPASSEQ(grafted_path_components([(b'/foo/bar/baz', b'/bax')],
+ b'/foo/bar/baz/1/2'),
+ [(b'', None),
+ (b'bax', b'/foo/bar/baz'),
+ (b'1', b'/foo/bar/baz/1'),
+ (b'2', b'/foo/bar/baz/1/2')])
+ WVPASSEQ(grafted_path_components([(b'/foo', b'/bar/baz/bax')],
+ b'/foo/bar'),
+ [(b'', None),
+ (b'bar', None),
+ (b'baz', None),
+ (b'bax', b'/foo'),
+ (b'bar', b'/foo/bar')])
+ WVPASSEQ(grafted_path_components([(b'/foo/bar/baz', b'/a/b/c')],
+ b'/foo/bar/baz'),
+ [(b'', None), (b'a', None), (b'b', None), (b'c', b'/foo/bar/baz')])
+ WVPASSEQ(grafted_path_components([(b'/', b'/a/b/c/')], b'/foo/bar'),
+ [(b'', None), (b'a', None), (b'b', None), (b'c', b'/'),
+ (b'foo', b'/foo'), (b'bar', b'/foo/bar')])
+ WVEXCEPT(Exception, grafted_path_components, b'foo', [])
-@wvtest
def test_shstr():
- with no_lingering_errors():
- # Do nothing for strings and bytes
- WVPASSEQ(shstr(b''), b'')
- WVPASSEQ(shstr(b'1'), b'1')
- WVPASSEQ(shstr(b'1 2'), b'1 2')
- WVPASSEQ(shstr(b"1'2"), b"1'2")
- WVPASSEQ(shstr(''), '')
- WVPASSEQ(shstr('1'), '1')
- WVPASSEQ(shstr('1 2'), '1 2')
- WVPASSEQ(shstr("1'2"), "1'2")
+ # Do nothing for strings and bytes
+ WVPASSEQ(shstr(b''), b'')
+ WVPASSEQ(shstr(b'1'), b'1')
+ WVPASSEQ(shstr(b'1 2'), b'1 2')
+ WVPASSEQ(shstr(b"1'2"), b"1'2")
+ WVPASSEQ(shstr(''), '')
+ WVPASSEQ(shstr('1'), '1')
+ WVPASSEQ(shstr('1 2'), '1 2')
+ WVPASSEQ(shstr("1'2"), "1'2")
+
+ # Escape parts of sequences
+ WVPASSEQ(shstr((b'1 2', b'3')), b"'1 2' 3")
+ WVPASSEQ(shstr((b"1'2", b'3')), b"'1'\"'\"'2' 3")
+ WVPASSEQ(shstr((b"'1", b'3')), b"''\"'\"'1' 3")
+ WVPASSEQ(shstr(('1 2', '3')), "'1 2' 3")
+ WVPASSEQ(shstr(("1'2", '3')), "'1'\"'\"'2' 3")
+ WVPASSEQ(shstr(("'1", '3')), "''\"'\"'1' 3")
- # Escape parts of sequences
- WVPASSEQ(shstr((b'1 2', b'3')), b"'1 2' 3")
- WVPASSEQ(shstr((b"1'2", b'3')), b"'1'\"'\"'2' 3")
- WVPASSEQ(shstr((b"'1", b'3')), b"''\"'\"'1' 3")
- WVPASSEQ(shstr(('1 2', '3')), "'1 2' 3")
- WVPASSEQ(shstr(("1'2", '3')), "'1'\"'\"'2' 3")
- WVPASSEQ(shstr(("'1", '3')), "''\"'\"'1' 3")
-
-@wvtest
def test_readpipe():
- with no_lingering_errors():
- x = readpipe([b'echo', b'42'])
- WVPASSEQ(x, b'42\n')
- try:
- readpipe([b'bash', b'-c', b'exit 42'])
- except Exception as ex:
- rx = '^subprocess b?"bash -c \'exit 42\'" failed with status 42$'
- if not re.match(rx, str(ex)):
- WVPASSEQ(str(ex), rx)
+ x = readpipe([b'echo', b'42'])
+ WVPASSEQ(x, b'42\n')
+ try:
+ readpipe([b'bash', b'-c', b'exit 42'])
+ except Exception as ex:
+ rx = '^subprocess b?"bash -c \'exit 42\'" failed with status 42$'
+ if not re.match(rx, str(ex)):
+ WVPASSEQ(str(ex), rx)
-@wvtest
def test_batchpipe():
- with no_lingering_errors():
- for chunk in batchpipe([b'echo'], []):
- WVPASS(False)
- out = b''
- for chunk in batchpipe([b'echo'], [b'42']):
- out += chunk
- WVPASSEQ(out, b'42\n')
- try:
- batchpipe([b'bash', b'-c'], [b'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([b'echo']) + helpers._argmax_args_size(args[:3])
- batches = batchpipe(['echo'], args, arg_max=arg_max)
- WVPASSEQ(next(batches), b'0 1 2\n')
- WVPASSEQ(next(batches), b'3 4 5\n')
- WVPASSEQ(next(batches, None), None)
- batches = batchpipe([b'echo'], [str(x) for x in range(5)], arg_max=arg_max)
- WVPASSEQ(next(batches), b'0 1 2\n')
- WVPASSEQ(next(batches), b'3 4\n')
- WVPASSEQ(next(batches, None), None)
-
-
-@wvtest
-def test_atomically_replaced_file():
- with no_lingering_errors():
- with test_tempdir(b'bup-thelper-') as tmpdir:
- target_file = os.path.join(tmpdir, b'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:
- WVPASSEQ(f.read(), 'asdf')
-
- with atomically_replaced_file(target_file, mode='wb') as f:
- f.write(os.urandom(20))
- WVPASSEQ(f.mode, 'wb')
+ for chunk in batchpipe([b'echo'], []):
+ WVPASS(False)
+ out = b''
+ for chunk in batchpipe([b'echo'], [b'42']):
+ out += chunk
+ WVPASSEQ(out, b'42\n')
+ try:
+ batchpipe([b'bash', b'-c'], [b'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([b'echo']) + helpers._argmax_args_size(args[:3])
+ batches = batchpipe(['echo'], args, arg_max=arg_max)
+ WVPASSEQ(next(batches), b'0 1 2\n')
+ WVPASSEQ(next(batches), b'3 4 5\n')
+ WVPASSEQ(next(batches, None), None)
+ batches = batchpipe([b'echo'], [str(x) for x in range(5)], arg_max=arg_max)
+ WVPASSEQ(next(batches), b'0 1 2\n')
+ WVPASSEQ(next(batches), b'3 4\n')
+ WVPASSEQ(next(batches, None), None)
+
+
+def test_atomically_replaced_file(tmpdir):
+ target_file = os.path.join(tmpdir, b'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:
+ WVPASSEQ(f.read(), 'asdf')
+
+ with atomically_replaced_file(target_file, mode='wb') as f:
+ f.write(os.urandom(20))
+ WVPASSEQ(f.mode, 'wb')
def set_tz(tz):
tzset()
-@wvtest
def test_utc_offset_str():
- with no_lingering_errors():
- tz = environ.get(b'TZ')
- tzset()
- try:
- set_tz(b'FOO+0:00')
- WVPASSEQ(utc_offset_str(0), b'+0000')
- set_tz(b'FOO+1:00')
- WVPASSEQ(utc_offset_str(0), b'-0100')
- set_tz(b'FOO-1:00')
- WVPASSEQ(utc_offset_str(0), b'+0100')
- set_tz(b'FOO+3:3')
- WVPASSEQ(utc_offset_str(0), b'-0303')
- set_tz(b'FOO-3:3')
- WVPASSEQ(utc_offset_str(0), b'+0303')
- # Offset is not an integer number of minutes
- set_tz(b'FOO+3:3:3')
- WVPASSEQ(utc_offset_str(1), b'-0303')
- set_tz(b'FOO-3:3:3')
- WVPASSEQ(utc_offset_str(1), b'+0303')
- WVPASSEQ(utc_offset_str(314159), b'+0303')
- finally:
- if tz:
- set_tz(tz)
- else:
- try:
- set_tz(None)
- except KeyError:
- pass
+ tz = environ.get(b'TZ')
+ tzset()
+ try:
+ set_tz(b'FOO+0:00')
+ WVPASSEQ(utc_offset_str(0), b'+0000')
+ set_tz(b'FOO+1:00')
+ WVPASSEQ(utc_offset_str(0), b'-0100')
+ set_tz(b'FOO-1:00')
+ WVPASSEQ(utc_offset_str(0), b'+0100')
+ set_tz(b'FOO+3:3')
+ WVPASSEQ(utc_offset_str(0), b'-0303')
+ set_tz(b'FOO-3:3')
+ WVPASSEQ(utc_offset_str(0), b'+0303')
+ # Offset is not an integer number of minutes
+ set_tz(b'FOO+3:3:3')
+ WVPASSEQ(utc_offset_str(1), b'-0303')
+ set_tz(b'FOO-3:3:3')
+ WVPASSEQ(utc_offset_str(1), b'+0303')
+ WVPASSEQ(utc_offset_str(314159), b'+0303')
+ finally:
+ if tz:
+ set_tz(tz)
+ else:
+ try:
+ set_tz(None)
+ except KeyError:
+ pass
-@wvtest
def test_valid_save_name():
- with no_lingering_errors():
- valid = helpers.valid_save_name
- WVPASS(valid(b'x'))
- WVPASS(valid(b'x@'))
- WVFAIL(valid(b'@'))
- WVFAIL(valid(b'/'))
- WVFAIL(valid(b'/foo'))
- WVFAIL(valid(b'foo/'))
- WVFAIL(valid(b'/foo/'))
- WVFAIL(valid(b'foo//bar'))
- WVFAIL(valid(b'.'))
- WVFAIL(valid(b'bar.'))
- WVFAIL(valid(b'foo@{'))
- for x in b' ~^:?*[\\':
- WVFAIL(valid(b'foo' + bytes_from_byte(x)))
- for i in range(20):
- WVFAIL(valid(b'foo' + bytes_from_uint(i)))
- WVFAIL(valid(b'foo' + bytes_from_uint(0x7f)))
- WVFAIL(valid(b'foo..bar'))
- WVFAIL(valid(b'bar.lock/baz'))
- WVFAIL(valid(b'foo/bar.lock/baz'))
- WVFAIL(valid(b'.bar/baz'))
- WVFAIL(valid(b'foo/.bar/baz'))
+ valid = helpers.valid_save_name
+ WVPASS(valid(b'x'))
+ WVPASS(valid(b'x@'))
+ WVFAIL(valid(b'@'))
+ WVFAIL(valid(b'/'))
+ WVFAIL(valid(b'/foo'))
+ WVFAIL(valid(b'foo/'))
+ WVFAIL(valid(b'/foo/'))
+ WVFAIL(valid(b'foo//bar'))
+ WVFAIL(valid(b'.'))
+ WVFAIL(valid(b'bar.'))
+ WVFAIL(valid(b'foo@{'))
+ for x in b' ~^:?*[\\':
+ WVFAIL(valid(b'foo' + bytes_from_byte(x)))
+ for i in range(20):
+ WVFAIL(valid(b'foo' + bytes_from_uint(i)))
+ WVFAIL(valid(b'foo' + bytes_from_uint(0x7f)))
+ WVFAIL(valid(b'foo..bar'))
+ WVFAIL(valid(b'bar.lock/baz'))
+ WVFAIL(valid(b'foo/bar.lock/baz'))
+ WVFAIL(valid(b'.bar/baz'))
+ WVFAIL(valid(b'foo/.bar/baz'))
from __future__ import absolute_import, print_function
import os, time
-from wvtest import *
+from wvpytest import *
from bup import index, metadata
from bup.compat import fsencode
from bup.helpers import mkdirp, resolve_parent
-from buptest import no_lingering_errors, test_tempdir
import bup.xstat as xstat
lib_t_dir = os.path.dirname(fsencode(__file__))
-@wvtest
-def index_basic():
- with no_lingering_errors():
- cd = os.path.realpath(b'../')
- WVPASS(cd)
- sd = os.path.realpath(cd + b'/sampledata')
- WVPASSEQ(resolve_parent(cd + b'/sampledata'), sd)
- WVPASSEQ(os.path.realpath(cd + b'/sampledata/x'), sd + b'/x')
- WVPASSEQ(os.path.realpath(cd + b'/sampledata/var/abs-symlink'),
- sd + b'/var/abs-symlink-target')
- WVPASSEQ(resolve_parent(cd + b'/sampledata/var/abs-symlink'),
- sd + b'/var/abs-symlink')
-
-
-@wvtest
-def index_writer():
- with no_lingering_errors():
- with test_tempdir(b'bup-tindex-') as tmpdir:
- orig_cwd = os.getcwd()
- try:
- os.chdir(tmpdir)
- ds = xstat.stat(b'.')
- fs = xstat.stat(lib_t_dir + b'/test_index.py')
- ms = index.MetaStoreWriter(b'index.meta.tmp');
- tmax = (time.time() - 1) * 10**9
- w = index.Writer(b'index.tmp', ms, tmax)
- w.add(b'/var/tmp/sporky', fs, 0)
- w.add(b'/etc/passwd', fs, 0)
- w.add(b'/etc/', ds, 0)
- w.add(b'/', ds, 0)
- ms.close()
- w.close()
- finally:
- os.chdir(orig_cwd)
+def test_index_basic():
+ cd = os.path.realpath(os.path.join(lib_t_dir, b'../'))
+ WVPASS(cd)
+ sd = os.path.realpath(cd + b'/sampledata')
+ WVPASSEQ(resolve_parent(cd + b'/sampledata'), sd)
+ WVPASSEQ(os.path.realpath(cd + b'/sampledata/x'), sd + b'/x')
+ WVPASSEQ(os.path.realpath(cd + b'/sampledata/var/abs-symlink'),
+ sd + b'/var/abs-symlink-target')
+ WVPASSEQ(resolve_parent(cd + b'/sampledata/var/abs-symlink'),
+ sd + b'/var/abs-symlink')
+
+
+def test_index_writer(tmpdir):
+ orig_cwd = os.getcwd()
+ try:
+ os.chdir(tmpdir)
+ ds = xstat.stat(b'.')
+ fs = xstat.stat(lib_t_dir + b'/test_index.py')
+ ms = index.MetaStoreWriter(b'index.meta.tmp');
+ tmax = (time.time() - 1) * 10**9
+ w = index.Writer(b'index.tmp', ms, tmax)
+ w.add(b'/var/tmp/sporky', fs, 0)
+ w.add(b'/etc/passwd', fs, 0)
+ w.add(b'/etc/', ds, 0)
+ w.add(b'/', ds, 0)
+ ms.close()
+ w.close()
+ finally:
+ os.chdir(orig_cwd)
def dump(m):
if e.name == ename:
return e
-@wvtest
-def index_negative_timestamps():
- with no_lingering_errors():
- with test_tempdir(b'bup-tindex-') as tmpdir:
- # Makes 'foo' exist
- foopath = tmpdir + b'/foo'
- f = open(foopath, 'wb')
- f.close()
-
- # Dec 31, 1969
- os.utime(foopath, (-86400, -86400))
- ns_per_sec = 10**9
- tmax = (time.time() - 1) * ns_per_sec
- e = index.BlankNewEntry(foopath, 0, tmax)
- e.update_from_stat(xstat.stat(foopath), 0)
- WVPASS(e.packed())
-
- # Jun 10, 1893
- os.utime(foopath, (-0x80000000, -0x80000000))
- e = index.BlankNewEntry(foopath, 0, tmax)
- e.update_from_stat(xstat.stat(foopath), 0)
- WVPASS(e.packed())
-
-
-@wvtest
-def index_dirty():
- with no_lingering_errors():
- with test_tempdir(b'bup-tindex-') as tmpdir:
- orig_cwd = os.getcwd()
- try:
- os.chdir(tmpdir)
- default_meta = metadata.Metadata()
- ms1 = index.MetaStoreWriter(b'index.meta.tmp')
- ms2 = index.MetaStoreWriter(b'index2.meta.tmp')
- ms3 = index.MetaStoreWriter(b'index3.meta.tmp')
- meta_ofs1 = ms1.store(default_meta)
- meta_ofs2 = ms2.store(default_meta)
- meta_ofs3 = ms3.store(default_meta)
-
- ds = xstat.stat(lib_t_dir)
- fs = xstat.stat(lib_t_dir + b'/test_index.py')
- tmax = (time.time() - 1) * 10**9
-
- w1 = index.Writer(b'index.tmp', ms1, tmax)
- w1.add(b'/a/b/x', fs, meta_ofs1)
- w1.add(b'/a/b/c', fs, meta_ofs1)
- w1.add(b'/a/b/', ds, meta_ofs1)
- w1.add(b'/a/', ds, meta_ofs1)
- #w1.close()
- WVPASS()
-
- w2 = index.Writer(b'index2.tmp', ms2, tmax)
- w2.add(b'/a/b/n/2', fs, meta_ofs2)
- #w2.close()
- WVPASS()
-
- w3 = index.Writer(b'index3.tmp', ms3, tmax)
- w3.add(b'/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,
- [b'/a/b/x', b'/a/b/c', b'/a/b/', b'/a/', b'/'])
- r2all = [e.name for e in r2]
- WVPASSEQ(r2all,
- [b'/a/b/n/2', b'/a/b/n/', b'/a/b/', b'/a/', b'/'])
- r3all = [e.name for e in r3]
- WVPASSEQ(r3all,
- [b'/a/c/n/3', b'/a/c/n/', b'/a/c/', b'/a/', b'/'])
- all = [e.name for e in index.merge(r2, r1, r3)]
- WVPASSEQ(all,
- [b'/a/c/n/3', b'/a/c/n/', b'/a/c/',
- b'/a/b/x', b'/a/b/n/2', b'/a/b/n/', b'/a/b/c',
- b'/a/b/', b'/a/', b'/'])
- fake_validate(r1)
- dump(r1)
-
- 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()],
- [b'/a/c/n/3', b'/a/c/n/', b'/a/c/',
- b'/a/b/n/2', b'/a/b/n/', b'/a/b/', b'/a/', b'/'])
-
- expect_invalid = [b'/'] + r2all + r3all
- expect_real = (set(r1all) - set(r2all) - set(r3all)) \
- | set([b'/a/b/n/2', b'/a/c/n/3'])
- dump(index.merge(r2, r1, r3))
- for e in index.merge(r2, r1, r3):
- print(e.name, hex(e.flags), e.ctime)
- 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), b'/a/b/c')
- e.invalidate()
- e.repack()
- dump(index.merge(r2, r1, r3))
- WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
- [b'/a/b/c', b'/a/b/', b'/a/', b'/'])
- w1.close()
- w2.close()
- w3.close()
- finally:
- os.chdir(orig_cwd)
+def test_index_negative_timestamps(tmpdir):
+ # Makes 'foo' exist
+ foopath = tmpdir + b'/foo'
+ f = open(foopath, 'wb')
+ f.close()
+
+ # Dec 31, 1969
+ os.utime(foopath, (-86400, -86400))
+ ns_per_sec = 10**9
+ tmax = (time.time() - 1) * ns_per_sec
+ e = index.BlankNewEntry(foopath, 0, tmax)
+ e.update_from_stat(xstat.stat(foopath), 0)
+ WVPASS(e.packed())
+
+ # Jun 10, 1893
+ os.utime(foopath, (-0x80000000, -0x80000000))
+ e = index.BlankNewEntry(foopath, 0, tmax)
+ e.update_from_stat(xstat.stat(foopath), 0)
+ WVPASS(e.packed())
+
+
+def test_index_dirty(tmpdir):
+ orig_cwd = os.getcwd()
+ try:
+ os.chdir(tmpdir)
+ default_meta = metadata.Metadata()
+ ms1 = index.MetaStoreWriter(b'index.meta.tmp')
+ ms2 = index.MetaStoreWriter(b'index2.meta.tmp')
+ ms3 = index.MetaStoreWriter(b'index3.meta.tmp')
+ meta_ofs1 = ms1.store(default_meta)
+ meta_ofs2 = ms2.store(default_meta)
+ meta_ofs3 = ms3.store(default_meta)
+
+ ds = xstat.stat(lib_t_dir)
+ fs = xstat.stat(lib_t_dir + b'/test_index.py')
+ tmax = (time.time() - 1) * 10**9
+
+ w1 = index.Writer(b'index.tmp', ms1, tmax)
+ w1.add(b'/a/b/x', fs, meta_ofs1)
+ w1.add(b'/a/b/c', fs, meta_ofs1)
+ w1.add(b'/a/b/', ds, meta_ofs1)
+ w1.add(b'/a/', ds, meta_ofs1)
+ #w1.close()
+ WVPASS()
+
+ w2 = index.Writer(b'index2.tmp', ms2, tmax)
+ w2.add(b'/a/b/n/2', fs, meta_ofs2)
+ #w2.close()
+ WVPASS()
+
+ w3 = index.Writer(b'index3.tmp', ms3, tmax)
+ w3.add(b'/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,
+ [b'/a/b/x', b'/a/b/c', b'/a/b/', b'/a/', b'/'])
+ r2all = [e.name for e in r2]
+ WVPASSEQ(r2all,
+ [b'/a/b/n/2', b'/a/b/n/', b'/a/b/', b'/a/', b'/'])
+ r3all = [e.name for e in r3]
+ WVPASSEQ(r3all,
+ [b'/a/c/n/3', b'/a/c/n/', b'/a/c/', b'/a/', b'/'])
+ all = [e.name for e in index.merge(r2, r1, r3)]
+ WVPASSEQ(all,
+ [b'/a/c/n/3', b'/a/c/n/', b'/a/c/',
+ b'/a/b/x', b'/a/b/n/2', b'/a/b/n/', b'/a/b/c',
+ b'/a/b/', b'/a/', b'/'])
+ fake_validate(r1)
+ dump(r1)
+
+ 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()],
+ [b'/a/c/n/3', b'/a/c/n/', b'/a/c/',
+ b'/a/b/n/2', b'/a/b/n/', b'/a/b/', b'/a/', b'/'])
+
+ expect_invalid = [b'/'] + r2all + r3all
+ expect_real = (set(r1all) - set(r2all) - set(r3all)) \
+ | set([b'/a/b/n/2', b'/a/c/n/3'])
+ dump(index.merge(r2, r1, r3))
+ for e in index.merge(r2, r1, r3):
+ print(e.name, hex(e.flags), e.ctime)
+ 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), b'/a/b/c')
+ e.invalidate()
+ e.repack()
+ dump(index.merge(r2, r1, r3))
+ WVPASSEQ([e.name for e in index.merge(r2, r1, r3) if not e.is_valid()],
+ [b'/a/b/c', b'/a/b/', b'/a/', b'/'])
+ w1.close()
+ w2.close()
+ w3.close()
+ finally:
+ os.chdir(orig_cwd)
from __future__ import absolute_import, print_function
import errno, glob, grp, pwd, stat, tempfile, subprocess
+import os, sys
+import pytest
-from wvtest import *
+from wvpytest import *
from bup import git, metadata
from bup import vfs
from bup.helpers import clear_errors, detect_fakeroot, is_superuser, resolve_parent
from bup.repo import LocalRepo
from bup.xstat import utime, lutime
-from buptest import no_lingering_errors, test_tempdir
import bup.helpers as helpers
+from bup.compat import fsencode
+lib_t_dir = os.path.dirname(fsencode(__file__))
+
+top_dir = os.path.realpath(os.path.join(lib_t_dir, b'..', b'..'))
-top_dir = b'../..'
bup_path = top_dir + b'/bup'
-start_dir = os.getcwd()
def ex(*cmd):
helpers.unlink(b'testfs.img')
-@wvtest
def test_clean_up_archive_path():
- with no_lingering_errors():
- cleanup = metadata._clean_up_path_for_archive
- WVPASSEQ(cleanup(b'foo'), b'foo')
- WVPASSEQ(cleanup(b'/foo'), b'foo')
- WVPASSEQ(cleanup(b'///foo'), b'foo')
- WVPASSEQ(cleanup(b'/foo/bar'), b'foo/bar')
- WVPASSEQ(cleanup(b'foo/./bar'), b'foo/bar')
- WVPASSEQ(cleanup(b'/foo/./bar'), b'foo/bar')
- WVPASSEQ(cleanup(b'/foo/./bar/././baz'), b'foo/bar/baz')
- WVPASSEQ(cleanup(b'/foo/./bar///././baz'), b'foo/bar/baz')
- WVPASSEQ(cleanup(b'//./foo/./bar///././baz/.///'), b'foo/bar/baz/')
- WVPASSEQ(cleanup(b'./foo/./.bar'), b'foo/.bar')
- WVPASSEQ(cleanup(b'./foo/.'), b'foo')
- WVPASSEQ(cleanup(b'./foo/..'), b'.')
- WVPASSEQ(cleanup(b'//./..//.../..//.'), b'.')
- WVPASSEQ(cleanup(b'//./..//..././/.'), b'...')
- WVPASSEQ(cleanup(b'/////.'), b'.')
- WVPASSEQ(cleanup(b'/../'), b'.')
- WVPASSEQ(cleanup(b''), b'.')
-
-
-@wvtest
+ cleanup = metadata._clean_up_path_for_archive
+ WVPASSEQ(cleanup(b'foo'), b'foo')
+ WVPASSEQ(cleanup(b'/foo'), b'foo')
+ WVPASSEQ(cleanup(b'///foo'), b'foo')
+ WVPASSEQ(cleanup(b'/foo/bar'), b'foo/bar')
+ WVPASSEQ(cleanup(b'foo/./bar'), b'foo/bar')
+ WVPASSEQ(cleanup(b'/foo/./bar'), b'foo/bar')
+ WVPASSEQ(cleanup(b'/foo/./bar/././baz'), b'foo/bar/baz')
+ WVPASSEQ(cleanup(b'/foo/./bar///././baz'), b'foo/bar/baz')
+ WVPASSEQ(cleanup(b'//./foo/./bar///././baz/.///'), b'foo/bar/baz/')
+ WVPASSEQ(cleanup(b'./foo/./.bar'), b'foo/.bar')
+ WVPASSEQ(cleanup(b'./foo/.'), b'foo')
+ WVPASSEQ(cleanup(b'./foo/..'), b'.')
+ WVPASSEQ(cleanup(b'//./..//.../..//.'), b'.')
+ WVPASSEQ(cleanup(b'//./..//..././/.'), b'...')
+ WVPASSEQ(cleanup(b'/////.'), b'.')
+ WVPASSEQ(cleanup(b'/../'), b'.')
+ WVPASSEQ(cleanup(b''), b'.')
+
+
def test_risky_path():
- with no_lingering_errors():
- risky = metadata._risky_path
- WVPASS(risky(b'/foo'))
- WVPASS(risky(b'///foo'))
- WVPASS(risky(b'/../foo'))
- WVPASS(risky(b'../foo'))
- WVPASS(risky(b'foo/..'))
- WVPASS(risky(b'foo/../'))
- WVPASS(risky(b'foo/../bar'))
- WVFAIL(risky(b'foo'))
- WVFAIL(risky(b'foo/'))
- WVFAIL(risky(b'foo///'))
- WVFAIL(risky(b'./foo'))
- WVFAIL(risky(b'foo/.'))
- WVFAIL(risky(b'./foo/.'))
- WVFAIL(risky(b'foo/bar'))
- WVFAIL(risky(b'foo/./bar'))
-
-
-@wvtest
+ risky = metadata._risky_path
+ WVPASS(risky(b'/foo'))
+ WVPASS(risky(b'///foo'))
+ WVPASS(risky(b'/../foo'))
+ WVPASS(risky(b'../foo'))
+ WVPASS(risky(b'foo/..'))
+ WVPASS(risky(b'foo/../'))
+ WVPASS(risky(b'foo/../bar'))
+ WVFAIL(risky(b'foo'))
+ WVFAIL(risky(b'foo/'))
+ WVFAIL(risky(b'foo///'))
+ WVFAIL(risky(b'./foo'))
+ WVFAIL(risky(b'foo/.'))
+ WVFAIL(risky(b'./foo/.'))
+ WVFAIL(risky(b'foo/bar'))
+ WVFAIL(risky(b'foo/./bar'))
+
+
def test_clean_up_extract_path():
- with no_lingering_errors():
- cleanup = metadata._clean_up_extract_path
- WVPASSEQ(cleanup(b'/foo'), b'foo')
- WVPASSEQ(cleanup(b'///foo'), b'foo')
- WVFAIL(cleanup(b'/../foo'))
- WVFAIL(cleanup(b'../foo'))
- WVFAIL(cleanup(b'foo/..'))
- WVFAIL(cleanup(b'foo/../'))
- WVFAIL(cleanup(b'foo/../bar'))
- WVPASSEQ(cleanup(b'foo'), b'foo')
- WVPASSEQ(cleanup(b'foo/'), b'foo/')
- WVPASSEQ(cleanup(b'foo///'), b'foo///')
- WVPASSEQ(cleanup(b'./foo'), b'./foo')
- WVPASSEQ(cleanup(b'foo/.'), b'foo/.')
- WVPASSEQ(cleanup(b'./foo/.'), b'./foo/.')
- WVPASSEQ(cleanup(b'foo/bar'), b'foo/bar')
- WVPASSEQ(cleanup(b'foo/./bar'), b'foo/./bar')
- WVPASSEQ(cleanup(b'/'), b'.')
- WVPASSEQ(cleanup(b'./'), b'./')
- WVPASSEQ(cleanup(b'///foo/bar'), b'foo/bar')
- WVPASSEQ(cleanup(b'///foo/bar'), b'foo/bar')
-
-
-@wvtest
-def test_metadata_method():
- with no_lingering_errors():
- with test_tempdir(b'bup-tmetadata-') as tmpdir:
- bup_dir = tmpdir + b'/bup'
- data_path = tmpdir + b'/foo'
- os.mkdir(data_path)
- ex(b'touch', data_path + b'/file')
- ex(b'ln', b'-s', b'file', data_path + b'/symlink')
- test_time1 = 13 * 1000000000
- test_time2 = 42 * 1000000000
- utime(data_path + b'/file', (0, test_time1))
- lutime(data_path + b'/symlink', (0, 0))
- utime(data_path, (0, test_time2))
- ex(bup_path, b'-d', bup_dir, b'init')
- ex(bup_path, b'-d', bup_dir, b'index', b'-v', data_path)
- ex(bup_path, b'-d', bup_dir, b'save', b'-tvvn', b'test', data_path)
- git.check_repo_or_die(bup_dir)
- repo = LocalRepo()
- resolved = vfs.resolve(repo,
- b'/test/latest' + resolve_parent(data_path),
- follow=False)
- leaf_name, leaf_item = resolved[-1]
- m = leaf_item.meta
- WVPASS(m.mtime == test_time2)
- WVPASS(leaf_name == b'foo')
- contents = tuple(vfs.contents(repo, leaf_item))
- WVPASS(len(contents) == 3)
- WVPASSEQ(frozenset(name for name, item in contents),
- frozenset((b'.', b'file', b'symlink')))
- for name, item in contents:
- if name == b'file':
- m = item.meta
- WVPASS(m.mtime == test_time1)
- elif name == b'symlink':
- m = item.meta
- WVPASSEQ(m.symlink_target, b'file')
- WVPASSEQ(m.size, 4)
- WVPASSEQ(m.mtime, 0)
+ cleanup = metadata._clean_up_extract_path
+ WVPASSEQ(cleanup(b'/foo'), b'foo')
+ WVPASSEQ(cleanup(b'///foo'), b'foo')
+ WVFAIL(cleanup(b'/../foo'))
+ WVFAIL(cleanup(b'../foo'))
+ WVFAIL(cleanup(b'foo/..'))
+ WVFAIL(cleanup(b'foo/../'))
+ WVFAIL(cleanup(b'foo/../bar'))
+ WVPASSEQ(cleanup(b'foo'), b'foo')
+ WVPASSEQ(cleanup(b'foo/'), b'foo/')
+ WVPASSEQ(cleanup(b'foo///'), b'foo///')
+ WVPASSEQ(cleanup(b'./foo'), b'./foo')
+ WVPASSEQ(cleanup(b'foo/.'), b'foo/.')
+ WVPASSEQ(cleanup(b'./foo/.'), b'./foo/.')
+ WVPASSEQ(cleanup(b'foo/bar'), b'foo/bar')
+ WVPASSEQ(cleanup(b'foo/./bar'), b'foo/./bar')
+ WVPASSEQ(cleanup(b'/'), b'.')
+ WVPASSEQ(cleanup(b'./'), b'./')
+ WVPASSEQ(cleanup(b'///foo/bar'), b'foo/bar')
+ WVPASSEQ(cleanup(b'///foo/bar'), b'foo/bar')
+
+
+def test_metadata_method(tmpdir):
+ bup_dir = tmpdir + b'/bup'
+ data_path = tmpdir + b'/foo'
+ os.mkdir(data_path)
+ ex(b'touch', data_path + b'/file')
+ ex(b'ln', b'-s', b'file', data_path + b'/symlink')
+ test_time1 = 13 * 1000000000
+ test_time2 = 42 * 1000000000
+ utime(data_path + b'/file', (0, test_time1))
+ lutime(data_path + b'/symlink', (0, 0))
+ utime(data_path, (0, test_time2))
+ ex(bup_path, b'-d', bup_dir, b'init')
+ ex(bup_path, b'-d', bup_dir, b'index', b'-v', data_path)
+ ex(bup_path, b'-d', bup_dir, b'save', b'-tvvn', b'test', data_path)
+ git.check_repo_or_die(bup_dir)
+ repo = LocalRepo()
+ resolved = vfs.resolve(repo,
+ b'/test/latest' + resolve_parent(data_path),
+ follow=False)
+ leaf_name, leaf_item = resolved[-1]
+ m = leaf_item.meta
+ WVPASS(m.mtime == test_time2)
+ WVPASS(leaf_name == b'foo')
+ contents = tuple(vfs.contents(repo, leaf_item))
+ WVPASS(len(contents) == 3)
+ WVPASSEQ(frozenset(name for name, item in contents),
+ frozenset((b'.', b'file', b'symlink')))
+ for name, item in contents:
+ if name == b'file':
+ m = item.meta
+ WVPASS(m.mtime == test_time1)
+ elif name == b'symlink':
+ m = item.meta
+ WVPASSEQ(m.symlink_target, b'file')
+ WVPASSEQ(m.size, 4)
+ WVPASSEQ(m.mtime, 0)
def _first_err():
return ''
-@wvtest
-def test_from_path_error():
+def test_from_path_error(tmpdir):
if is_superuser() or detect_fakeroot():
return
- with no_lingering_errors():
- with test_tempdir(b'bup-tmetadata-') as tmpdir:
- path = tmpdir + b'/foo'
- os.mkdir(path)
- m = metadata.from_path(path, archive_path=path, save_symlinks=True)
- WVPASSEQ(m.path, path)
- os.chmod(path, 0o000)
- metadata.from_path(path, archive_path=path, save_symlinks=True)
- if metadata.get_linux_file_attr:
- print('saved_errors:', helpers.saved_errors, file=sys.stderr)
- WVPASS(len(helpers.saved_errors) == 1)
- errmsg = _first_err()
- WVPASS(errmsg.startswith('read Linux attr'))
- clear_errors()
+ path = tmpdir + b'/foo'
+ os.mkdir(path)
+ m = metadata.from_path(path, archive_path=path, save_symlinks=True)
+ WVPASSEQ(m.path, path)
+ os.chmod(path, 0o000)
+ metadata.from_path(path, archive_path=path, save_symlinks=True)
+ if metadata.get_linux_file_attr:
+ print('saved_errors:', helpers.saved_errors, file=sys.stderr)
+ WVPASS(len(helpers.saved_errors) == 1)
+ errmsg = _first_err()
+ WVPASS(errmsg.startswith('read Linux attr'))
+ clear_errors()
def _linux_attr_supported(path):
return True
-@wvtest
-def test_apply_to_path_restricted_access():
+def test_apply_to_path_restricted_access(tmpdir):
if is_superuser() or detect_fakeroot():
return
if sys.platform.startswith('cygwin'):
return # chmod 000 isn't effective.
- with no_lingering_errors():
- with test_tempdir(b'bup-tmetadata-') as tmpdir:
- parent = tmpdir + b'/foo'
- path = parent + b'/bar'
- os.mkdir(parent)
- os.mkdir(path)
- clear_errors()
- if metadata.xattr:
- try:
- metadata.xattr.set(path, b'user.buptest', b'bup')
- except:
- print("failed to set test xattr")
- # ignore any failures here - maybe FS cannot do it
- pass
- m = metadata.from_path(path, archive_path=path, save_symlinks=True)
- WVPASSEQ(m.path, path)
- os.chmod(parent, 0o000)
- m.apply_to_path(path)
- print('saved_errors:', helpers.saved_errors, file=sys.stderr)
- 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 range(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():
- with test_tempdir(b'bup-tmetadata-') as tmpdir:
- path = tmpdir + b'/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 + b'/bar', 'w').close()
- WVEXCEPT(Exception, file_m.create_path, path, create_symlinks=True)
- # Restore dir over non-empty dir.
- os.remove(path + b'/bar')
- os.mkdir(path + b'/bar')
- WVEXCEPT(Exception, dir_m.create_path, path, create_symlinks=True)
+ try:
+ parent = tmpdir + b'/foo'
+ path = parent + b'/bar'
+ os.mkdir(parent)
+ os.mkdir(path)
+ clear_errors()
+ if metadata.xattr:
+ try:
+ metadata.xattr.set(path, b'user.buptest', b'bup')
+ except:
+ print("failed to set test xattr")
+ # ignore any failures here - maybe FS cannot do it
+ pass
+ m = metadata.from_path(path, archive_path=path, save_symlinks=True)
+ WVPASSEQ(m.path, path)
+ os.chmod(parent, 0o000)
+ m.apply_to_path(path)
+ print(b'saved_errors:', helpers.saved_errors, file=sys.stderr)
+ 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 range(len(expected_errors)):
+ assert str(helpers.saved_errors[i]).startswith(expected_errors[i])
+ finally:
+ clear_errors()
+
+
+def test_restore_over_existing_target(tmpdir):
+ path = tmpdir + b'/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 + b'/bar', 'w').close()
+ WVEXCEPT(Exception, file_m.create_path, path, create_symlinks=True)
+ # Restore dir over non-empty dir.
+ os.remove(path + b'/bar')
+ os.mkdir(path + b'/bar')
+ WVEXCEPT(Exception, dir_m.create_path, path, create_symlinks=True)
from bup.metadata import read_acl
-if not read_acl:
- @wvtest
- def POSIX1E_ACL_SUPPORT_IS_MISSING():
- pass
-
from bup.metadata import xattr
if xattr:
return list(filter(lambda i: not i in (b'security.selinux', ),
attrs))
- @wvtest
def test_handling_of_incorrect_existing_linux_xattrs():
if not is_superuser() or detect_fakeroot():
- WVMSG('skipping test -- not superuser')
+ pytest.skip('skipping test -- not superuser')
return
if not setup_testfs():
- WVMSG('unable to load loop module; skipping dependent tests')
+ pytest.skip('unable to load loop module; skipping dependent tests')
return
for f in glob.glob(b'testfs/*'):
ex(b'rm', b'-rf', f)
m.apply_to_path(path, restore_numeric_ids=False)
WVPASSEQ(remove_selinux(xattr.list(path)), [b'user.foo'])
WVPASSEQ(xattr.get(path, b'user.foo'), b'bar')
- os.chdir(start_dir)
cleanup_testfs()
from __future__ import absolute_import
-from wvtest import *
+from wvpytest import *
from bup import options
-from buptest import no_lingering_errors
-@wvtest
def test_optdict():
- 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)
+ 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():
- with no_lingering_errors():
- WVPASS(options.Options(invalid_optspec0).parse([]))
- WVPASS(options.Options(invalid_optspec1).parse([]))
- WVPASS(options.Options(invalid_optspec2).parse([]))
+ WVPASS(options.Options(invalid_optspec0).parse([]))
+ WVPASS(options.Options(invalid_optspec1).parse([]))
+ WVPASS(options.Options(invalid_optspec2).parse([]))
optspec = """
#,compress= set compression level [5]
"""
-@wvtest
def test_options():
- 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))
+ 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 os import symlink
from stat import S_IFDIR
from sys import stderr
+import os
from time import localtime, strftime
-from wvtest import *
+from wvpytest import *
from bup import git, path, vfs
from bup.compat import environ
from bup.io import path_msg
from bup.metadata import Metadata
from bup.repo import LocalRepo, RemoteRepo
-from buptest import ex, exo, no_lingering_errors, test_tempdir
+from buptest import ex, exo
from buptest.vfs import tree_dict
bup_path = path.exe()
## be promoted from a mode to a Metadata instance once the tree it
## refers to is traversed.
-def prep_and_test_repo(name, create_repo, test_repo):
- with no_lingering_errors():
- with test_tempdir(b'bup-t' + name) as tmpdir:
- bup_dir = tmpdir + b'/bup'
- environ[b'GIT_DIR'] = bup_dir
- environ[b'BUP_DIR'] = bup_dir
- ex((bup_path, b'init'))
- git.repodir = bup_dir
- with create_repo(bup_dir) as repo:
- test_repo(repo, tmpdir)
+def prep_and_test_repo(tmpdir, create_repo, test_repo):
+ bup_dir = tmpdir + b'/bup'
+ environ[b'GIT_DIR'] = bup_dir
+ environ[b'BUP_DIR'] = bup_dir
+ ex((bup_path, b'init'))
+ git.repodir = bup_dir
+ with create_repo(bup_dir) as repo:
+ test_repo(repo, tmpdir)
# Currently, we just test through the repos since LocalRepo resolve is
# just a straight redirection to vfs.resolve.
-def test_resolve(repo, tmpdir):
- data_path = tmpdir + b'/src'
- resolve = repo.resolve
- save_time = 100000
- save_time_str = strftime('%Y-%m-%d-%H%M%S', localtime(save_time)).encode('ascii')
- os.mkdir(data_path)
- os.mkdir(data_path + b'/dir')
- with open(data_path + b'/file', 'wb+') as tmpfile:
- tmpfile.write(b'canary\n')
- symlink(b'file', data_path + b'/file-symlink')
- symlink(b'dir', data_path + b'/dir-symlink')
- symlink(b'not-there', data_path + b'/bad-symlink')
- ex((bup_path, b'index', b'-v', data_path))
- ex((bup_path, b'save', b'-d', b'%d' % save_time, b'-tvvn', b'test',
- b'--strip', data_path))
- ex((bup_path, b'tag', b'test-tag', b'test'))
+def _test_resolve(repo, tmpdir):
+ data_path = tmpdir + b'/src'
+ resolve = repo.resolve
+ save_time = 100000
+ save_time_str = strftime('%Y-%m-%d-%H%M%S', localtime(save_time)).encode('ascii')
+ os.mkdir(data_path)
+ os.mkdir(data_path + b'/dir')
+ with open(data_path + b'/file', 'wb+') as tmpfile:
+ tmpfile.write(b'canary\n')
+ symlink(b'file', data_path + b'/file-symlink')
+ symlink(b'dir', data_path + b'/dir-symlink')
+ symlink(b'not-there', data_path + b'/bad-symlink')
+ ex((bup_path, b'index', b'-v', data_path))
+ ex((bup_path, b'save', b'-d', b'%d' % save_time, b'-tvvn', b'test',
+ b'--strip', data_path))
+ ex((bup_path, b'tag', b'test-tag', b'test'))
- tip_hash = exo((b'git', b'show-ref', b'refs/heads/test'))[0]
- tip_oidx = tip_hash.strip().split()[0]
- tip_oid = unhexlify(tip_oidx)
- tip_tree_oidx = exo((b'git', b'log', b'--pretty=%T', b'-n1',
- tip_oidx))[0].strip()
- tip_tree_oid = unhexlify(tip_tree_oidx)
- tip_tree = tree_dict(repo, tip_tree_oid)
- test_revlist_w_meta = vfs.RevList(meta=tip_tree[b'.'].meta,
- oid=tip_oid)
- expected_latest_item = vfs.Commit(meta=S_IFDIR | 0o755,
- oid=tip_tree_oid,
- coid=tip_oid)
- expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
- oid=tip_tree_oid,
- coid=tip_oid)
- expected_latest_link = vfs.FakeLink(meta=vfs.default_symlink_mode,
- target=save_time_str)
- expected_test_tag_item = expected_latest_item
+ tip_hash = exo((b'git', b'show-ref', b'refs/heads/test'))[0]
+ tip_oidx = tip_hash.strip().split()[0]
+ tip_oid = unhexlify(tip_oidx)
+ tip_tree_oidx = exo((b'git', b'log', b'--pretty=%T', b'-n1',
+ tip_oidx))[0].strip()
+ tip_tree_oid = unhexlify(tip_tree_oidx)
+ tip_tree = tree_dict(repo, tip_tree_oid)
+ test_revlist_w_meta = vfs.RevList(meta=tip_tree[b'.'].meta,
+ oid=tip_oid)
+ expected_latest_item = vfs.Commit(meta=S_IFDIR | 0o755,
+ oid=tip_tree_oid,
+ coid=tip_oid)
+ expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
+ oid=tip_tree_oid,
+ coid=tip_oid)
+ expected_latest_link = vfs.FakeLink(meta=vfs.default_symlink_mode,
+ target=save_time_str)
+ expected_test_tag_item = expected_latest_item
- wvstart('resolve: /')
+ vfs.clear_cache()
+ res = resolve(b'/')
+ wvpasseq(1, len(res))
+ wvpasseq(((b'', vfs._root),), res)
+ ignore, root_item = res[0]
+ root_content = frozenset(vfs.contents(repo, root_item))
+ wvpasseq(frozenset([(b'.', root_item),
+ (b'.tag', vfs._tags),
+ (b'test', test_revlist_w_meta)]),
+ root_content)
+ for path in (b'//', b'/.', b'/./', b'/..', b'/../',
+ b'/test/latest/dir/../../..',
+ b'/test/latest/dir/../../../',
+ b'/test/latest/dir/../../../.',
+ b'/test/latest/dir/../../..//',
+ b'/test//latest/dir/../../..',
+ b'/test/./latest/dir/../../..',
+ b'/test/././latest/dir/../../..',
+ b'/test/.//./latest/dir/../../..',
+ b'/test//.//.//latest/dir/../../..'
+ b'/test//./latest/dir/../../..'):
vfs.clear_cache()
- res = resolve(b'/')
- wvpasseq(1, len(res))
+ res = resolve(path)
wvpasseq(((b'', vfs._root),), res)
- ignore, root_item = res[0]
- root_content = frozenset(vfs.contents(repo, root_item))
- wvpasseq(frozenset([(b'.', root_item),
- (b'.tag', vfs._tags),
- (b'test', test_revlist_w_meta)]),
- root_content)
- for path in (b'//', b'/.', b'/./', b'/..', b'/../',
- b'/test/latest/dir/../../..',
- b'/test/latest/dir/../../../',
- b'/test/latest/dir/../../../.',
- b'/test/latest/dir/../../..//',
- b'/test//latest/dir/../../..',
- b'/test/./latest/dir/../../..',
- b'/test/././latest/dir/../../..',
- b'/test/.//./latest/dir/../../..',
- b'/test//.//.//latest/dir/../../..'
- b'/test//./latest/dir/../../..'):
- wvstart('resolve: ' + path_msg(path))
- vfs.clear_cache()
- res = resolve(path)
- wvpasseq(((b'', vfs._root),), res)
- wvstart('resolve: /.tag')
- vfs.clear_cache()
- res = resolve(b'/.tag')
- wvpasseq(2, len(res))
- wvpasseq(((b'', vfs._root), (b'.tag', vfs._tags)),
- res)
- ignore, tag_item = res[1]
- tag_content = frozenset(vfs.contents(repo, tag_item))
- wvpasseq(frozenset([(b'.', tag_item),
- (b'test-tag', expected_test_tag_item)]),
- tag_content)
+ vfs.clear_cache()
+ res = resolve(b'/.tag')
+ wvpasseq(2, len(res))
+ wvpasseq(((b'', vfs._root), (b'.tag', vfs._tags)),
+ res)
+ ignore, tag_item = res[1]
+ tag_content = frozenset(vfs.contents(repo, tag_item))
+ wvpasseq(frozenset([(b'.', tag_item),
+ (b'test-tag', expected_test_tag_item)]),
+ tag_content)
- wvstart('resolve: /test')
- vfs.clear_cache()
- res = resolve(b'/test')
- wvpasseq(2, len(res))
- wvpasseq(((b'', vfs._root), (b'test', test_revlist_w_meta)), res)
- ignore, test_item = res[1]
- test_content = frozenset(vfs.contents(repo, test_item))
- # latest has metadata here due to caching
- wvpasseq(frozenset([(b'.', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta),
- (b'latest', expected_latest_link)]),
- test_content)
+ vfs.clear_cache()
+ res = resolve(b'/test')
+ wvpasseq(2, len(res))
+ wvpasseq(((b'', vfs._root), (b'test', test_revlist_w_meta)), res)
+ ignore, test_item = res[1]
+ test_content = frozenset(vfs.contents(repo, test_item))
+ # latest has metadata here due to caching
+ wvpasseq(frozenset([(b'.', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
+ (b'latest', expected_latest_link)]),
+ test_content)
- wvstart('resolve: /test/latest')
- vfs.clear_cache()
- res = resolve(b'/test/latest')
- wvpasseq(3, len(res))
- expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
- oid=tip_tree_oid,
- coid=tip_oid)
- expected = ((b'', vfs._root),
- (b'test', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta))
- wvpasseq(expected, res)
- ignore, latest_item = res[2]
- latest_content = frozenset(vfs.contents(repo, latest_item))
- expected = frozenset((x.name, vfs.Item(oid=x.oid, meta=x.meta))
- for x in (tip_tree[name]
- for name in (b'.',
- b'bad-symlink',
- b'dir',
- b'dir-symlink',
- b'file',
- b'file-symlink')))
- wvpasseq(expected, latest_content)
+ vfs.clear_cache()
+ res = resolve(b'/test/latest')
+ wvpasseq(3, len(res))
+ expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
+ oid=tip_tree_oid,
+ coid=tip_oid)
+ expected = ((b'', vfs._root),
+ (b'test', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta))
+ wvpasseq(expected, res)
+ ignore, latest_item = res[2]
+ latest_content = frozenset(vfs.contents(repo, latest_item))
+ expected = frozenset((x.name, vfs.Item(oid=x.oid, meta=x.meta))
+ for x in (tip_tree[name]
+ for name in (b'.',
+ b'bad-symlink',
+ b'dir',
+ b'dir-symlink',
+ b'file',
+ b'file-symlink')))
+ wvpasseq(expected, latest_content)
- wvstart('resolve: /test/latest/file')
- vfs.clear_cache()
- res = resolve(b'/test/latest/file')
- wvpasseq(4, len(res))
- expected_file_item_w_meta = vfs.Item(meta=tip_tree[b'file'].meta,
- oid=tip_tree[b'file'].oid)
- expected = ((b'', vfs._root),
- (b'test', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta),
- (b'file', expected_file_item_w_meta))
- wvpasseq(expected, res)
+ vfs.clear_cache()
+ res = resolve(b'/test/latest/file')
+ wvpasseq(4, len(res))
+ expected_file_item_w_meta = vfs.Item(meta=tip_tree[b'file'].meta,
+ oid=tip_tree[b'file'].oid)
+ expected = ((b'', vfs._root),
+ (b'test', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
+ (b'file', expected_file_item_w_meta))
+ wvpasseq(expected, res)
- wvstart('resolve: /test/latest/bad-symlink')
- vfs.clear_cache()
- res = resolve(b'/test/latest/bad-symlink')
- wvpasseq(4, len(res))
- expected = ((b'', vfs._root),
- (b'test', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta),
- (b'not-there', None))
- wvpasseq(expected, res)
+ vfs.clear_cache()
+ res = resolve(b'/test/latest/bad-symlink')
+ wvpasseq(4, len(res))
+ expected = ((b'', vfs._root),
+ (b'test', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
+ (b'not-there', None))
+ wvpasseq(expected, res)
- wvstart('resolve nofollow: /test/latest/bad-symlink')
- vfs.clear_cache()
- res = resolve(b'/test/latest/bad-symlink', follow=False)
- wvpasseq(4, len(res))
- bad_symlink_value = tip_tree[b'bad-symlink']
- expected_bad_symlink_item_w_meta = vfs.Item(meta=bad_symlink_value.meta,
- oid=bad_symlink_value.oid)
- expected = ((b'', vfs._root),
- (b'test', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta),
- (b'bad-symlink', expected_bad_symlink_item_w_meta))
- wvpasseq(expected, res)
+ vfs.clear_cache()
+ res = resolve(b'/test/latest/bad-symlink', follow=False)
+ wvpasseq(4, len(res))
+ bad_symlink_value = tip_tree[b'bad-symlink']
+ expected_bad_symlink_item_w_meta = vfs.Item(meta=bad_symlink_value.meta,
+ oid=bad_symlink_value.oid)
+ expected = ((b'', vfs._root),
+ (b'test', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
+ (b'bad-symlink', expected_bad_symlink_item_w_meta))
+ wvpasseq(expected, res)
- wvstart('resolve: /test/latest/file-symlink')
- vfs.clear_cache()
- res = resolve(b'/test/latest/file-symlink')
- wvpasseq(4, len(res))
- expected = ((b'', vfs._root),
- (b'test', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta),
- (b'file', expected_file_item_w_meta))
- wvpasseq(expected, res)
+ vfs.clear_cache()
+ res = resolve(b'/test/latest/file-symlink')
+ wvpasseq(4, len(res))
+ expected = ((b'', vfs._root),
+ (b'test', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
+ (b'file', expected_file_item_w_meta))
+ wvpasseq(expected, res)
- wvstart('resolve nofollow: /test/latest/file-symlink')
- vfs.clear_cache()
- res = resolve(b'/test/latest/file-symlink', follow=False)
- wvpasseq(4, len(res))
- file_symlink_value = tip_tree[b'file-symlink']
- expected_file_symlink_item_w_meta = vfs.Item(meta=file_symlink_value.meta,
- oid=file_symlink_value.oid)
- expected = ((b'', vfs._root),
- (b'test', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta),
- (b'file-symlink', expected_file_symlink_item_w_meta))
- wvpasseq(expected, res)
+ vfs.clear_cache()
+ res = resolve(b'/test/latest/file-symlink', follow=False)
+ wvpasseq(4, len(res))
+ file_symlink_value = tip_tree[b'file-symlink']
+ expected_file_symlink_item_w_meta = vfs.Item(meta=file_symlink_value.meta,
+ oid=file_symlink_value.oid)
+ expected = ((b'', vfs._root),
+ (b'test', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
+ (b'file-symlink', expected_file_symlink_item_w_meta))
+ wvpasseq(expected, res)
- wvstart('resolve: /test/latest/missing')
- vfs.clear_cache()
- res = resolve(b'/test/latest/missing')
- wvpasseq(4, len(res))
- name, item = res[-1]
- wvpasseq(b'missing', name)
- wvpass(item is None)
+ vfs.clear_cache()
+ res = resolve(b'/test/latest/missing')
+ wvpasseq(4, len(res))
+ name, item = res[-1]
+ wvpasseq(b'missing', name)
+ wvpass(item is None)
- for path in (b'/test/latest/file/',
- b'/test/latest/file/.',
- b'/test/latest/file/..',
- b'/test/latest/file/../',
- b'/test/latest/file/../.',
- b'/test/latest/file/../..',
- b'/test/latest/file/foo'):
- wvstart('resolve: ' + path_msg(path))
- vfs.clear_cache()
- try:
- resolve(path)
- except vfs.IOError as res_ex:
- wvpasseq(ENOTDIR, res_ex.errno)
- wvpasseq([b'', b'test', save_time_str, b'file'],
- [name for name, item in res_ex.terminus])
-
- for path in (b'/test/latest/file-symlink/',
- b'/test/latest/file-symlink/.',
- b'/test/latest/file-symlink/..',
- b'/test/latest/file-symlink/../',
- b'/test/latest/file-symlink/../.',
- b'/test/latest/file-symlink/../..'):
- wvstart('resolve nofollow: ' + path_msg(path))
- vfs.clear_cache()
- try:
- resolve(path, follow=False)
- except vfs.IOError as res_ex:
- wvpasseq(ENOTDIR, res_ex.errno)
- wvpasseq([b'', b'test', save_time_str, b'file'],
- [name for name, item in res_ex.terminus])
-
- wvstart('resolve: non-directory parent')
+ for path in (b'/test/latest/file/',
+ b'/test/latest/file/.',
+ b'/test/latest/file/..',
+ b'/test/latest/file/../',
+ b'/test/latest/file/../.',
+ b'/test/latest/file/../..',
+ b'/test/latest/file/foo'):
vfs.clear_cache()
- file_res = resolve(b'/test/latest/file')
try:
- resolve(b'foo', parent=file_res)
+ resolve(path)
except vfs.IOError as res_ex:
wvpasseq(ENOTDIR, res_ex.errno)
- wvpasseq(None, res_ex.terminus)
+ wvpasseq([b'', b'test', save_time_str, b'file'],
+ [name for name, item in res_ex.terminus])
- wvstart('resolve nofollow: /test/latest/dir-symlink')
+ for path in (b'/test/latest/file-symlink/',
+ b'/test/latest/file-symlink/.',
+ b'/test/latest/file-symlink/..',
+ b'/test/latest/file-symlink/../',
+ b'/test/latest/file-symlink/../.',
+ b'/test/latest/file-symlink/../..'):
vfs.clear_cache()
- res = resolve(b'/test/latest/dir-symlink', follow=False)
- wvpasseq(4, len(res))
- dir_symlink_value = tip_tree[b'dir-symlink']
- expected_dir_symlink_item_w_meta = vfs.Item(meta=dir_symlink_value.meta,
- oid=dir_symlink_value.oid)
- expected = ((b'', vfs._root),
- (b'test', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta),
- (b'dir-symlink', expected_dir_symlink_item_w_meta))
- wvpasseq(expected, res)
+ try:
+ resolve(path, follow=False)
+ except vfs.IOError as res_ex:
+ wvpasseq(ENOTDIR, res_ex.errno)
+ wvpasseq([b'', b'test', save_time_str, b'file'],
+ [name for name, item in res_ex.terminus])
- dir_value = tip_tree[b'dir']
- expected_dir_item = vfs.Item(oid=dir_value.oid,
- meta=tree_dict(repo, dir_value.oid)[b'.'].meta)
- expected = ((b'', vfs._root),
- (b'test', test_revlist_w_meta),
- (save_time_str, expected_latest_item_w_meta),
- (b'dir', expected_dir_item))
- def lresolve(*args, **keys):
- return resolve(*args, **dict(keys, follow=False))
- for resname, resolver in (('resolve', resolve),
- ('resolve nofollow', lresolve)):
- for path in (b'/test/latest/dir-symlink/',
- b'/test/latest/dir-symlink/.'):
- wvstart(resname + ': ' + path_msg(path))
- vfs.clear_cache()
- res = resolver(path)
- wvpasseq(4, len(res))
- wvpasseq(expected, res)
- wvstart('resolve: /test/latest/dir-symlink')
- vfs.clear_cache()
- res = resolve(path)
- wvpasseq(4, len(res))
- wvpasseq(expected, res)
+ vfs.clear_cache()
+ file_res = resolve(b'/test/latest/file')
+ try:
+ resolve(b'foo', parent=file_res)
+ except vfs.IOError as res_ex:
+ wvpasseq(ENOTDIR, res_ex.errno)
+ wvpasseq(None, res_ex.terminus)
+
+ vfs.clear_cache()
+ res = resolve(b'/test/latest/dir-symlink', follow=False)
+ wvpasseq(4, len(res))
+ dir_symlink_value = tip_tree[b'dir-symlink']
+ expected_dir_symlink_item_w_meta = vfs.Item(meta=dir_symlink_value.meta,
+ oid=dir_symlink_value.oid)
+ expected = ((b'', vfs._root),
+ (b'test', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
+ (b'dir-symlink', expected_dir_symlink_item_w_meta))
+ wvpasseq(expected, res)
+
+ dir_value = tip_tree[b'dir']
+ expected_dir_item = vfs.Item(oid=dir_value.oid,
+ meta=tree_dict(repo, dir_value.oid)[b'.'].meta)
+ expected = ((b'', vfs._root),
+ (b'test', test_revlist_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
+ (b'dir', expected_dir_item))
+ def lresolve(*args, **keys):
+ return resolve(*args, **dict(keys, follow=False))
+ for resname, resolver in (('resolve', resolve),
+ ('resolve nofollow', lresolve)):
+ for path in (b'/test/latest/dir-symlink/',
+ b'/test/latest/dir-symlink/.'):
+ vfs.clear_cache()
+ res = resolver(path)
+ wvpasseq(4, len(res))
+ wvpasseq(expected, res)
+ vfs.clear_cache()
+ res = resolve(path)
+ wvpasseq(4, len(res))
+ wvpasseq(expected, res)
-@wvtest
-def test_local_resolve():
- prep_and_test_repo(b'local-vfs-resolve',
- lambda x: LocalRepo(repo_dir=x), test_resolve)
+def test_local_resolve(tmpdir):
+ prep_and_test_repo(tmpdir,
+ lambda x: LocalRepo(repo_dir=x), _test_resolve)
-@wvtest
-def test_remote_resolve():
- prep_and_test_repo(b'remote-vfs-resolve',
- lambda x: RemoteRepo(x), test_resolve)
+def test_remote_resolve(tmpdir):
+ prep_and_test_repo(tmpdir,
+ lambda x: RemoteRepo(x), _test_resolve)
-def test_resolve_loop(repo, tmpdir):
+def _test_resolve_loop(repo, tmpdir):
data_path = tmpdir + b'/src'
os.mkdir(data_path)
symlink(b'loop', data_path + b'/loop')
wvpasseq([b'', b'test', save_name, b'loop'],
[name for name, item in res_ex.terminus])
-@wvtest
-def test_local_resolve_loop():
- prep_and_test_repo(b'local-vfs-resolve-loop',
- lambda x: LocalRepo(x), test_resolve_loop)
+def test_local_resolve_loop(tmpdir):
+ prep_and_test_repo(tmpdir,
+ lambda x: LocalRepo(x), _test_resolve_loop)
-@wvtest
-def test_remote_resolve_loop():
- prep_and_test_repo(b'remote-vfs-resolve-loop',
- lambda x: RemoteRepo(x), test_resolve_loop)
+def test_remote_resolve_loop(tmpdir):
+ prep_and_test_repo(tmpdir,
+ lambda x: RemoteRepo(x), _test_resolve_loop)
# FIXME: add tests for the want_meta=False cases.
from __future__ import absolute_import
-from wvtest import *
+from wvpytest 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():
- with no_lingering_errors():
- WVPASSEQ(qst(b""" this is basic \t\n\r text """),
- [b'this', b'is', b'basic', b'text'])
- WVPASSEQ(qst(br""" \"x\" "help" 'yelp' """), [b'"x"', b'help', b'yelp'])
- WVPASSEQ(qst(br""" "'\"\"'" '\"\'' """), [b"'\"\"'", b'\\"\''])
-
- WVPASSEQ(shquote.quotesplit(b' this is "unfinished'),
- [(2, b'this'), (7, b'is'), (10, b'unfinished')])
-
- WVPASSEQ(shquote.quotesplit(b'"silly"\'will'),
- [(0, b'silly'), (7, b'will')])
-
- WVPASSEQ(shquote.unfinished_word(b'this is a "billy" "goat'),
- (b'"', b'goat'))
- WVPASSEQ(shquote.unfinished_word(b"'x"),
- (b"'", b'x'))
- WVPASSEQ(shquote.unfinished_word(b"abra cadabra "),
- (None, b''))
- WVPASSEQ(shquote.unfinished_word(b"abra cadabra"),
- (None, b'cadabra'))
-
- qtype, word = shquote.unfinished_word(b"this is /usr/loc")
- WVPASSEQ(shquote.what_to_add(qtype, word, b"/usr/local", True),
- b"al")
- qtype, word = shquote.unfinished_word(b"this is '/usr/loc")
- WVPASSEQ(shquote.what_to_add(qtype, word, b"/usr/local", True),
- b"al'")
- qtype, word = shquote.unfinished_word(b"this is \"/usr/loc")
- WVPASSEQ(shquote.what_to_add(qtype, word, b"/usr/local", True),
- b"al\"")
- qtype, word = shquote.unfinished_word(b"this is \"/usr/loc")
- WVPASSEQ(shquote.what_to_add(qtype, word, b"/usr/local", False),
- b"al")
- qtype, word = shquote.unfinished_word(b"this is \\ hammer\\ \"")
- WVPASSEQ(word, b' hammer "')
- WVPASSEQ(shquote.what_to_add(qtype, word, b" hammer \"time\"", True),
- b"time\\\"")
-
- WVPASSEQ(shquote.quotify_list([b'a', b'', b'"word"', b"'third'", b"'",
- b"x y"]),
- b"a '' '\"word\"' \"'third'\" \"'\" 'x y'")
+ WVPASSEQ(qst(b""" this is basic \t\n\r text """),
+ [b'this', b'is', b'basic', b'text'])
+ WVPASSEQ(qst(br""" \"x\" "help" 'yelp' """), [b'"x"', b'help', b'yelp'])
+ WVPASSEQ(qst(br""" "'\"\"'" '\"\'' """), [b"'\"\"'", b'\\"\''])
+
+ WVPASSEQ(shquote.quotesplit(b' this is "unfinished'),
+ [(2, b'this'), (7, b'is'), (10, b'unfinished')])
+
+ WVPASSEQ(shquote.quotesplit(b'"silly"\'will'),
+ [(0, b'silly'), (7, b'will')])
+
+ WVPASSEQ(shquote.unfinished_word(b'this is a "billy" "goat'),
+ (b'"', b'goat'))
+ WVPASSEQ(shquote.unfinished_word(b"'x"),
+ (b"'", b'x'))
+ WVPASSEQ(shquote.unfinished_word(b"abra cadabra "),
+ (None, b''))
+ WVPASSEQ(shquote.unfinished_word(b"abra cadabra"),
+ (None, b'cadabra'))
+
+ qtype, word = shquote.unfinished_word(b"this is /usr/loc")
+ WVPASSEQ(shquote.what_to_add(qtype, word, b"/usr/local", True),
+ b"al")
+ qtype, word = shquote.unfinished_word(b"this is '/usr/loc")
+ WVPASSEQ(shquote.what_to_add(qtype, word, b"/usr/local", True),
+ b"al'")
+ qtype, word = shquote.unfinished_word(b"this is \"/usr/loc")
+ WVPASSEQ(shquote.what_to_add(qtype, word, b"/usr/local", True),
+ b"al\"")
+ qtype, word = shquote.unfinished_word(b"this is \"/usr/loc")
+ WVPASSEQ(shquote.what_to_add(qtype, word, b"/usr/local", False),
+ b"al")
+ qtype, word = shquote.unfinished_word(b"this is \\ hammer\\ \"")
+ WVPASSEQ(word, b' hammer "')
+ WVPASSEQ(shquote.what_to_add(qtype, word, b" hammer \"time\"", True),
+ b"time\\\"")
+
+ WVPASSEQ(shquote.quotify_list([b'a', b'', b'"word"', b"'third'", b"'",
+ b"x y"]),
+ b"a '' '\"word\"' \"'third'\" \"'\" 'x y'")
from random import Random, randint
from stat import S_IFDIR, S_IFLNK, S_IFREG, S_ISDIR, S_ISREG
from sys import stderr
+import os
+import sys
from time import localtime, strftime, tzset
-from wvtest import *
+from wvpytest import *
from bup._helpers import write_random
from bup import git, metadata, vfs
from bup.helpers import exc, shstr
from bup.metadata import Metadata
from bup.repo import LocalRepo
-from buptest import ex, exo, no_lingering_errors, test_tempdir
+from buptest import ex, exo
from buptest.vfs import tree_dict
-top_dir = b'../..'
+lib_t_dir = os.path.dirname(fsencode(__file__))
+top_dir = os.path.join(lib_t_dir, b'../..')
bup_path = top_dir + b'/bup'
-start_dir = os.getcwd()
def ex(cmd, **kwargs):
print(shstr(cmd), file=stderr)
return exc(cmd, **kwargs)
-@wvtest
def test_default_modes():
wvpasseq(S_IFREG | 0o644, vfs.default_file_mode)
wvpasseq(S_IFDIR | 0o755, vfs.default_dir_mode)
wvpasseq(S_IFLNK | 0o755, vfs.default_symlink_mode)
-@wvtest
def test_cache_behavior():
orig_max = vfs._cache_max_items
try:
wvpasseq(len(link_target), augmented.meta.size)
-@wvtest
def test_item_mode():
- with no_lingering_errors():
- mode = S_IFDIR | 0o755
- meta = metadata.from_path(b'.')
- oid = b'\0' * 20
- wvpasseq(mode, vfs.item_mode(vfs.Item(oid=oid, meta=mode)))
- wvpasseq(meta.mode, vfs.item_mode(vfs.Item(oid=oid, meta=meta)))
-
-@wvtest
+ mode = S_IFDIR | 0o755
+ meta = metadata.from_path(b'.')
+ oid = b'\0' * 20
+ wvpasseq(mode, vfs.item_mode(vfs.Item(oid=oid, meta=mode)))
+ wvpasseq(meta.mode, vfs.item_mode(vfs.Item(oid=oid, meta=meta)))
+
def test_reverse_suffix_duplicates():
suffix = lambda x: tuple(vfs._reverse_suffix_duplicates(x))
wvpasseq((b'x',), suffix((b'x',)))
wvpasseq((b'x', b'y-1', b'y-0'), suffix((b'x', b'y', b'y')))
wvpasseq((b'x', b'y-1', b'y-0', b'z'), suffix((b'x', b'y', b'y', b'z')))
-@wvtest
-def test_misc():
- with no_lingering_errors():
- with test_tempdir(b'bup-tvfs-') as tmpdir:
- bup_dir = tmpdir + b'/bup'
- environ[b'GIT_DIR'] = bup_dir
- environ[b'BUP_DIR'] = bup_dir
- git.repodir = bup_dir
- data_path = tmpdir + b'/src'
- os.mkdir(data_path)
- with open(data_path + b'/file', 'wb+') as tmpfile:
- tmpfile.write(b'canary\n')
- symlink(b'file', data_path + b'/symlink')
- ex((bup_path, b'init'))
- ex((bup_path, b'index', b'-v', data_path))
- ex((bup_path, b'save', b'-d', b'100000', b'-tvvn', b'test',
- b'--strip', data_path))
- repo = LocalRepo()
-
- wvstart('readlink')
- ls_tree = exo((b'git', b'ls-tree', b'test', b'symlink')).out
- mode, typ, oidx, name = ls_tree.strip().split(None, 3)
- assert name == b'symlink'
- link_item = vfs.Item(oid=unhexlify(oidx), meta=int(mode, 8))
- wvpasseq(b'file', vfs.readlink(repo, link_item))
-
- ls_tree = exo((b'git', b'ls-tree', b'test', b'file')).out
- mode, typ, oidx, name = ls_tree.strip().split(None, 3)
- assert name == b'file'
- file_item = vfs.Item(oid=unhexlify(oidx), meta=int(mode, 8))
- wvexcept(Exception, vfs.readlink, repo, file_item)
-
- wvstart('item_size')
- wvpasseq(4, vfs.item_size(repo, link_item))
- wvpasseq(7, vfs.item_size(repo, file_item))
- meta = metadata.from_path(fsencode(__file__))
- meta.size = 42
- fake_item = file_item._replace(meta=meta)
- wvpasseq(42, vfs.item_size(repo, fake_item))
-
- _, fakelink_item = vfs.resolve(repo, b'/test/latest', follow=False)[-1]
- wvpasseq(17, vfs.item_size(repo, fakelink_item))
-
- wvstart('augment_item_meta')
- run_augment_item_meta_tests(repo,
- b'/test/latest/file', 7,
- b'/test/latest/symlink', b'file')
-
- wvstart('copy_item')
- # FIXME: this caused StopIteration
- #_, file_item = vfs.resolve(repo, '/file')[-1]
- _, file_item = vfs.resolve(repo, b'/test/latest/file')[-1]
- file_copy = vfs.copy_item(file_item)
- wvpass(file_copy is not file_item)
- wvpass(file_copy.meta is not file_item.meta)
- wvpass(isinstance(file_copy, tuple))
- wvpass(file_item.meta.user)
- wvpass(file_copy.meta.user)
- file_copy.meta.user = None
- wvpass(file_item.meta.user)
+def test_misc(tmpdir):
+ bup_dir = tmpdir + b'/bup'
+ environ[b'GIT_DIR'] = bup_dir
+ environ[b'BUP_DIR'] = bup_dir
+ git.repodir = bup_dir
+ data_path = tmpdir + b'/src'
+ os.mkdir(data_path)
+ with open(data_path + b'/file', 'wb+') as tmpfile:
+ tmpfile.write(b'canary\n')
+ symlink(b'file', data_path + b'/symlink')
+ ex((bup_path, b'init'))
+ ex((bup_path, b'index', b'-v', data_path))
+ ex((bup_path, b'save', b'-d', b'100000', b'-tvvn', b'test',
+ b'--strip', data_path))
+ repo = LocalRepo()
+
+ ls_tree = exo((b'git', b'ls-tree', b'test', b'symlink')).out
+ mode, typ, oidx, name = ls_tree.strip().split(None, 3)
+ assert name == b'symlink'
+ link_item = vfs.Item(oid=unhexlify(oidx), meta=int(mode, 8))
+ wvpasseq(b'file', vfs.readlink(repo, link_item))
+
+ ls_tree = exo((b'git', b'ls-tree', b'test', b'file')).out
+ mode, typ, oidx, name = ls_tree.strip().split(None, 3)
+ assert name == b'file'
+ file_item = vfs.Item(oid=unhexlify(oidx), meta=int(mode, 8))
+ wvexcept(Exception, vfs.readlink, repo, file_item)
+
+ wvpasseq(4, vfs.item_size(repo, link_item))
+ wvpasseq(7, vfs.item_size(repo, file_item))
+ meta = metadata.from_path(fsencode(__file__))
+ meta.size = 42
+ fake_item = file_item._replace(meta=meta)
+ wvpasseq(42, vfs.item_size(repo, fake_item))
+
+ _, fakelink_item = vfs.resolve(repo, b'/test/latest', follow=False)[-1]
+ wvpasseq(17, vfs.item_size(repo, fakelink_item))
+
+ run_augment_item_meta_tests(repo,
+ b'/test/latest/file', 7,
+ b'/test/latest/symlink', b'file')
+
+ # FIXME: this caused StopIteration
+ #_, file_item = vfs.resolve(repo, '/file')[-1]
+ _, file_item = vfs.resolve(repo, b'/test/latest/file')[-1]
+ file_copy = vfs.copy_item(file_item)
+ wvpass(file_copy is not file_item)
+ wvpass(file_copy.meta is not file_item.meta)
+ wvpass(isinstance(file_copy, tuple))
+ wvpass(file_item.meta.user)
+ wvpass(file_copy.meta.user)
+ file_copy.meta.user = None
+ wvpass(file_item.meta.user)
def write_sized_random_content(parent_dir, size, seed):
verbose = 0
wvpasseq(b'', ex_buf)
wvpasseq(b'', act_buf)
-@wvtest
-def test_read_and_seek():
+def test_read_and_seek(tmpdir):
# Write a set of randomly sized files containing random data whose
# names are their sizes, and then verify that what we get back
# from the vfs when seeking and reading with various block sizes
# matches the original content.
- with no_lingering_errors():
- with test_tempdir(b'bup-tvfs-read-') as tmpdir:
- resolve = vfs.resolve
- bup_dir = tmpdir + b'/bup'
- environ[b'GIT_DIR'] = bup_dir
- environ[b'BUP_DIR'] = bup_dir
- git.repodir = bup_dir
- repo = LocalRepo()
- data_path = tmpdir + b'/src'
- os.mkdir(data_path)
- seed = randint(-(1 << 31), (1 << 31) - 1)
- rand = Random()
- rand.seed(seed)
- print('test_read seed:', seed, file=sys.stderr)
- max_size = 2 * 1024 * 1024
- sizes = set((rand.randint(1, max_size) for _ in range(5)))
- sizes.add(1)
- sizes.add(max_size)
- for size in sizes:
- write_sized_random_content(data_path, size, seed)
- ex((bup_path, b'init'))
- ex((bup_path, b'index', b'-v', data_path))
- ex((bup_path, b'save', b'-d', b'100000', b'-tvvn', b'test',
- b'--strip', data_path))
- read_sizes = set((rand.randint(1, max_size) for _ in range(10)))
- sizes.add(1)
- sizes.add(max_size)
- print('test_read src sizes:', sizes, file=sys.stderr)
- print('test_read read sizes:', read_sizes, file=sys.stderr)
- for size in sizes:
- res = resolve(repo, b'/test/latest/' + str(size).encode('ascii'))
- _, item = res[-1]
- wvpasseq(size, vfs.item_size(repo, res[-1][1]))
- validate_vfs_streaming_read(repo, item,
- b'%s/%d' % (data_path, size),
- read_sizes)
- validate_vfs_seeking_read(repo, item,
- b'%s/%d' % (data_path, size),
- read_sizes)
-
-@wvtest
-def test_contents_with_mismatched_bupm_git_ordering():
- with no_lingering_errors():
- with test_tempdir(b'bup-tvfs-') as tmpdir:
- bup_dir = tmpdir + b'/bup'
- environ[b'GIT_DIR'] = bup_dir
- environ[b'BUP_DIR'] = bup_dir
- git.repodir = bup_dir
- data_path = tmpdir + b'/src'
- os.mkdir(data_path)
- os.mkdir(data_path + b'/foo')
- with open(data_path + b'/foo.', 'wb+') as tmpfile:
- tmpfile.write(b'canary\n')
- ex((bup_path, b'init'))
- ex((bup_path, b'index', b'-v', data_path))
- save_utc = 100000
- save_name = strftime('%Y-%m-%d-%H%M%S', localtime(save_utc)).encode('ascii')
- ex((bup_path, b'save', b'-tvvn', b'test', b'-d', b'%d' % save_utc,
- b'--strip', data_path))
- repo = LocalRepo()
- tip_sref = exo((b'git', b'show-ref', b'refs/heads/test')).out
- tip_oidx = tip_sref.strip().split()[0]
- tip_tree_oidx = exo((b'git', b'log', b'--pretty=%T', b'-n1',
- tip_oidx)).out.strip()
- tip_tree_oid = unhexlify(tip_tree_oidx)
- tip_tree = tree_dict(repo, tip_tree_oid)
-
- name, item = vfs.resolve(repo, b'/test/latest')[2]
- wvpasseq(save_name, name)
- expected = frozenset((x.name, vfs.Item(oid=x.oid, meta=x.meta))
- for x in (tip_tree[name]
- for name in (b'.', b'foo', b'foo.')))
- contents = tuple(vfs.contents(repo, item))
- wvpasseq(expected, frozenset(contents))
- # Spot check, in case tree_dict shares too much code with the vfs
- name, item = next(((n, i) for n, i in contents if n == b'foo'))
- wvpass(S_ISDIR(item.meta))
- name, item = next(((n, i) for n, i in contents if n == b'foo.'))
- wvpass(S_ISREG(item.meta.mode))
-
-@wvtest
-def test_duplicate_save_dates():
- with no_lingering_errors():
- with test_tempdir(b'bup-tvfs-') as tmpdir:
- bup_dir = tmpdir + b'/bup'
- environ[b'GIT_DIR'] = bup_dir
- environ[b'BUP_DIR'] = bup_dir
- environ[b'TZ'] = b'UTC'
- tzset()
- git.repodir = bup_dir
- data_path = tmpdir + b'/src'
- os.mkdir(data_path)
- with open(data_path + b'/file', 'wb+') as tmpfile:
- tmpfile.write(b'canary\n')
- ex((b'env',))
- ex((bup_path, b'init'))
- ex((bup_path, b'index', b'-v', data_path))
- for i in range(11):
- ex((bup_path, b'save', b'-d', b'100000', b'-n', b'test',
- data_path))
- repo = LocalRepo()
- res = vfs.resolve(repo, b'/test')
- wvpasseq(2, len(res))
- name, revlist = res[-1]
- wvpasseq(b'test', name)
- wvpasseq((b'.',
- b'1970-01-02-034640-00',
- b'1970-01-02-034640-01',
- b'1970-01-02-034640-02',
- b'1970-01-02-034640-03',
- b'1970-01-02-034640-04',
- b'1970-01-02-034640-05',
- b'1970-01-02-034640-06',
- b'1970-01-02-034640-07',
- b'1970-01-02-034640-08',
- b'1970-01-02-034640-09',
- b'1970-01-02-034640-10',
- b'latest'),
- tuple(sorted(x[0] for x in vfs.contents(repo, revlist))))
-
-@wvtest
+ resolve = vfs.resolve
+ bup_dir = tmpdir + b'/bup'
+ environ[b'GIT_DIR'] = bup_dir
+ environ[b'BUP_DIR'] = bup_dir
+ git.repodir = bup_dir
+ repo = LocalRepo()
+ data_path = tmpdir + b'/src'
+ os.mkdir(data_path)
+ seed = randint(-(1 << 31), (1 << 31) - 1)
+ rand = Random()
+ rand.seed(seed)
+ print('test_read seed:', seed, file=sys.stderr)
+ max_size = 2 * 1024 * 1024
+ sizes = set((rand.randint(1, max_size) for _ in range(5)))
+ sizes.add(1)
+ sizes.add(max_size)
+ for size in sizes:
+ write_sized_random_content(data_path, size, seed)
+ ex((bup_path, b'init'))
+ ex((bup_path, b'index', b'-v', data_path))
+ ex((bup_path, b'save', b'-d', b'100000', b'-tvvn', b'test',
+ b'--strip', data_path))
+ read_sizes = set((rand.randint(1, max_size) for _ in range(10)))
+ sizes.add(1)
+ sizes.add(max_size)
+ print('test_read src sizes:', sizes, file=sys.stderr)
+ print('test_read read sizes:', read_sizes, file=sys.stderr)
+ for size in sizes:
+ res = resolve(repo, b'/test/latest/' + str(size).encode('ascii'))
+ _, item = res[-1]
+ wvpasseq(size, vfs.item_size(repo, res[-1][1]))
+ validate_vfs_streaming_read(repo, item,
+ b'%s/%d' % (data_path, size),
+ read_sizes)
+ validate_vfs_seeking_read(repo, item,
+ b'%s/%d' % (data_path, size),
+ read_sizes)
+
+def test_contents_with_mismatched_bupm_git_ordering(tmpdir):
+ bup_dir = tmpdir + b'/bup'
+ environ[b'GIT_DIR'] = bup_dir
+ environ[b'BUP_DIR'] = bup_dir
+ git.repodir = bup_dir
+ data_path = tmpdir + b'/src'
+ os.mkdir(data_path)
+ os.mkdir(data_path + b'/foo')
+ with open(data_path + b'/foo.', 'wb+') as tmpfile:
+ tmpfile.write(b'canary\n')
+ ex((bup_path, b'init'))
+ ex((bup_path, b'index', b'-v', data_path))
+ save_utc = 100000
+ save_name = strftime('%Y-%m-%d-%H%M%S', localtime(save_utc)).encode('ascii')
+ ex((bup_path, b'save', b'-tvvn', b'test', b'-d', b'%d' % save_utc,
+ b'--strip', data_path))
+ repo = LocalRepo()
+ tip_sref = exo((b'git', b'show-ref', b'refs/heads/test')).out
+ tip_oidx = tip_sref.strip().split()[0]
+ tip_tree_oidx = exo((b'git', b'log', b'--pretty=%T', b'-n1',
+ tip_oidx)).out.strip()
+ tip_tree_oid = unhexlify(tip_tree_oidx)
+ tip_tree = tree_dict(repo, tip_tree_oid)
+
+ name, item = vfs.resolve(repo, b'/test/latest')[2]
+ wvpasseq(save_name, name)
+ expected = frozenset((x.name, vfs.Item(oid=x.oid, meta=x.meta))
+ for x in (tip_tree[name]
+ for name in (b'.', b'foo', b'foo.')))
+ contents = tuple(vfs.contents(repo, item))
+ wvpasseq(expected, frozenset(contents))
+ # Spot check, in case tree_dict shares too much code with the vfs
+ name, item = next(((n, i) for n, i in contents if n == b'foo'))
+ wvpass(S_ISDIR(item.meta))
+ name, item = next(((n, i) for n, i in contents if n == b'foo.'))
+ wvpass(S_ISREG(item.meta.mode))
+
+def test_duplicate_save_dates(tmpdir):
+ bup_dir = tmpdir + b'/bup'
+ environ[b'GIT_DIR'] = bup_dir
+ environ[b'BUP_DIR'] = bup_dir
+ environ[b'TZ'] = b'UTC'
+ tzset()
+ git.repodir = bup_dir
+ data_path = tmpdir + b'/src'
+ os.mkdir(data_path)
+ with open(data_path + b'/file', 'wb+') as tmpfile:
+ tmpfile.write(b'canary\n')
+ ex((b'env',))
+ ex((bup_path, b'init'))
+ ex((bup_path, b'index', b'-v', data_path))
+ for i in range(11):
+ ex((bup_path, b'save', b'-d', b'100000', b'-n', b'test',
+ data_path))
+ repo = LocalRepo()
+ res = vfs.resolve(repo, b'/test')
+ wvpasseq(2, len(res))
+ name, revlist = res[-1]
+ wvpasseq(b'test', name)
+ wvpasseq((b'.',
+ b'1970-01-02-034640-00',
+ b'1970-01-02-034640-01',
+ b'1970-01-02-034640-02',
+ b'1970-01-02-034640-03',
+ b'1970-01-02-034640-04',
+ b'1970-01-02-034640-05',
+ b'1970-01-02-034640-06',
+ b'1970-01-02-034640-07',
+ b'1970-01-02-034640-08',
+ b'1970-01-02-034640-09',
+ b'1970-01-02-034640-10',
+ b'latest'),
+ tuple(sorted(x[0] for x in vfs.contents(repo, revlist))))
+
def test_item_read_write():
- with no_lingering_errors():
- x = vfs.Root(meta=13)
- stream = BytesIO()
- vfs.write_item(stream, x)
- print('stream:', repr(stream.getvalue()), stream.tell(), file=sys.stderr)
- stream.seek(0)
- wvpasseq(x, vfs.read_item(stream))
+ x = vfs.Root(meta=13)
+ stream = BytesIO()
+ vfs.write_item(stream, x)
+ print('stream:', repr(stream.getvalue()), stream.tell(), file=sys.stderr)
+ stream.seek(0)
+ wvpasseq(x, vfs.read_item(stream))
from __future__ import absolute_import
from io import BytesIO
-from wvtest import *
+from wvpytest import *
from bup import vint
-from buptest import no_lingering_errors
def encode_and_decode_vuint(x):
return vint.read_vuint(BytesIO(f.getvalue()))
-@wvtest
def test_vuint():
- 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)
return vint.read_vint(BytesIO(f.getvalue()))
-@wvtest
def test_vint():
- 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())
- WVEXCEPT(EOFError, vint.read_vint, BytesIO(b"\x80\x80"))
+ 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())
+ WVEXCEPT(EOFError, vint.read_vint, BytesIO(b"\x80\x80"))
def encode_and_decode_bvec(x):
return vint.read_bvec(BytesIO(f.getvalue()))
-@wvtest
def test_bvec():
- with no_lingering_errors():
- values = (b'', b'x', b'foo', b'\0', b'\0foo', b'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 (b'foo', b'bar', b'baz', b'bax'):
- vint.write_bvec(outf, x)
- inf = BytesIO(outf.getvalue())
- WVPASSEQ(vint.read_bvec(inf), b'foo')
- WVPASSEQ(vint.read_bvec(inf), b'bar')
- vint.skip_bvec(inf)
- WVPASSEQ(vint.read_bvec(inf), b'bax')
+ values = (b'', b'x', b'foo', b'\0', b'\0foo', b'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 (b'foo', b'bar', b'baz', b'bax'):
+ vint.write_bvec(outf, x)
+ inf = BytesIO(outf.getvalue())
+ WVPASSEQ(vint.read_bvec(inf), b'foo')
+ WVPASSEQ(vint.read_bvec(inf), b'bar')
+ vint.skip_bvec(inf)
+ WVPASSEQ(vint.read_bvec(inf), b'bax')
def pack_and_unpack(types, *values):
return vint.unpack(types, data)
-@wvtest
def test_pack_and_unpack():
- with no_lingering_errors():
- tests = [('', []),
- ('s', [b'foo']),
- ('ss', [b'foo', b'bar']),
- ('sV', [b'foo', 0]),
- ('sv', [b'foo', -1]),
- ('V', [0]),
- ('Vs', [0, b'foo']),
- ('VV', [0, 1]),
- ('Vv', [0, -1]),
- ('v', [0]),
- ('vs', [0, b'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', '')
+ tests = [('', []),
+ ('s', [b'foo']),
+ ('ss', [b'foo', b'bar']),
+ ('sV', [b'foo', 0]),
+ ('sv', [b'foo', -1]),
+ ('V', [0]),
+ ('Vs', [0, b'foo']),
+ ('VV', [0, 1]),
+ ('Vv', [0, -1]),
+ ('v', [0]),
+ ('vs', [0, b'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', '')
from __future__ import absolute_import
import math, tempfile, subprocess
-from wvtest import *
+from wvpytest import *
import bup._helpers as _helpers
from bup import xstat
-from buptest import no_lingering_errors, test_tempdir
-@wvtest
def test_fstime():
- 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.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():
+def test_bup_utimensat(tmpdir):
if not xstat._bup_utimensat:
return
- with no_lingering_errors():
- with test_tempdir(b'bup-txstat-') as tmpdir:
- path = tmpdir + b'/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])
+ path = tmpdir + b'/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():
+def test_bup_utimes(tmpdir):
if not xstat._bup_utimes:
return
- with no_lingering_errors():
- with test_tempdir(b'bup-txstat-') as tmpdir:
- path = tmpdir + b'/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)
+ path = tmpdir + b'/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():
+def test_bup_lutimes(tmpdir):
if not xstat._bup_lutimes:
return
- with no_lingering_errors():
- with test_tempdir(b'bup-txstat-') as tmpdir:
- path = tmpdir + b'/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)
+ path = tmpdir + b'/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)
from bup.io import byte_stream
-@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(fsencode(__file__))) + b'/test/tmp'
try:
--- /dev/null
+import pytest
+
+def WVPASS(cond = True):
+ assert cond
+
+def WVFAIL(cond = True):
+ assert not cond
+
+def WVPASSEQ(a, b):
+ assert a == b
+
+def WVPASSNE(a, b):
+ assert a != b
+
+def WVPASSLT(a, b):
+ assert a < b
+
+def WVPASSLE(a, b):
+ assert a <= b
+
+def WVPASSGT(a, b):
+ assert a > b
+
+def WVPASSGE(a, b):
+ assert a >= b
+
+def WVEXCEPT(etype, func, *args, **kwargs):
+ with pytest.raises(etype):
+ func(*args, **kwargs)
+
+def WVCHECK(cond, msg):
+ assert cond, msg
+
+def WVMSG(msg):
+ print(msg)
+
+wvpass = WVPASS
+wvfail = WVFAIL
+wvpasseq = WVPASSEQ
+wvpassne = WVPASSNE
+wvpaslt = WVPASSLT
+wvpassle = WVPASSLE
+wvpassgt = WVPASSGT
+wvpassge = WVPASSGE
+wvexcept = WVEXCEPT
+wvcheck = WVCHECK
+wvmsg = WVMSG
+wvstart = WVMSG
projects / bup.git / commitdiff