diff options
Diffstat (limited to 'gitdb/test/test_util.py')
| -rw-r--r-- | gitdb/test/test_util.py | 57 |
1 files changed, 28 insertions, 29 deletions
diff --git a/gitdb/test/test_util.py b/gitdb/test/test_util.py index 35f9f44..e79355a 100644 --- a/gitdb/test/test_util.py +++ b/gitdb/test/test_util.py @@ -6,74 +6,74 @@ import tempfile import os -from lib import TestBase +from gitdb.test.lib import TestBase from gitdb.util import ( - to_hex_sha, - to_bin_sha, - NULL_HEX_SHA, + to_hex_sha, + to_bin_sha, + NULL_HEX_SHA, LockedFD - ) +) + - class TestUtils(TestBase): def test_basics(self): assert to_hex_sha(NULL_HEX_SHA) == NULL_HEX_SHA assert len(to_bin_sha(NULL_HEX_SHA)) == 20 - assert to_hex_sha(to_bin_sha(NULL_HEX_SHA)) == NULL_HEX_SHA - + assert to_hex_sha(to_bin_sha(NULL_HEX_SHA)) == NULL_HEX_SHA.encode("ascii") + def _cmp_contents(self, file_path, data): - # raise if data from file at file_path + # raise if data from file at file_path # does not match data string fp = open(file_path, "rb") try: - assert fp.read() == data + assert fp.read() == data.encode("ascii") finally: fp.close() - + def test_lockedfd(self): my_file = tempfile.mktemp() orig_data = "hello" new_data = "world" my_file_fp = open(my_file, "wb") - my_file_fp.write(orig_data) + my_file_fp.write(orig_data.encode("ascii")) my_file_fp.close() - + try: lfd = LockedFD(my_file) - lockfilepath = lfd._lockfilepath() - + lockfilepath = lfd._lockfilepath() + # cannot end before it was started self.failUnlessRaises(AssertionError, lfd.rollback) self.failUnlessRaises(AssertionError, lfd.commit) - + # open for writing assert not os.path.isfile(lockfilepath) wfd = lfd.open(write=True) assert lfd._fd is wfd assert os.path.isfile(lockfilepath) - + # write data and fail - os.write(wfd, new_data) + os.write(wfd, new_data.encode("ascii")) lfd.rollback() assert lfd._fd is None self._cmp_contents(my_file, orig_data) assert not os.path.isfile(lockfilepath) - + # additional call doesnt fail lfd.commit() lfd.rollback() - + # test reading lfd = LockedFD(my_file) rfd = lfd.open(write=False) - assert os.read(rfd, len(orig_data)) == orig_data - + assert os.read(rfd, len(orig_data)) == orig_data.encode("ascii") + assert os.path.isfile(lockfilepath) # deletion rolls back del(lfd) assert not os.path.isfile(lockfilepath) - - + + # write data - concurrently lfd = LockedFD(my_file) olfd = LockedFD(my_file) @@ -82,17 +82,17 @@ class TestUtils(TestBase): assert os.path.isfile(lockfilepath) # another one fails self.failUnlessRaises(IOError, olfd.open) - - wfdstream.write(new_data) + + wfdstream.write(new_data.encode("ascii")) lfd.commit() assert not os.path.isfile(lockfilepath) self._cmp_contents(my_file, new_data) - + # could test automatic _end_writing on destruction finally: os.remove(my_file) # END final cleanup - + # try non-existing file for reading lfd = LockedFD(tempfile.mktemp()) try: @@ -102,4 +102,3 @@ class TestUtils(TestBase): else: self.fail("expected OSError") # END handle exceptions - |
