- Notifications
You must be signed in to change notification settings - Fork68
Expand file tree
/
Copy pathlib.py
More file actions
192 lines (145 loc) · 5.36 KB
/
lib.py
File metadata and controls
192 lines (145 loc) · 5.36 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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
# 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/
"""Utilities used in ODB testing"""
fromgitdbimportOStream
importsys
importrandom
fromarrayimportarray
fromioimportBytesIO
importglob
importunittest
importtempfile
importshutil
importos
importgc
importlogging
fromfunctoolsimportwraps
#{ Bases
classTestBase(unittest.TestCase):
"""Base class for all tests
TestCase providing access to readonly repositories using the following member variables.
* gitrepopath
* read-only base path of the git source repository, i.e. .../git/.git
"""
#{ Invvariants
k_env_git_repo="GITDB_TEST_GIT_REPO_BASE"
#} END invariants
@classmethod
defsetUpClass(cls):
try:
super().setUpClass()
exceptAttributeError:
pass
cls.gitrepopath=os.environ.get(cls.k_env_git_repo)
ifnotcls.gitrepopath:
logging.info(
"You can set the %s environment variable to a .git repository of your choice - defaulting to the gitdb repository",cls.k_env_git_repo)
ospd=os.path.dirname
cls.gitrepopath=os.path.join(ospd(ospd(ospd(__file__))),'.git')
# end assure gitrepo is set
assertcls.gitrepopath.endswith('.git')
#} END bases
#{ Decorators
defwith_rw_directory(func):
"""Create a temporary directory which can be written to, remove it if the
test succeeds, but leave it otherwise to aid additional debugging"""
defwrapper(self):
path=tempfile.mktemp(prefix=func.__name__)
os.mkdir(path)
keep=False
try:
try:
returnfunc(self,path)
exceptException:
sys.stderr.write(f"Test{type(self).__name__}.{func.__name__} failed, output is at{path!r}\n")
keep=True
raise
finally:
# Need to collect here to be sure all handles have been closed. It appears
# a windows-only issue. In fact things should be deleted, as well as
# memory maps closed, once objects go out of scope. For some reason
# though this is not the case here unless we collect explicitly.
ifnotkeep:
gc.collect()
shutil.rmtree(path)
# END handle exception
# END wrapper
wrapper.__name__=func.__name__
returnwrapper
defwith_packs_rw(func):
"""Function that provides a path into which the packs for testing should be
copied. Will pass on the path to the actual function afterwards"""
defwrapper(self,path):
src_pack_glob=fixture_path('packs/*')
copy_files_globbed(src_pack_glob,path,hard_link_ok=True)
returnfunc(self,path)
# END wrapper
wrapper.__name__=func.__name__
returnwrapper
#} END decorators
#{ Routines
deffixture_path(relapath=''):
""":return: absolute path into the fixture directory
:param relapath: relative path into the fixtures directory, or ''
to obtain the fixture directory itself"""
returnos.path.join(os.path.dirname(__file__),'fixtures',relapath)
defcopy_files_globbed(source_glob,target_dir,hard_link_ok=False):
"""Copy all files found according to the given source glob into the target directory
:param hard_link_ok: if True, hard links will be created if possible. Otherwise
the files will be copied"""
forsrc_fileinglob.glob(source_glob):
ifhard_link_okandhasattr(os,'link'):
target=os.path.join(target_dir,os.path.basename(src_file))
try:
os.link(src_file,target)
exceptOSError:
shutil.copy(src_file,target_dir)
# END handle cross device links ( and resulting failure )
else:
shutil.copy(src_file,target_dir)
# END try hard link
# END for each file to copy
defmake_bytes(size_in_bytes,randomize=False):
""":return: string with given size in bytes
:param randomize: try to produce a very random stream"""
actual_size=size_in_bytes//4
producer=range(actual_size)
ifrandomize:
producer=list(producer)
random.shuffle(producer)
# END randomize
a=array('i',producer)
returna.tobytes()
defmake_object(type,data):
""":return: bytes resembling an uncompressed object"""
odata="blob %i\0"%len(data)
returnodata.encode("ascii")+data
defmake_memory_file(size_in_bytes,randomize=False):
""":return: tuple(size_of_stream, stream)
:param randomize: try to produce a very random stream"""
d=make_bytes(size_in_bytes,randomize)
returnlen(d),BytesIO(d)
#} END routines
#{ Stream Utilities
classDummyStream:
def__init__(self):
self.was_read=False
self.bytes=0
self.closed=False
defread(self,size):
self.was_read=True
self.bytes=size
defclose(self):
self.closed=True
def_assert(self):
assertself.was_read
classDeriveTest(OStream):
def__init__(self,sha,type,size,stream,*args,**kwargs):
self.myarg=kwargs.pop('myarg')
self.args=args
def_assert(self):
assertself.args
assertself.myarg
#} END stream utilitiess