summaryrefslogtreecommitdiff
path: root/gitdb/test/test_util.py
diff options
context:
space:
mode:
Diffstat (limited to 'gitdb/test/test_util.py')
-rw-r--r--gitdb/test/test_util.py57
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
-