thelpers.py: use t/tmp as the tmpdir parent

[bup.git] / lib / bup / t / thelpers.py

diff --git a/lib/bup/t/thelpers.py b/lib/bup/t/thelpers.py
index 38596d7447e1e33c416cb399748506a2613e232a..97438142dcf71a01ed3fd5d5dc5ded141c89fcf9 100644 (file)
--- a/lib/bup/t/thelpers.py
+++ b/lib/bup/t/thelpers.py
@@ -1,11 +1,15 @@
-import config
 import helpers
 import math
 import os
+import os.path
+import tempfile
+import stat
 import bup._helpers as _helpers
 from bup.helpers import *
 from wvtest import *
 
+bup_tmp = os.path.realpath('../../../t/tmp')
+mkdirp(bup_tmp)
 
 @wvtest
 def test_next():
@@ -131,18 +135,36 @@ def test_batchpipe():
         batchpipe(['bash', '-c'], ['exit 42'])
     except Exception, ex:
         WVPASSEQ(str(ex), "subprocess 'bash -c exit 42' failed with status 42")
-    oldmax = config.arg_max
     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.
-    config.arg_max = \
+    arg_max = \
         helpers._argmax_base(['echo']) + helpers._argmax_args_size(args[:3])
-    batches = batchpipe(['echo'], args)
+    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)])
+    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)
-    config.arg_max = oldmax
+
+
+@wvtest
+def test_atomically_replaced_file():
+    tmpdir = tempfile.mkdtemp(dir=bup_tmp, prefix='bup-thelper-')
+    target_file = os.path.join(tmpdir, 'test-atomic-write')
+    initial_failures = wvfailure_count()
+
+    with atomically_replaced_file(target_file, mode='w') as f:
+        f.write('asdf')
+        WVPASSEQ(f.mode, 'w')
+    f = open(target_file, 'r')
+    WVPASSEQ(f.read(), 'asdf')
+
+    with atomically_replaced_file(target_file, mode='wb') as f:
+        f.write(os.urandom(20))
+        WVPASSEQ(f.mode, 'wb')
+
+    if wvfailure_count() == initial_failures:
+        subprocess.call(['rm', '-rf', tmpdir])