diff options
Diffstat (limited to 'git/test/lib/base.py')
| -rw-r--r-- | git/test/lib/base.py | 76 |
1 files changed, 44 insertions, 32 deletions
diff --git a/git/test/lib/base.py b/git/test/lib/base.py index 298e8e05..39bc9b73 100644 --- a/git/test/lib/base.py +++ b/git/test/lib/base.py @@ -4,15 +4,15 @@ # the New BSD License: http://www.opensource.org/licenses/bsd-license.php """Utilities used in ODB testing""" from git.base import OStream -from git.stream import ( - Sha1Writer, - ZippedStoreShaWriter - ) +from git.stream import ( + Sha1Writer, + ZippedStoreShaWriter +) from git.util import ( - zlib, - dirname - ) + zlib, + dirname +) import sys import random @@ -32,6 +32,7 @@ import gc def with_rw_directory(func): """Create a temporary directory which can be written to, remove it if the test suceeds, but leave it otherwise to aid additional debugging""" + def wrapper(self): path = maketemp(prefix=func.__name__) os.mkdir(path) @@ -45,7 +46,7 @@ def with_rw_directory(func): 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 + # 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. if not keep: @@ -53,7 +54,7 @@ def with_rw_directory(func): shutil.rmtree(path) # END handle exception # END wrapper - + wrapper.__name__ = func.__name__ return wrapper @@ -65,6 +66,7 @@ def with_rw_repo(func): being on a certain branch or on anything really except for the default tags that should exist Wrapped function obtains a git repository """ + def wrapper(self, path): src_dir = dirname(dirname(dirname(__file__))) assert(os.path.isdir(path)) @@ -73,24 +75,24 @@ def with_rw_repo(func): target_gitdir = os.path.join(path, '.git') assert os.path.isdir(target_gitdir) return func(self, self.RepoCls(target_gitdir)) - #END wrapper + # END wrapper wrapper.__name__ = func.__name__ return with_rw_directory(wrapper) - def with_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 - + :note: needs with_rw_directory wrapped around it""" + def wrapper(self, path): src_pack_glob = fixture_path('packs/*') print src_pack_glob copy_files_globbed(src_pack_glob, path, hard_link_ok=True) return func(self, path) # END wrapper - + wrapper.__name__ = func.__name__ return with_rw_directory(wrapper) @@ -98,6 +100,7 @@ def with_packs_rw(func): #{ Routines + def rorepo_dir(): """:return: path to our own repository, being our own .git directory. :note: doesn't work in bare repositories""" @@ -105,6 +108,7 @@ def rorepo_dir(): assert os.path.isdir(base) return base + def maketemp(*args, **kwargs): """Wrapper around default tempfile.mktemp to fix an osx issue""" tdir = tempfile.mktemp(*args, **kwargs) @@ -112,19 +116,23 @@ def maketemp(*args, **kwargs): tdir = '/private' + tdir return tdir + def fixture_path(relapath=''): """:return: absolute path into the fixture directory :param relapath: relative path into the fixtures directory, or '' to obtain the fixture directory itself""" test_dir = os.path.dirname(os.path.dirname(__file__)) return os.path.join(test_dir, "fixtures", relapath) - + + def fixture(name): return open(fixture_path(name), 'rb').read() + def absolute_project_path(): return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) + def copy_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 @@ -141,7 +149,7 @@ def copy_files_globbed(source_glob, target_dir, hard_link_ok=False): shutil.copy(src_file, target_dir) # END try hard link # END for each file to copy - + def make_bytes(size_in_bytes, randomize=False): """:return: string with given size in bytes @@ -155,11 +163,13 @@ def make_bytes(size_in_bytes, randomize=False): a = array('i', producer) return a.tostring() + def make_object(type, data): """:return: bytes resembling an uncompressed object""" odata = "blob %i\0" % len(data) return odata + data - + + def make_memory_file(size_in_bytes, randomize=False): """:return: tuple(size_of_stream, stream) :param randomize: try to produce a very random stream""" @@ -170,31 +180,33 @@ def make_memory_file(size_in_bytes, randomize=False): #{ Stream Utilities + class DummyStream(object): - def __init__(self): - self.was_read = False - self.bytes = 0 - self.closed = False - - def read(self, size): - self.was_read = True - self.bytes = size - - def close(self): - self.closed = True - - def _assert(self): - assert self.was_read + + def __init__(self): + self.was_read = False + self.bytes = 0 + self.closed = False + + def read(self, size): + self.was_read = True + self.bytes = size + + def close(self): + self.closed = True + + def _assert(self): + assert self.was_read class DeriveTest(OStream): + def __init__(self, sha, type, size, stream, *args, **kwargs): self.myarg = kwargs.pop('myarg') self.args = args - + def _assert(self): assert self.args assert self.myarg #} END stream utilitiess - |