test-atomictempfile.py
125 lines
| 4.2 KiB
| text/x-python
|
PythonLexer
/ tests / test-atomictempfile.py
|
r29194 | from __future__ import absolute_import | ||
import glob | ||||
|
r14007 | import os | ||
|
r29391 | import shutil | ||
|
r36799 | import stat | ||
|
r29391 | import tempfile | ||
|
r18666 | import unittest | ||
|
r29194 | from mercurial import ( | ||
|
r36299 | pycompat, | ||
|
r29194 | util, | ||
) | ||||
atomictempfile = util.atomictempfile | ||||
|
r14007 | |||
|
r36299 | if pycompat.ispy3: | ||
xrange = range | ||||
|
r18666 | class testatomictempfile(unittest.TestCase): | ||
|
r29391 | def setUp(self): | ||
|
r36634 | self._testdir = tempfile.mkdtemp(b'atomictempfiletest') | ||
self._filename = os.path.join(self._testdir, b'testfilename') | ||||
|
r29391 | |||
def tearDown(self): | ||||
shutil.rmtree(self._testdir, True) | ||||
|
r29392 | def testsimple(self): | ||
|
r29391 | file = atomictempfile(self._filename) | ||
self.assertFalse(os.path.isfile(self._filename)) | ||||
tempfilename = file._tempname | ||||
self.assertTrue(tempfilename in glob.glob( | ||||
|
r36634 | os.path.join(self._testdir, b'.testfilename-*'))) | ||
|
r14007 | |||
|
r29188 | file.write(b'argh\n') | ||
|
r18666 | file.close() | ||
|
r14007 | |||
|
r29391 | self.assertTrue(os.path.isfile(self._filename)) | ||
self.assertTrue(tempfilename not in glob.glob( | ||||
|
r36634 | os.path.join(self._testdir, b'.testfilename-*'))) | ||
|
r14007 | |||
|
r18666 | # discard() removes the temp file without making the write permanent | ||
|
r29392 | def testdiscard(self): | ||
|
r29391 | file = atomictempfile(self._filename) | ||
|
r18666 | (dir, basename) = os.path.split(file._tempname) | ||
|
r14007 | |||
|
r29188 | file.write(b'yo\n') | ||
|
r18666 | file.discard() | ||
|
r14007 | |||
|
r29391 | self.assertFalse(os.path.isfile(self._filename)) | ||
|
r36634 | self.assertTrue(basename not in os.listdir(b'.')) | ||
|
r18666 | |||
# if a programmer screws up and passes bad args to atomictempfile, they | ||||
# get a plain ordinary TypeError, not infinite recursion | ||||
|
r29392 | def testoops(self): | ||
|
r32279 | with self.assertRaises(TypeError): | ||
atomictempfile() | ||||
|
r14007 | |||
|
r29201 | # checkambig=True avoids ambiguity of timestamp | ||
|
r29392 | def testcheckambig(self): | ||
|
r29201 | def atomicwrite(checkambig): | ||
|
r29391 | f = atomictempfile(self._filename, checkambig=checkambig) | ||
|
r36634 | f.write(b'FOO') | ||
|
r29201 | f.close() | ||
# try some times, because reproduction of ambiguity depends on | ||||
# "filesystem time" | ||||
for i in xrange(5): | ||||
atomicwrite(False) | ||||
|
r29391 | oldstat = os.stat(self._filename) | ||
|
r36799 | if oldstat[stat.ST_CTIME] != oldstat[stat.ST_MTIME]: | ||
|
r29201 | # subsequent changing never causes ambiguity | ||
continue | ||||
repetition = 3 | ||||
# repeat atomic write with checkambig=True, to examine | ||||
|
r30332 | # whether st_mtime is advanced multiple times as expected | ||
|
r29201 | for j in xrange(repetition): | ||
atomicwrite(True) | ||||
|
r29391 | newstat = os.stat(self._filename) | ||
|
r36799 | if oldstat[stat.ST_CTIME] != newstat[stat.ST_CTIME]: | ||
|
r29201 | # timestamp ambiguity was naturally avoided while repetition | ||
continue | ||||
# st_mtime should be advanced "repetition" times, because | ||||
|
r30332 | # all atomicwrite() occurred at same time (in sec) | ||
|
r36799 | oldtime = (oldstat[stat.ST_MTIME] + repetition) & 0x7fffffff | ||
self.assertTrue(newstat[stat.ST_MTIME] == oldtime) | ||||
|
r29201 | # no more examination is needed, if assumption above is true | ||
break | ||||
else: | ||||
# This platform seems too slow to examine anti-ambiguity | ||||
# of file timestamp (or test happened to be executed at | ||||
# bad timing). Exit silently in this case, because running | ||||
# on other faster platforms can detect problems | ||||
pass | ||||
|
r29393 | def testread(self): | ||
with open(self._filename, 'wb') as f: | ||||
f.write(b'foobar\n') | ||||
|
r36634 | file = atomictempfile(self._filename, mode=b'rb') | ||
|
r29393 | self.assertTrue(file.read(), b'foobar\n') | ||
file.discard() | ||||
|
r29394 | def testcontextmanagersuccess(self): | ||
"""When the context closes, the file is closed""" | ||||
|
r36634 | with atomictempfile(b'foo') as f: | ||
self.assertFalse(os.path.isfile(b'foo')) | ||||
|
r29394 | f.write(b'argh\n') | ||
|
r36634 | self.assertTrue(os.path.isfile(b'foo')) | ||
|
r29394 | |||
def testcontextmanagerfailure(self): | ||||
"""On exception, the file is discarded""" | ||||
try: | ||||
|
r36634 | with atomictempfile(b'foo') as f: | ||
self.assertFalse(os.path.isfile(b'foo')) | ||||
|
r29394 | f.write(b'argh\n') | ||
raise ValueError | ||||
except ValueError: | ||||
pass | ||||
|
r36634 | self.assertFalse(os.path.isfile(b'foo')) | ||
|
r29394 | |||
|
r14007 | if __name__ == '__main__': | ||
|
r29194 | import silenttestrunner | ||
|
r18666 | silenttestrunner.main(__name__) | ||