diff --git a/dulwich/tests/__init__.py b/dulwich/tests/__init__.py index 03712374..8e33051f 100644 --- a/dulwich/tests/__init__.py +++ b/dulwich/tests/__init__.py @@ -1,145 +1,158 @@ # __init__.py -- The tests for dulwich # Copyright (C) 2007 James Westby # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; version 2 # of the License or (at your option) any later version of # the License. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, # MA 02110-1301, USA. """Tests for Dulwich.""" import doctest import os import shutil import subprocess import sys import tempfile if sys.version_info >= (2, 7): # If Python itself provides an exception, use that import unittest from unittest import SkipTest, TestCase as _TestCase else: import unittest2 as unittest from unittest2 import SkipTest, TestCase as _TestCase +def get_safe_env(env=None): + """Returns the environment "env" (or a copy of "os.environ" by default) + modified to avoid side-effects caused by user's ~/.gitconfig""" + + if env is None: + env = os.environ.copy() + # On Windows it's not enough to set "HOME" to a non-existing + # directory. Git.cmd takes the first existing directory out of + # "%HOME%", "%HOMEDRIVE%%HOMEPATH%" and "%USERPROFILE%". + for e in 'HOME', 'HOMEPATH', 'USERPROFILE': + env[e] = '/nosuchdir' + return env class TestCase(_TestCase): def makeSafeEnv(self): - """Modifies HOME to point to a non-existing directory to avoid + """Modifies "os.environ" for the duration of a test case to avoid side-effects caused by user's ~/.gitconfig""" - if "HOME" in os.environ: - self.addCleanup(os.environ.__setitem__, "HOME", os.environ["HOME"]) - else: - self.addCleanup(os.environ.__delitem__, "HOME") - os.environ["HOME"] = "/nonexistant" + for e in 'HOME', 'HOMEPATH', 'USERPROFILE': + if e in os.environ: + self.addCleanup(os.environ.__setitem__, e, os.environ[e]) + else: + self.addCleanup(os.environ.__delitem__, e) + os.environ[e] = '/nosuchdir' class BlackboxTestCase(TestCase): """Blackbox testing.""" bin_directory = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "bin")) def bin_path(self, name): """Determine the full path of a binary. :param name: Name of the script :return: Full path """ return os.path.join(self.bin_directory, name) def run_command(self, name, args): """Run a Dulwich command. :param name: Name of the command, as it exists in bin/ :param args: Arguments to the command """ env = dict(os.environ) env["PYTHONPATH"] = os.pathsep.join(sys.path) # Since they don't have any extensions, Windows can't recognize # executablility of the Python files in /bin. Even then, we'd have to # expect the user to set up file associations for .py files. # # Save us from all that headache and call python with the bin script. argv = [sys.executable, self.bin_path(name)] + args return subprocess.Popen(argv, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, env=env) def self_test_suite(): names = [ 'blackbox', 'client', 'config', 'diff_tree', 'fastexport', 'file', 'index', 'lru_cache', 'objects', 'object_store', 'pack', 'patch', 'protocol', 'repository', 'server', 'walk', 'web', ] module_names = ['dulwich.tests.test_' + name for name in names] loader = unittest.TestLoader() return loader.loadTestsFromNames(module_names) def tutorial_test_suite(): tutorial = [ 'introduction', 'repo', 'object-store', 'remote', 'conclusion', ] tutorial_files = ["../../docs/tutorial/%s.txt" % name for name in tutorial] def setup(test): test.__old_cwd = os.getcwd() test.__dulwich_tempdir = tempfile.mkdtemp() os.chdir(test.__dulwich_tempdir) def teardown(test): os.chdir(test.__old_cwd) shutil.rmtree(test.__dulwich_tempdir) return doctest.DocFileSuite(setUp=setup, tearDown=teardown, *tutorial_files) def nocompat_test_suite(): result = unittest.TestSuite() result.addTests(self_test_suite()) result.addTests(tutorial_test_suite()) return result def test_suite(): result = unittest.TestSuite() result.addTests(self_test_suite()) result.addTests(tutorial_test_suite()) from dulwich.tests.compat import test_suite as compat_test_suite result.addTests(compat_test_suite()) return result diff --git a/dulwich/tests/compat/test_client.py b/dulwich/tests/compat/test_client.py index 1ab711e2..aace589c 100644 --- a/dulwich/tests/compat/test_client.py +++ b/dulwich/tests/compat/test_client.py @@ -1,470 +1,472 @@ # test_client.py -- Compatibilty tests for git client. # Copyright (C) 2010 Google, Inc. # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; version 2 # of the License or (at your option) any later version of # the License. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, # MA 02110-1301, USA. """Compatibilty tests between the Dulwich client and the cgit server.""" from cStringIO import StringIO import BaseHTTPServer import SimpleHTTPServer import copy import os import select import shutil import signal import subprocess import tarfile import tempfile import threading import urllib from dulwich import ( client, errors, file, index, protocol, objects, repo, ) from dulwich.tests import ( + get_safe_env, SkipTest, ) from dulwich.tests.compat.utils import ( CompatTestCase, check_for_daemon, import_repo_to_dir, run_git_or_fail, ) from dulwich.tests.compat.server_utils import ( ShutdownServerMixIn, ) class DulwichClientTestBase(object): """Tests for client/server compatibility.""" def setUp(self): self.gitroot = os.path.dirname(import_repo_to_dir('server_new.export')) self.dest = os.path.join(self.gitroot, 'dest') file.ensure_dir_exists(self.dest) run_git_or_fail(['init', '--quiet', '--bare'], cwd=self.dest) def tearDown(self): shutil.rmtree(self.gitroot) def assertDestEqualsSrc(self): src = repo.Repo(os.path.join(self.gitroot, 'server_new.export')) dest = repo.Repo(os.path.join(self.gitroot, 'dest')) self.assertReposEqual(src, dest) def _client(self): raise NotImplementedError() def _build_path(self): raise NotImplementedError() def _do_send_pack(self): c = self._client() srcpath = os.path.join(self.gitroot, 'server_new.export') src = repo.Repo(srcpath) sendrefs = dict(src.get_refs()) del sendrefs['HEAD'] c.send_pack(self._build_path('/dest'), lambda _: sendrefs, src.object_store.generate_pack_contents) def test_send_pack(self): self._do_send_pack() self.assertDestEqualsSrc() def test_send_pack_nothing_to_send(self): self._do_send_pack() self.assertDestEqualsSrc() # nothing to send, but shouldn't raise either. self._do_send_pack() def test_send_without_report_status(self): c = self._client() c._send_capabilities.remove('report-status') srcpath = os.path.join(self.gitroot, 'server_new.export') src = repo.Repo(srcpath) sendrefs = dict(src.get_refs()) del sendrefs['HEAD'] c.send_pack(self._build_path('/dest'), lambda _: sendrefs, src.object_store.generate_pack_contents) self.assertDestEqualsSrc() def make_dummy_commit(self, dest): b = objects.Blob.from_string('hi') dest.object_store.add_object(b) t = index.commit_tree(dest.object_store, [('hi', b.id, 0100644)]) c = objects.Commit() c.author = c.committer = 'Foo Bar ' c.author_time = c.commit_time = 0 c.author_timezone = c.commit_timezone = 0 c.message = 'hi' c.tree = t dest.object_store.add_object(c) return c.id def disable_ff_and_make_dummy_commit(self): # disable non-fast-forward pushes to the server dest = repo.Repo(os.path.join(self.gitroot, 'dest')) run_git_or_fail(['config', 'receive.denyNonFastForwards', 'true'], cwd=dest.path) commit_id = self.make_dummy_commit(dest) return dest, commit_id def compute_send(self): srcpath = os.path.join(self.gitroot, 'server_new.export') src = repo.Repo(srcpath) sendrefs = dict(src.get_refs()) del sendrefs['HEAD'] return sendrefs, src.object_store.generate_pack_contents def test_send_pack_one_error(self): dest, dummy_commit = self.disable_ff_and_make_dummy_commit() dest.refs['refs/heads/master'] = dummy_commit sendrefs, gen_pack = self.compute_send() c = self._client() try: c.send_pack(self._build_path('/dest'), lambda _: sendrefs, gen_pack) except errors.UpdateRefsError, e: self.assertEqual('refs/heads/master failed to update', str(e)) self.assertEqual({'refs/heads/branch': 'ok', 'refs/heads/master': 'non-fast-forward'}, e.ref_status) def test_send_pack_multiple_errors(self): dest, dummy = self.disable_ff_and_make_dummy_commit() # set up for two non-ff errors dest.refs['refs/heads/branch'] = dest.refs['refs/heads/master'] = dummy sendrefs, gen_pack = self.compute_send() c = self._client() try: c.send_pack(self._build_path('/dest'), lambda _: sendrefs, gen_pack) except errors.UpdateRefsError, e: self.assertEqual('refs/heads/branch, refs/heads/master failed to ' 'update', str(e)) self.assertEqual({'refs/heads/branch': 'non-fast-forward', 'refs/heads/master': 'non-fast-forward'}, e.ref_status) def test_archive(self): c = self._client() f = StringIO() c.archive(self._build_path('/server_new.export'), 'HEAD', f.write) f.seek(0) tf = tarfile.open(fileobj=f) self.assertEquals(['baz', 'foo'], tf.getnames()) def test_fetch_pack(self): c = self._client() dest = repo.Repo(os.path.join(self.gitroot, 'dest')) refs = c.fetch(self._build_path('/server_new.export'), dest) map(lambda r: dest.refs.set_if_equals(r[0], None, r[1]), refs.items()) self.assertDestEqualsSrc() def test_incremental_fetch_pack(self): self.test_fetch_pack() dest, dummy = self.disable_ff_and_make_dummy_commit() dest.refs['refs/heads/master'] = dummy c = self._client() dest = repo.Repo(os.path.join(self.gitroot, 'server_new.export')) refs = c.fetch(self._build_path('/dest'), dest) map(lambda r: dest.refs.set_if_equals(r[0], None, r[1]), refs.items()) self.assertDestEqualsSrc() def test_fetch_pack_zero_sha(self): # zero sha1s are already present on the client, and should # be ignored c = self._client() dest = repo.Repo(os.path.join(self.gitroot, 'dest')) refs = c.fetch(self._build_path('/server_new.export'), dest, lambda refs: [protocol.ZERO_SHA]) map(lambda r: dest.refs.set_if_equals(r[0], None, r[1]), refs.items()) def test_send_remove_branch(self): dest = repo.Repo(os.path.join(self.gitroot, 'dest')) dummy_commit = self.make_dummy_commit(dest) dest.refs['refs/heads/master'] = dummy_commit dest.refs['refs/heads/abranch'] = dummy_commit sendrefs = dict(dest.refs) sendrefs['refs/heads/abranch'] = "00" * 20 del sendrefs['HEAD'] gen_pack = lambda have, want: [] c = self._client() self.assertEquals(dest.refs["refs/heads/abranch"], dummy_commit) c.send_pack(self._build_path('/dest'), lambda _: sendrefs, gen_pack) self.assertFalse("refs/heads/abranch" in dest.refs) class DulwichTCPClientTest(CompatTestCase, DulwichClientTestBase): def setUp(self): CompatTestCase.setUp(self) DulwichClientTestBase.setUp(self) if check_for_daemon(limit=1): raise SkipTest('git-daemon was already running on port %s' % protocol.TCP_GIT_PORT) fd, self.pidfile = tempfile.mkstemp(prefix='dulwich-test-git-client', suffix=".pid") os.fdopen(fd).close() run_git_or_fail( ['daemon', '--verbose', '--export-all', '--pid-file=%s' % self.pidfile, '--base-path=%s' % self.gitroot, '--detach', '--reuseaddr', '--enable=receive-pack', '--enable=upload-archive', '--listen=localhost', self.gitroot], cwd=self.gitroot) if not check_for_daemon(): raise SkipTest('git-daemon failed to start') def tearDown(self): try: os.kill(int(open(self.pidfile).read().strip()), signal.SIGKILL) os.unlink(self.pidfile) except (OSError, IOError): pass DulwichClientTestBase.tearDown(self) CompatTestCase.tearDown(self) def _client(self): return client.TCPGitClient('localhost') def _build_path(self, path): return path class TestSSHVendor(object): @staticmethod def connect_ssh(host, command, username=None, port=None): cmd, path = command[0].replace("'", '').split(' ') cmd = cmd.split('-', 1) - p = subprocess.Popen(cmd + [path], stdin=subprocess.PIPE, + p = subprocess.Popen(cmd + [path], env=get_safe_env(), stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) return client.SubprocessWrapper(p) class DulwichMockSSHClientTest(CompatTestCase, DulwichClientTestBase): def setUp(self): CompatTestCase.setUp(self) DulwichClientTestBase.setUp(self) self.real_vendor = client.get_ssh_vendor client.get_ssh_vendor = TestSSHVendor def tearDown(self): DulwichClientTestBase.tearDown(self) CompatTestCase.tearDown(self) client.get_ssh_vendor = self.real_vendor def _client(self): return client.SSHGitClient('localhost') def _build_path(self, path): return self.gitroot + path class DulwichSubprocessClientTest(CompatTestCase, DulwichClientTestBase): def setUp(self): CompatTestCase.setUp(self) DulwichClientTestBase.setUp(self) + self.makeSafeEnv() def tearDown(self): DulwichClientTestBase.tearDown(self) CompatTestCase.tearDown(self) def _client(self): return client.SubprocessGitClient(stderr=subprocess.PIPE) def _build_path(self, path): return self.gitroot + path class GitHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler): """HTTP Request handler that calls out to 'git http-backend'.""" # Make rfile unbuffered -- we need to read one line and then pass # the rest to a subprocess, so we can't use buffered input. rbufsize = 0 def do_POST(self): self.run_backend() def do_GET(self): self.run_backend() def send_head(self): return self.run_backend() def log_request(self, code='-', size='-'): # Let's be quiet, the test suite is noisy enough already pass def run_backend(self): """Call out to git http-backend.""" # Based on CGIHTTPServer.CGIHTTPRequestHandler.run_cgi: # Copyright (c) 2001-2010 Python Software Foundation; All Rights Reserved # Licensed under the Python Software Foundation License. rest = self.path # find an explicit query string, if present. i = rest.rfind('?') if i >= 0: rest, query = rest[:i], rest[i+1:] else: query = '' env = copy.deepcopy(os.environ) env['SERVER_SOFTWARE'] = self.version_string() env['SERVER_NAME'] = self.server.server_name env['GATEWAY_INTERFACE'] = 'CGI/1.1' env['SERVER_PROTOCOL'] = self.protocol_version env['SERVER_PORT'] = str(self.server.server_port) env['GIT_PROJECT_ROOT'] = self.server.root_path env["GIT_HTTP_EXPORT_ALL"] = "1" env['REQUEST_METHOD'] = self.command uqrest = urllib.unquote(rest) env['PATH_INFO'] = uqrest env['SCRIPT_NAME'] = "/" if query: env['QUERY_STRING'] = query host = self.address_string() if host != self.client_address[0]: env['REMOTE_HOST'] = host env['REMOTE_ADDR'] = self.client_address[0] authorization = self.headers.getheader("authorization") if authorization: authorization = authorization.split() if len(authorization) == 2: import base64, binascii env['AUTH_TYPE'] = authorization[0] if authorization[0].lower() == "basic": try: authorization = base64.decodestring(authorization[1]) except binascii.Error: pass else: authorization = authorization.split(':') if len(authorization) == 2: env['REMOTE_USER'] = authorization[0] # XXX REMOTE_IDENT if self.headers.typeheader is None: env['CONTENT_TYPE'] = self.headers.type else: env['CONTENT_TYPE'] = self.headers.typeheader length = self.headers.getheader('content-length') if length: env['CONTENT_LENGTH'] = length referer = self.headers.getheader('referer') if referer: env['HTTP_REFERER'] = referer accept = [] for line in self.headers.getallmatchingheaders('accept'): if line[:1] in "\t\n\r ": accept.append(line.strip()) else: accept = accept + line[7:].split(',') env['HTTP_ACCEPT'] = ','.join(accept) ua = self.headers.getheader('user-agent') if ua: env['HTTP_USER_AGENT'] = ua co = filter(None, self.headers.getheaders('cookie')) if co: env['HTTP_COOKIE'] = ', '.join(co) # XXX Other HTTP_* headers # Since we're setting the env in the parent, provide empty # values to override previously set values for k in ('QUERY_STRING', 'REMOTE_HOST', 'CONTENT_LENGTH', 'HTTP_USER_AGENT', 'HTTP_COOKIE', 'HTTP_REFERER'): env.setdefault(k, "") self.send_response(200, "Script output follows") decoded_query = query.replace('+', ' ') try: nbytes = int(length) except (TypeError, ValueError): nbytes = 0 if self.command.lower() == "post" and nbytes > 0: data = self.rfile.read(nbytes) else: data = None # throw away additional data [see bug #427345] while select.select([self.rfile._sock], [], [], 0)[0]: if not self.rfile._sock.recv(1): break args = ['http-backend'] if '=' not in decoded_query: args.append(decoded_query) stdout = run_git_or_fail(args, input=data, env=env, stderr=subprocess.PIPE) self.wfile.write(stdout) class HTTPGitServer(BaseHTTPServer.HTTPServer): allow_reuse_address = True def __init__(self, server_address, root_path): BaseHTTPServer.HTTPServer.__init__(self, server_address, GitHTTPRequestHandler) self.root_path = root_path def get_url(self): return 'http://%s:%s/' % (self.server_name, self.server_port) if not getattr(HTTPGitServer, 'shutdown', None): _HTTPGitServer = HTTPGitServer class TCPGitServer(ShutdownServerMixIn, HTTPGitServer): """Subclass of HTTPGitServer that can be shut down.""" def __init__(self, *args, **kwargs): # BaseServer is old-style so we have to call both __init__s ShutdownServerMixIn.__init__(self) _HTTPGitServer.__init__(self, *args, **kwargs) class DulwichHttpClientTest(CompatTestCase, DulwichClientTestBase): min_git_version = (1, 7, 0, 2) def setUp(self): CompatTestCase.setUp(self) DulwichClientTestBase.setUp(self) self._httpd = HTTPGitServer(("localhost", 0), self.gitroot) self.addCleanup(self._httpd.shutdown) threading.Thread(target=self._httpd.serve_forever).start() run_git_or_fail(['config', 'http.uploadpack', 'true'], cwd=self.dest) run_git_or_fail(['config', 'http.receivepack', 'true'], cwd=self.dest) def tearDown(self): DulwichClientTestBase.tearDown(self) CompatTestCase.tearDown(self) def _client(self): return client.HttpGitClient(self._httpd.get_url()) def _build_path(self, path): return path def test_archive(self): raise SkipTest("exporting archives not supported over http") diff --git a/dulwich/tests/compat/utils.py b/dulwich/tests/compat/utils.py index afab0ee7..500d8d82 100644 --- a/dulwich/tests/compat/utils.py +++ b/dulwich/tests/compat/utils.py @@ -1,224 +1,228 @@ # utils.py -- Git compatibility utilities # Copyright (C) 2010 Google, Inc. # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; version 2 # of the License or (at your option) any later version of # the License. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, # MA 02110-1301, USA. """Utilities for interacting with cgit.""" import errno import os import socket import subprocess import tempfile import time from dulwich.repo import Repo from dulwich.protocol import TCP_GIT_PORT from dulwich.tests import ( + get_safe_env, SkipTest, TestCase, ) _DEFAULT_GIT = 'git' _VERSION_LEN = 4 _REPOS_DATA_DIR = os.path.abspath(os.path.join( os.path.dirname(__file__), os.pardir, 'data', 'repos')) def git_version(git_path=_DEFAULT_GIT): """Attempt to determine the version of git currently installed. :param git_path: Path to the git executable; defaults to the version in the system path. :return: A tuple of ints of the form (major, minor, point, sub-point), or None if no git installation was found. """ try: output = run_git_or_fail(['--version'], git_path=git_path) except OSError: return None version_prefix = 'git version ' if not output.startswith(version_prefix): return None parts = output[len(version_prefix):].split('.') nums = [] for part in parts: try: nums.append(int(part)) except ValueError: break while len(nums) < _VERSION_LEN: nums.append(0) return tuple(nums[:_VERSION_LEN]) def require_git_version(required_version, git_path=_DEFAULT_GIT): """Require git version >= version, or skip the calling test. :param required_version: A tuple of ints of the form (major, minor, point, sub-point); ommitted components default to 0. :param git_path: Path to the git executable; defaults to the version in the system path. :raise ValueError: if the required version tuple has too many parts. :raise SkipTest: if no suitable git version was found at the given path. """ found_version = git_version(git_path=git_path) if found_version is None: raise SkipTest('Test requires git >= %s, but c git not found' % (required_version, )) if len(required_version) > _VERSION_LEN: raise ValueError('Invalid version tuple %s, expected %i parts' % (required_version, _VERSION_LEN)) required_version = list(required_version) while len(found_version) < len(required_version): required_version.append(0) required_version = tuple(required_version) if found_version < required_version: required_version = '.'.join(map(str, required_version)) found_version = '.'.join(map(str, found_version)) raise SkipTest('Test requires git >= %s, found %s' % (required_version, found_version)) def run_git(args, git_path=_DEFAULT_GIT, input=None, capture_stdout=False, **popen_kwargs): """Run a git command. Input is piped from the input parameter and output is sent to the standard streams, unless capture_stdout is set. :param args: A list of args to the git command. :param git_path: Path to to the git executable. :param input: Input data to be sent to stdin. :param capture_stdout: Whether to capture and return stdout. :param popen_kwargs: Additional kwargs for subprocess.Popen; stdin/stdout args are ignored. :return: A tuple of (returncode, stdout contents). If capture_stdout is False, None will be returned as stdout contents. :raise OSError: if the git executable was not found. """ + + env = get_safe_env(popen_kwargs.pop('env', None)) + args = [git_path] + args popen_kwargs['stdin'] = subprocess.PIPE if capture_stdout: popen_kwargs['stdout'] = subprocess.PIPE else: popen_kwargs.pop('stdout', None) - p = subprocess.Popen(args, **popen_kwargs) + p = subprocess.Popen(args, env=env, **popen_kwargs) stdout, stderr = p.communicate(input=input) return (p.returncode, stdout) def run_git_or_fail(args, git_path=_DEFAULT_GIT, input=None, **popen_kwargs): """Run a git command, capture stdout/stderr, and fail if git fails.""" popen_kwargs['stderr'] = subprocess.STDOUT returncode, stdout = run_git(args, git_path=git_path, input=input, capture_stdout=True, **popen_kwargs) if returncode != 0: raise AssertionError("git with args %r failed with %d" % ( args, returncode)) return stdout def import_repo_to_dir(name): """Import a repo from a fast-export file in a temporary directory. These are used rather than binary repos for compat tests because they are more compact an human-editable, and we already depend on git. :param name: The name of the repository export file, relative to dulwich/tests/data/repos. :returns: The path to the imported repository. """ temp_dir = tempfile.mkdtemp() export_path = os.path.join(_REPOS_DATA_DIR, name) temp_repo_dir = os.path.join(temp_dir, name) export_file = open(export_path, 'rb') run_git_or_fail(['init', '--quiet', '--bare', temp_repo_dir]) run_git_or_fail(['fast-import'], input=export_file.read(), cwd=temp_repo_dir) export_file.close() return temp_repo_dir def import_repo(name): """Import a repo from a fast-export file in a temporary directory. :param name: The name of the repository export file, relative to dulwich/tests/data/repos. :returns: An initialized Repo object that lives in a temporary directory. """ return Repo(import_repo_to_dir(name)) def check_for_daemon(limit=10, delay=0.1, timeout=0.1, port=TCP_GIT_PORT): """Check for a running TCP daemon. Defaults to checking 10 times with a delay of 0.1 sec between tries. :param limit: Number of attempts before deciding no daemon is running. :param delay: Delay between connection attempts. :param timeout: Socket timeout for connection attempts. :param port: Port on which we expect the daemon to appear. :returns: A boolean, true if a daemon is running on the specified port, false if not. """ for _ in xrange(limit): time.sleep(delay) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(delay) try: s.connect(('localhost', port)) s.close() return True except socket.error, e: if getattr(e, 'errno', False) and e.errno != errno.ECONNREFUSED: raise elif e.args[0] != errno.ECONNREFUSED: raise return False class CompatTestCase(TestCase): """Test case that requires git for compatibility checks. Subclasses can change the git version required by overriding min_git_version. """ min_git_version = (1, 5, 0) def setUp(self): super(CompatTestCase, self).setUp() require_git_version(self.min_git_version) def assertReposEqual(self, repo1, repo2): self.assertEqual(repo1.get_refs(), repo2.get_refs()) self.assertEqual(sorted(set(repo1.object_store)), sorted(set(repo2.object_store))) def assertReposNotEqual(self, repo1, repo2): refs1 = repo1.get_refs() objs1 = set(repo1.object_store) refs2 = repo2.get_refs() objs2 = set(repo2.object_store) self.assertFalse(refs1 == refs2 and objs1 == objs2)