diff options
author | noptuno <repollo.marrero@gmail.com> | 2023-04-28 02:29:30 +0200 |
---|---|---|
committer | noptuno <repollo.marrero@gmail.com> | 2023-04-28 02:29:30 +0200 |
commit | 355dee533bb34a571b9367820a63cccb668cf866 (patch) | |
tree | 838af886b4fec07320aeb10f0d1e74ba79e79b5c /venv/lib/python3.9/site-packages/gitdb/test/lib.py | |
parent | added pyproject.toml file (diff) | |
download | gpt4free-355dee533bb34a571b9367820a63cccb668cf866.tar gpt4free-355dee533bb34a571b9367820a63cccb668cf866.tar.gz gpt4free-355dee533bb34a571b9367820a63cccb668cf866.tar.bz2 gpt4free-355dee533bb34a571b9367820a63cccb668cf866.tar.lz gpt4free-355dee533bb34a571b9367820a63cccb668cf866.tar.xz gpt4free-355dee533bb34a571b9367820a63cccb668cf866.tar.zst gpt4free-355dee533bb34a571b9367820a63cccb668cf866.zip |
Diffstat (limited to 'venv/lib/python3.9/site-packages/gitdb/test/lib.py')
-rw-r--r-- | venv/lib/python3.9/site-packages/gitdb/test/lib.py | 192 |
1 files changed, 192 insertions, 0 deletions
diff --git a/venv/lib/python3.9/site-packages/gitdb/test/lib.py b/venv/lib/python3.9/site-packages/gitdb/test/lib.py new file mode 100644 index 00000000..da59d3b3 --- /dev/null +++ b/venv/lib/python3.9/site-packages/gitdb/test/lib.py @@ -0,0 +1,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: http://www.opensource.org/licenses/bsd-license.php +"""Utilities used in ODB testing""" +from gitdb import OStream + +import sys +import random +from array import array + +from io import BytesIO + +import glob +import unittest +import tempfile +import shutil +import os +import gc +import logging +from functools import wraps + + +#{ Bases + +class TestBase(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 + def setUpClass(cls): + try: + super().setUpClass() + except AttributeError: + pass + + cls.gitrepopath = os.environ.get(cls.k_env_git_repo) + if not cls.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 + assert cls.gitrepopath.endswith('.git') + + +#} END bases + +#{ Decorators + +def with_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""" + + def wrapper(self): + path = tempfile.mktemp(prefix=func.__name__) + os.mkdir(path) + keep = False + try: + try: + return func(self, path) + except Exception: + 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. + if not keep: + gc.collect() + shutil.rmtree(path) + # END handle exception + # END wrapper + + wrapper.__name__ = func.__name__ + return 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""" + + def wrapper(self, path): + src_pack_glob = fixture_path('packs/*') + copy_files_globbed(src_pack_glob, path, hard_link_ok=True) + return func(self, path) + # END wrapper + + wrapper.__name__ = func.__name__ + return wrapper + +#} END decorators + +#{ Routines + + +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""" + return os.path.join(os.path.dirname(__file__), 'fixtures', relapath) + + +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 = range(actual_size) + if randomize: + producer = list(producer) + random.shuffle(producer) + # END randomize + a = array('i', producer) + return a.tobytes() + + +def make_object(type, data): + """:return: bytes resembling an uncompressed object""" + odata = "blob %i\0" % len(data) + return odata.encode("ascii") + 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), BytesIO(d) + +#} END routines + +#{ Stream Utilities + + +class DummyStream: + + 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 |