X-Git-Url: https://arthur.barton.de/cgi-bin/gitweb.cgi?a=blobdiff_plain;f=lib%2Fbup%2Ft%2Fthelpers.py;h=e0ef21fd5098c25963b55ef7eb1858b99ac6d084;hb=c40b3dd5fd74e72024fbaad3daf5a958aefa1c54;hp=31ecbb9513c15a63b691ef80ab661b8d9859d53c;hpb=37f43d741eb44ae8f9e36f42770d2bb494999d92;p=bup.git diff --git a/lib/bup/t/thelpers.py b/lib/bup/t/thelpers.py index 31ecbb9..e0ef21f 100644 --- a/lib/bup/t/thelpers.py +++ b/lib/bup/t/thelpers.py @@ -1,84 +1,223 @@ -import math -import os -import bup._helpers as _helpers -from bup.helpers import * + +from __future__ import absolute_import +import helpers, math, os, os.path, stat, subprocess + from wvtest import * +from bup.helpers import (atomically_replaced_file, batchpipe, detect_fakeroot, + grafted_path_components, mkdirp, parse_num, + path_components, readpipe, stripped_path_components, + utc_offset_str) +from buptest import no_lingering_errors, test_tempdir +import bup._helpers as _helpers + + +bup_tmp = os.path.realpath('../../../t/tmp') +mkdirp(bup_tmp) + + @wvtest def test_parse_num(): - pn = parse_num - WVPASSEQ(pn('1'), 1) - WVPASSEQ(pn('0'), 0) - WVPASSEQ(pn('1.5k'), 1536) - WVPASSEQ(pn('2 gb'), 2*1024*1024*1024) - WVPASSEQ(pn('1e+9 k'), 1000000000 * 1024) - WVPASSEQ(pn('-3e-3mb'), int(-0.003 * 1024 * 1024)) + with no_lingering_errors(): + pn = parse_num + WVPASSEQ(pn('1'), 1) + WVPASSEQ(pn('0'), 0) + WVPASSEQ(pn('1.5k'), 1536) + WVPASSEQ(pn('2 gb'), 2*1024*1024*1024) + WVPASSEQ(pn('1e+9 k'), 1000000000 * 1024) + WVPASSEQ(pn('-3e-3mb'), int(-0.003 * 1024 * 1024)) @wvtest def test_detect_fakeroot(): - if os.getenv('FAKEROOTKEY'): - WVPASS(detect_fakeroot()) - else: - WVPASS(not detect_fakeroot()) + with no_lingering_errors(): + if os.getenv('FAKEROOTKEY'): + WVPASS(detect_fakeroot()) + else: + WVPASS(not detect_fakeroot()) + +@wvtest +def test_path_components(): + with no_lingering_errors(): + WVPASSEQ(path_components('/'), [('', '/')]) + WVPASSEQ(path_components('/foo'), [('', '/'), ('foo', '/foo')]) + WVPASSEQ(path_components('/foo/'), [('', '/'), ('foo', '/foo')]) + WVPASSEQ(path_components('/foo/bar'), + [('', '/'), ('foo', '/foo'), ('bar', '/foo/bar')]) + WVEXCEPT(Exception, path_components, 'foo') + + +@wvtest +def test_stripped_path_components(): + with no_lingering_errors(): + WVPASSEQ(stripped_path_components('/', []), [('', '/')]) + WVPASSEQ(stripped_path_components('/', ['']), [('', '/')]) + WVPASSEQ(stripped_path_components('/', ['/']), [('', '/')]) + WVPASSEQ(stripped_path_components('/foo', ['/']), + [('', '/'), ('foo', '/foo')]) + WVPASSEQ(stripped_path_components('/', ['/foo']), [('', '/')]) + WVPASSEQ(stripped_path_components('/foo', ['/bar']), + [('', '/'), ('foo', '/foo')]) + WVPASSEQ(stripped_path_components('/foo', ['/foo']), [('', '/foo')]) + WVPASSEQ(stripped_path_components('/foo/bar', ['/foo']), + [('', '/foo'), ('bar', '/foo/bar')]) + WVPASSEQ(stripped_path_components('/foo/bar', ['/bar', '/foo', '/baz']), + [('', '/foo'), ('bar', '/foo/bar')]) + WVPASSEQ(stripped_path_components('/foo/bar/baz', ['/foo/bar/baz']), + [('', '/foo/bar/baz')]) + WVEXCEPT(Exception, stripped_path_components, 'foo', []) + + +@wvtest +def test_grafted_path_components(): + with no_lingering_errors(): + WVPASSEQ(grafted_path_components([('/chroot', '/')], '/foo'), + [('', '/'), ('foo', '/foo')]) + WVPASSEQ(grafted_path_components([('/foo/bar', '/')], + '/foo/bar/baz/bax'), + [('', '/foo/bar'), + ('baz', '/foo/bar/baz'), + ('bax', '/foo/bar/baz/bax')]) + WVPASSEQ(grafted_path_components([('/foo/bar/baz', '/bax')], + '/foo/bar/baz/1/2'), + [('', None), + ('bax', '/foo/bar/baz'), + ('1', '/foo/bar/baz/1'), + ('2', '/foo/bar/baz/1/2')]) + WVPASSEQ(grafted_path_components([('/foo', '/bar/baz/bax')], + '/foo/bar'), + [('', None), + ('bar', None), + ('baz', None), + ('bax', '/foo'), + ('bar', '/foo/bar')]) + WVPASSEQ(grafted_path_components([('/foo/bar/baz', '/a/b/c')], + '/foo/bar/baz'), + [('', None), ('a', None), ('b', None), ('c', '/foo/bar/baz')]) + WVPASSEQ(grafted_path_components([('/', '/a/b/c/')], '/foo/bar'), + [('', None), ('a', None), ('b', None), ('c', '/'), + ('foo', '/foo'), ('bar', '/foo/bar')]) + WVEXCEPT(Exception, grafted_path_components, 'foo', []) + @wvtest -def test_strip_path(): - prefix = "/var/backup/daily.0/localhost" - empty_prefix = "" - non_matching_prefix = "/home" - path = "/var/backup/daily.0/localhost/etc/" +def test_readpipe(): + with no_lingering_errors(): + x = readpipe(['echo', '42']) + WVPASSEQ(x, '42\n') + try: + readpipe(['bash', '-c', 'exit 42']) + except Exception as ex: + WVPASSEQ(str(ex), + "subprocess 'bash -c exit 42' failed with status 42") - WVPASSEQ(strip_path(prefix, path), '/etc') - WVPASSEQ(strip_path(empty_prefix, path), path) - WVPASSEQ(strip_path(non_matching_prefix, path), path) - WVEXCEPT(Exception, strip_path, None, path) @wvtest -def test_strip_base_path(): - path = "/var/backup/daily.0/localhost/etc/" - base_paths = ["/var", "/var/backup", "/var/backup/daily.0/localhost"] - WVPASSEQ(strip_base_path(path, base_paths), '/etc') +def test_batchpipe(): + with no_lingering_errors(): + for chunk in batchpipe(['echo'], []): + WVPASS(False) + out = '' + for chunk in batchpipe(['echo'], ['42']): + out += chunk + WVPASSEQ(out, '42\n') + try: + batchpipe(['bash', '-c'], ['exit 42']) + except Exception as ex: + WVPASSEQ(str(ex), + "subprocess 'bash -c exit 42' failed with status 42") + args = [str(x) for x in range(6)] + # Force batchpipe to break the args into batches of 3. This + # approach assumes all args are the same length. + arg_max = \ + helpers._argmax_base(['echo']) + helpers._argmax_args_size(args[:3]) + batches = batchpipe(['echo'], args, arg_max=arg_max) + WVPASSEQ(next(batches), '0 1 2\n') + WVPASSEQ(next(batches), '3 4 5\n') + WVPASSEQ(next(batches, None), None) + batches = batchpipe(['echo'], [str(x) for x in range(5)], arg_max=arg_max) + WVPASSEQ(next(batches), '0 1 2\n') + WVPASSEQ(next(batches), '3 4\n') + WVPASSEQ(next(batches, None), None) + @wvtest -def test_strip_symlinked_base_path(): - tmpdir = os.path.join(os.getcwd(),"test_strip_symlinked_base_path.tmp") - symlink_src = os.path.join(tmpdir, "private", "var") - symlink_dst = os.path.join(tmpdir, "var") - path = os.path.join(symlink_dst, "a") +def test_atomically_replaced_file(): + with no_lingering_errors(): + with test_tempdir('bup-thelper-') as tmpdir: + target_file = os.path.join(tmpdir, 'test-atomic-write') - os.mkdir(tmpdir) - os.mkdir(os.path.join(tmpdir, "private")) - os.mkdir(symlink_src) - os.symlink(symlink_src, symlink_dst) + 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') - result = strip_base_path(path, [symlink_dst]) + 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') - os.remove(symlink_dst) - os.rmdir(symlink_src) - os.rmdir(os.path.join(tmpdir, "private")) - os.rmdir(tmpdir) + with atomically_replaced_file(target_file, mode='wb') as f: + f.write(os.urandom(20)) + WVPASSEQ(f.mode, 'wb') - WVPASSEQ(result, "/a") @wvtest -def test_graft_path(): - middle_matching_old_path = "/user" - non_matching_old_path = "/usr" - matching_old_path = "/home" - matching_full_path = "/home/user" - new_path = "/opt" - - all_graft_points = [(middle_matching_old_path, new_path), - (non_matching_old_path, new_path), - (matching_old_path, new_path)] - - path = "/home/user/" - - WVPASSEQ(graft_path([(middle_matching_old_path, new_path)], path), - "/home/user") - WVPASSEQ(graft_path([(non_matching_old_path, new_path)], path), - "/home/user") - WVPASSEQ(graft_path([(matching_old_path, new_path)], path), "/opt/user") - WVPASSEQ(graft_path(all_graft_points, path), "/opt/user") - WVPASSEQ(graft_path([(matching_full_path, new_path)], path), - "/opt") +def test_utc_offset_str(): + with no_lingering_errors(): + tz = os.environ.get('TZ') + try: + os.environ['TZ'] = 'FOO+0:00' + WVPASSEQ(utc_offset_str(0), '+0000') + os.environ['TZ'] = 'FOO+1:00' + WVPASSEQ(utc_offset_str(0), '-0100') + os.environ['TZ'] = 'FOO-1:00' + WVPASSEQ(utc_offset_str(0), '+0100') + os.environ['TZ'] = 'FOO+3:3' + WVPASSEQ(utc_offset_str(0), '-0303') + os.environ['TZ'] = 'FOO-3:3' + WVPASSEQ(utc_offset_str(0), '+0303') + # Offset is not an integer number of minutes + os.environ['TZ'] = 'FOO+3:3:3' + WVPASSEQ(utc_offset_str(1), '-0303') + os.environ['TZ'] = 'FOO-3:3:3' + WVPASSEQ(utc_offset_str(1), '+0303') + WVPASSEQ(utc_offset_str(314159), '+0303') + finally: + if tz: + os.environ['TZ'] = tz + else: + try: + del os.environ['TZ'] + except KeyError: + pass + +@wvtest +def test_valid_save_name(): + with no_lingering_errors(): + valid = helpers.valid_save_name + WVPASS(valid('x')) + WVPASS(valid('x@')) + WVFAIL(valid('@')) + WVFAIL(valid('/')) + WVFAIL(valid('/foo')) + WVFAIL(valid('foo/')) + WVFAIL(valid('/foo/')) + WVFAIL(valid('foo//bar')) + WVFAIL(valid('.')) + WVFAIL(valid('bar.')) + WVFAIL(valid('foo@{')) + for x in ' ~^:?*[\\': + WVFAIL(valid('foo' + x)) + for i in range(20): + WVFAIL(valid('foo' + chr(i))) + WVFAIL(valid('foo' + chr(0x7f))) + WVFAIL(valid('foo..bar')) + WVFAIL(valid('bar.lock/baz')) + WVFAIL(valid('foo/bar.lock/baz')) + WVFAIL(valid('.bar/baz')) + WVFAIL(valid('foo/.bar/baz'))