- Notifications
You must be signed in to change notification settings - Fork68
Expand file tree
/
Copy pathtest_util.py
More file actions
100 lines (83 loc) · 3.17 KB
/
test_util.py
File metadata and controls
100 lines (83 loc) · 3.17 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
# Copyright (C) 2010, 2011 Sebastian Thiel (byronimo@gmail.com) and contributors
#
# This module is part of GitDB and is released under
# the New BSD License: https://opensource.org/license/bsd-3-clause/
"""Test for object db"""
importtempfile
importos
fromgitdb.test.libimportTestBase
fromgitdb.utilimport (
to_hex_sha,
to_bin_sha,
NULL_HEX_SHA,
LockedFD
)
classTestUtils(TestBase):
deftest_basics(self):
assertto_hex_sha(NULL_HEX_SHA)==NULL_HEX_SHA
assertlen(to_bin_sha(NULL_HEX_SHA))==20
assertto_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
# does not match data string
withopen(file_path,"rb")asfp:
assertfp.read()==data.encode("ascii")
deftest_lockedfd(self):
my_file=tempfile.mktemp()
orig_data="hello"
new_data="world"
withopen(my_file,"wb")asmy_file_fp:
my_file_fp.write(orig_data.encode("ascii"))
try:
lfd=LockedFD(my_file)
lockfilepath=lfd._lockfilepath()
# cannot end before it was started
self.assertRaises(AssertionError,lfd.rollback)
self.assertRaises(AssertionError,lfd.commit)
# open for writing
assertnotos.path.isfile(lockfilepath)
wfd=lfd.open(write=True)
assertlfd._fdiswfd
assertos.path.isfile(lockfilepath)
# write data and fail
os.write(wfd,new_data.encode("ascii"))
lfd.rollback()
assertlfd._fdisNone
self._cmp_contents(my_file,orig_data)
assertnotos.path.isfile(lockfilepath)
# additional call doesn't fail
lfd.commit()
lfd.rollback()
# test reading
lfd=LockedFD(my_file)
rfd=lfd.open(write=False)
assertos.read(rfd,len(orig_data))==orig_data.encode("ascii")
assertos.path.isfile(lockfilepath)
# deletion rolls back
del(lfd)
assertnotos.path.isfile(lockfilepath)
# write data - concurrently
lfd=LockedFD(my_file)
olfd=LockedFD(my_file)
assertnotos.path.isfile(lockfilepath)
wfdstream=lfd.open(write=True,stream=True)# this time as stream
assertos.path.isfile(lockfilepath)
# another one fails
self.assertRaises(IOError,olfd.open)
wfdstream.write(new_data.encode("ascii"))
lfd.commit()
assertnotos.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:
lfd.open(write=False)
exceptOSError:
assertnotos.path.exists(lfd._lockfilepath())
else:
self.fail("expected OSError")
# END handle exceptions