diff options
Diffstat (limited to 'git/test/lib')
| -rw-r--r-- | git/test/lib/__init__.py | 5 | ||||
| -rw-r--r-- | git/test/lib/base.py | 200 | ||||
| -rw-r--r-- | git/test/lib/helper.py | 75 |
3 files changed, 239 insertions, 41 deletions
diff --git a/git/test/lib/__init__.py b/git/test/lib/__init__.py index 77512794..a0656438 100644 --- a/git/test/lib/__init__.py +++ b/git/test/lib/__init__.py @@ -5,9 +5,14 @@ # the BSD License: http://www.opensource.org/licenses/bsd-license.php import inspect +# TODO: Separate names - they do repeat unfortunately. Also deduplicate it, +# redesign decorators to support multiple database types in succession. +from base import * + from mock import * from asserts import * from helper import * + __all__ = [ name for name, obj in locals().items() if not (name.startswith('_') or inspect.ismodule(obj)) ] diff --git a/git/test/lib/base.py b/git/test/lib/base.py new file mode 100644 index 00000000..bc160783 --- /dev/null +++ b/git/test/lib/base.py @@ -0,0 +1,200 @@ +# Copyright (C) 2010, 2011 Sebastian Thiel (byronimo@gmail.com) and contributors +# +# This module is part of PureCompatibilityGitDB and is released under +# 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.util import ( + zlib, + dirname + ) + +import sys +import random +from array import array +from cStringIO import StringIO + +import glob +import unittest +import tempfile +import shutil +import os +import gc + + +#{ Decorators + +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) + keep = False + try: + try: + return func(self, path) + except Exception: + print >> sys.stderr, "Test %s.%s failed, output is at %r" % (type(self).__name__, func.__name__, path) + 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. + if not keep: + gc.collect() + shutil.rmtree(path) + # END handle exception + # END wrapper + + wrapper.__name__ = func.__name__ + return wrapper + + +def with_rw_repo(func): + """Create a copy of our repository and put it into a writable location. It will + be removed if the test doesn't result in an error. + As we can currently only copy the fully working tree, tests must not rely on + 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)) + os.rmdir(path) # created by wrapper, but must not exist for copy operation + shutil.copytree(src_dir, path) + target_gitdir = os.path.join(path, '.git') + assert os.path.isdir(target_gitdir) + return func(self, self.RepoCls(target_gitdir)) + #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) + +#} END decorators + +#{ Routines + +def rorepo_dir(): + """:return: path to our own repository, being our own .git directory. + :note: doesn't work in bare repositories""" + base = os.path.join(dirname(dirname(dirname(dirname(__file__)))), '.git') + 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) + if sys.platform == 'darwin': + 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 + the files will be copied""" + for src_file in glob.glob(source_glob): + if hard_link_ok and hasattr(os, 'link'): + target = os.path.join(target_dir, os.path.basename(src_file)) + try: + os.link(src_file, target) + except OSError: + 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 + + +def make_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 = xrange(actual_size) + if randomize: + producer = list(producer) + random.shuffle(producer) + # END randomize + 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""" + d = make_bytes(size_in_bytes, randomize) + return len(d), StringIO(d) + +#} END routines + +#{ 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 + + +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 + diff --git a/git/test/lib/helper.py b/git/test/lib/helper.py index 76aaaa38..5776f526 100644 --- a/git/test/lib/helper.py +++ b/git/test/lib/helper.py @@ -12,26 +12,18 @@ import tempfile import shutil import cStringIO -GIT_REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) +from base import ( + maketemp, + rorepo_dir + ) -__all__ = ( - 'fixture_path', 'fixture', 'absolute_project_path', 'StringProcessAdapter', - 'with_rw_repo', 'with_rw_and_rw_remote_repo', 'TestBase', 'TestCase', 'GIT_REPO' - ) - -#{ Routines - -def fixture_path(name): - test_dir = os.path.dirname(os.path.dirname(__file__)) - return os.path.join(test_dir, "fixtures", name) -def fixture(name): - return open(fixture_path(name), 'rb').read() +__all__ = ( + 'StringProcessAdapter', 'GlobalsItemDeletorMetaCls', + 'with_rw_repo', 'with_rw_and_rw_remote_repo', 'TestBase', 'TestCase', + ) -def absolute_project_path(): - return os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) -#} END routines #{ Adapters @@ -52,13 +44,6 @@ class StringProcessAdapter(object): #{ Decorators -def _mktemp(*args): - """Wrapper around default tempfile.mktemp to fix an osx issue""" - tdir = tempfile.mktemp(*args) - if sys.platform == 'darwin': - tdir = '/private' + tdir - return tdir - def _rmtree_onerror(osremove, fullpath, exec_info): """ Handle the case on windows that read-only files cannot be deleted by @@ -87,7 +72,7 @@ def with_rw_repo(working_tree_ref, bare=False): if bare: prefix = '' #END handle prefix - repo_dir = _mktemp("%sbare_%s" % (prefix, func.__name__)) + repo_dir = maketemp("%sbare_%s" % (prefix, func.__name__)) rw_repo = self.rorepo.clone(repo_dir, shared=True, bare=bare, n=True) rw_repo.head.commit = rw_repo.commit(working_tree_ref) @@ -143,8 +128,8 @@ def with_rw_and_rw_remote_repo(working_tree_ref): assert isinstance(working_tree_ref, basestring), "Decorator requires ref name for working tree checkout" def argument_passer(func): def remote_repo_creator(self): - remote_repo_dir = _mktemp("remote_repo_%s" % func.__name__) - repo_dir = _mktemp("remote_clone_non_bare_repo") + remote_repo_dir = maketemp("remote_repo_%s" % func.__name__) + repo_dir = maketemp("remote_clone_non_bare_repo") rw_remote_repo = self.rorepo.clone(remote_repo_dir, shared=True, bare=True) rw_repo = rw_remote_repo.clone(repo_dir, shared=True, bare=False, n=True) # recursive alternates info ? @@ -205,32 +190,40 @@ def with_rw_and_rw_remote_repo(working_tree_ref): return argument_passer #} END decorators + +#{ Meta Classes +class GlobalsItemDeletorMetaCls(type): + """Utiltiy to prevent the RepoBase to be picked up by nose as the metacls + will delete the instance from the globals""" + #{ Configuration + # Set this to a string name of the module to delete + ModuleToDelete = None + #} END configuration + + def __new__(metacls, name, bases, clsdict): + assert metacls.ModuleToDelete is not None, "Invalid metaclass configuration" + new_type = super(GlobalsItemDeletorMetaCls, metacls).__new__(metacls, name, bases, clsdict) + if name != metacls.ModuleToDelete: + mod = __import__(new_type.__module__, globals(), locals(), new_type.__module__) + delattr(mod, metacls.ModuleToDelete) + #END handle deletion + return new_type + +#} END meta classes class TestBase(TestCase): """ Base Class providing default functionality to all tests such as: - - Utility functions provided by the TestCase base of the unittest method such as:: self.fail("todo") self.failUnlessRaises(...) - - - Class level repository which is considered read-only as it is shared among - all test cases in your type. - Access it using:: - self.rorepo # 'ro' stands for read-only - - The rorepo is in fact your current project's git repo. If you refer to specific - shas for your objects, be sure you choose some that are part of the immutable portion - of the project history ( to assure tests don't fail for others ). """ @classmethod def setUpAll(cls): - """ - Dynamically add a read-only repository to our actual type. This way - each test type has its own repository - """ - cls.rorepo = Repo(GIT_REPO) + """This method is only called to provide the most basic functionality + Subclasses may just override it or implement it differently""" + cls.rorepo = Repo(rorepo_dir()) def _make_file(self, rela_path, data, repo=None): """ |