diff --git a/dulwich/cli.py b/dulwich/cli.py index 975b8bbb..583adf6c 100755 --- a/dulwich/cli.py +++ b/dulwich/cli.py @@ -1,723 +1,732 @@ #!/usr/bin/python3 -u # # dulwich - Simple command-line interface to Dulwich # Copyright (C) 2008-2011 Jelmer Vernooij # vim: expandtab # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Simple command-line interface to Dulwich> This is a very simple command-line wrapper for Dulwich. It is by no means intended to be a full-blown Git command-line interface but just a way to test Dulwich. """ import os import sys from getopt import getopt import optparse import signal from typing import Dict -def signal_int(signal, frame): - sys.exit(1) - - -def signal_quit(signal, frame): - import pdb - pdb.set_trace() - from dulwich import porcelain from dulwich.client import get_transport_and_path from dulwich.errors import ApplyDeltaError from dulwich.index import Index from dulwich.pack import Pack, sha_to_hex from dulwich.patch import write_tree_diff from dulwich.repo import Repo +def signal_int(signal, frame): + sys.exit(1) + + +def signal_quit(signal, frame): + import pdb + pdb.set_trace() + + class Command(object): """A Dulwich subcommand.""" def run(self, args): """Run the command.""" raise NotImplementedError(self.run) class cmd_archive(Command): def run(self, args): parser = optparse.OptionParser() parser.add_option("--remote", type=str, help="Retrieve archive from specified remote repo") options, args = parser.parse_args(args) committish = args.pop(0) if options.remote: client, path = get_transport_and_path(options.remote) - client.archive(path, committish, sys.stdout.write, - write_error=sys.stderr.write) + client.archive( + path, committish, sys.stdout.write, + write_error=sys.stderr.write) else: - porcelain.archive('.', committish, outstream=sys.stdout, + porcelain.archive( + '.', committish, outstream=sys.stdout, errstream=sys.stderr) class cmd_add(Command): def run(self, args): opts, args = getopt(args, "", []) porcelain.add(".", paths=args) class cmd_rm(Command): def run(self, args): opts, args = getopt(args, "", []) porcelain.rm(".", paths=args) class cmd_fetch_pack(Command): def run(self, args): opts, args = getopt(args, "", ["all"]) opts = dict(opts) client, path = get_transport_and_path(args.pop(0)) r = Repo(".") if "--all" in opts: determine_wants = r.object_store.determine_wants_all else: - determine_wants = lambda x: [y for y in args if not y in r.object_store] + def determine_wants(x): + return [y for y in args if y not in r.object_store] client.fetch(path, r, determine_wants) class cmd_fetch(Command): def run(self, args): opts, args = getopt(args, "", []) opts = dict(opts) client, path = get_transport_and_path(args.pop(0)) r = Repo(".") - if "--all" in opts: - determine_wants = r.object_store.determine_wants_all refs = client.fetch(path, r, progress=sys.stdout.write) print("Remote refs:") for item in refs.items(): print("%s -> %s" % item) class cmd_fsck(Command): def run(self, args): opts, args = getopt(args, "", []) opts = dict(opts) for (obj, msg) in porcelain.fsck('.'): print("%s: %s" % (obj, msg)) class cmd_log(Command): def run(self, args): parser = optparse.OptionParser() parser.add_option("--reverse", dest="reverse", action="store_true", help="Reverse order in which entries are printed") - parser.add_option("--name-status", dest="name_status", action="store_true", + parser.add_option("--name-status", dest="name_status", + action="store_true", help="Print name/status for each changed file") options, args = parser.parse_args(args) porcelain.log(".", paths=args, reverse=options.reverse, name_status=options.name_status, outstream=sys.stdout) class cmd_diff(Command): def run(self, args): opts, args = getopt(args, "", []) if args == []: print("Usage: dulwich diff COMMITID") sys.exit(1) r = Repo(".") commit_id = args[0] commit = r[commit_id] parent_commit = r[commit.parents[0]] - write_tree_diff(sys.stdout, r.object_store, parent_commit.tree, commit.tree) + write_tree_diff( + sys.stdout, r.object_store, parent_commit.tree, commit.tree) class cmd_dump_pack(Command): def run(self, args): opts, args = getopt(args, "", []) if args == []: print("Usage: dulwich dump-pack FILENAME") sys.exit(1) basename, _ = os.path.splitext(args[0]) x = Pack(basename) print("Object names checksum: %s" % x.name()) print("Checksum: %s" % sha_to_hex(x.get_stored_checksum())) if not x.check(): print("CHECKSUM DOES NOT MATCH") print("Length: %d" % len(x)) for name in x: try: print("\t%s" % x[name]) except KeyError as k: print("\t%s: Unable to resolve base %s" % (name, k)) except ApplyDeltaError as e: print("\t%s: Unable to apply delta: %r" % (name, e)) class cmd_dump_index(Command): def run(self, args): opts, args = getopt(args, "", []) if args == []: print("Usage: dulwich dump-index FILENAME") sys.exit(1) filename = args[0] idx = Index(filename) for o in idx: print(o, idx[o]) class cmd_init(Command): def run(self, args): opts, args = getopt(args, "", ["bare"]) opts = dict(opts) if args == []: path = os.getcwd() else: path = args[0] porcelain.init(path, bare=("--bare" in opts)) class cmd_clone(Command): def run(self, args): parser = optparse.OptionParser() parser.add_option("--bare", dest="bare", help="Whether to create a bare repository.", action="store_true") parser.add_option("--depth", dest="depth", type=int, help="Depth at which to fetch") options, args = parser.parse_args(args) if args == []: print("usage: dulwich clone host:path [PATH]") sys.exit(1) source = args.pop(0) if len(args) > 0: target = args.pop(0) else: target = None porcelain.clone(source, target, bare=options.bare, depth=options.depth) class cmd_commit(Command): def run(self, args): opts, args = getopt(args, "", ["message"]) opts = dict(opts) porcelain.commit(".", message=opts["--message"]) class cmd_commit_tree(Command): def run(self, args): opts, args = getopt(args, "", ["message"]) if args == []: print("usage: dulwich commit-tree tree") sys.exit(1) opts = dict(opts) porcelain.commit_tree(".", tree=args[0], message=opts["--message"]) class cmd_update_server_info(Command): def run(self, args): porcelain.update_server_info(".") class cmd_symbolic_ref(Command): def run(self, args): opts, args = getopt(args, "", ["ref-name", "force"]) if not args: print("Usage: dulwich symbolic-ref REF_NAME [--force]") sys.exit(1) ref_name = args.pop(0) porcelain.symbolic_ref(".", ref_name=ref_name, force='--force' in args) class cmd_show(Command): def run(self, args): opts, args = getopt(args, "", []) porcelain.show(".", args) class cmd_diff_tree(Command): def run(self, args): opts, args = getopt(args, "", []) if len(args) < 2: print("Usage: dulwich diff-tree OLD-TREE NEW-TREE") sys.exit(1) porcelain.diff_tree(".", args[0], args[1]) class cmd_rev_list(Command): def run(self, args): opts, args = getopt(args, "", []) if len(args) < 1: print('Usage: dulwich rev-list COMMITID...') sys.exit(1) porcelain.rev_list('.', args) class cmd_tag(Command): def run(self, args): parser = optparse.OptionParser() - parser.add_option("-a", "--annotated", help="Create an annotated tag.", action="store_true") - parser.add_option("-s", "--sign", help="Sign the annotated tag.", action="store_true") + parser.add_option( + "-a", "--annotated", help="Create an annotated tag.", + action="store_true") + parser.add_option( + "-s", "--sign", help="Sign the annotated tag.", + action="store_true") options, args = parser.parse_args(args) porcelain.tag_create( '.', args[0], annotated=options.annotated, sign=options.sign) class cmd_repack(Command): def run(self, args): opts, args = getopt(args, "", []) opts = dict(opts) porcelain.repack('.') class cmd_reset(Command): def run(self, args): opts, args = getopt(args, "", ["hard", "soft", "mixed"]) opts = dict(opts) mode = "" if "--hard" in opts: mode = "hard" elif "--soft" in opts: mode = "soft" elif "--mixed" in opts: mode = "mixed" porcelain.reset('.', mode=mode, *args) class cmd_daemon(Command): def run(self, args): from dulwich import log_utils from dulwich.protocol import TCP_GIT_PORT parser = optparse.OptionParser() parser.add_option("-l", "--listen_address", dest="listen_address", default="localhost", help="Binding IP address.") parser.add_option("-p", "--port", dest="port", type=int, default=TCP_GIT_PORT, help="Binding TCP port.") options, args = parser.parse_args(args) log_utils.default_logging_config() if len(args) >= 1: gitdir = args[0] else: gitdir = '.' from dulwich import porcelain porcelain.daemon(gitdir, address=options.listen_address, port=options.port) class cmd_web_daemon(Command): def run(self, args): from dulwich import log_utils parser = optparse.OptionParser() parser.add_option("-l", "--listen_address", dest="listen_address", default="", help="Binding IP address.") parser.add_option("-p", "--port", dest="port", type=int, default=8000, help="Binding TCP port.") options, args = parser.parse_args(args) log_utils.default_logging_config() if len(args) >= 1: gitdir = args[0] else: gitdir = '.' from dulwich import porcelain porcelain.web_daemon(gitdir, address=options.listen_address, port=options.port) class cmd_write_tree(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) sys.stdout.write('%s\n' % porcelain.write_tree('.')) class cmd_receive_pack(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) if len(args) >= 1: gitdir = args[0] else: gitdir = '.' porcelain.receive_pack(gitdir) class cmd_upload_pack(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) if len(args) >= 1: gitdir = args[0] else: gitdir = '.' porcelain.upload_pack(gitdir) class cmd_status(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) if len(args) >= 1: gitdir = args[0] else: gitdir = '.' status = porcelain.status(gitdir) if any(names for (kind, names) in status.staged.items()): sys.stdout.write("Changes to be committed:\n\n") for kind, names in status.staged.items(): for name in names: sys.stdout.write("\t%s: %s\n" % ( kind, name.decode(sys.getfilesystemencoding()))) sys.stdout.write("\n") if status.unstaged: sys.stdout.write("Changes not staged for commit:\n\n") for name in status.unstaged: - sys.stdout.write("\t%s\n" % - name.decode(sys.getfilesystemencoding())) + sys.stdout.write( + "\t%s\n" % name.decode(sys.getfilesystemencoding())) sys.stdout.write("\n") if status.untracked: sys.stdout.write("Untracked files:\n\n") for name in status.untracked: sys.stdout.write("\t%s\n" % name) sys.stdout.write("\n") class cmd_ls_remote(Command): def run(self, args): opts, args = getopt(args, '', []) if len(args) < 1: print('Usage: dulwich ls-remote URL') sys.exit(1) refs = porcelain.ls_remote(args[0]) for ref in sorted(refs): sys.stdout.write("%s\t%s\n" % (ref, refs[ref])) class cmd_ls_tree(Command): def run(self, args): parser = optparse.OptionParser() parser.add_option("-r", "--recursive", action="store_true", help="Recusively list tree contents.") parser.add_option("--name-only", action="store_true", help="Only display name.") options, args = parser.parse_args(args) try: treeish = args.pop(0) except IndexError: treeish = None porcelain.ls_tree( '.', treeish, outstream=sys.stdout, recursive=options.recursive, name_only=options.name_only) class cmd_pack_objects(Command): def run(self, args): opts, args = getopt(args, '', ['stdout']) opts = dict(opts) - if len(args) < 1 and not '--stdout' in args: + if len(args) < 1 and '--stdout' not in args: print('Usage: dulwich pack-objects basename') sys.exit(1) - object_ids = [l.strip() for l in sys.stdin.readlines()] + object_ids = [line.strip() for line in sys.stdin.readlines()] basename = args[0] if '--stdout' in opts: packf = getattr(sys.stdout, 'buffer', sys.stdout) idxf = None close = [] else: packf = open(basename + '.pack', 'w') idxf = open(basename + '.idx', 'w') close = [packf, idxf] porcelain.pack_objects('.', object_ids, packf, idxf) for f in close: f.close() class cmd_pull(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) try: from_location = args[0] except IndexError: from_location = None porcelain.pull('.', from_location) class cmd_push(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) if len(args) < 2: print("Usage: dulwich push TO-LOCATION REFSPEC..") sys.exit(1) to_location = args[0] refspecs = args[1:] porcelain.push('.', to_location, refspecs) class cmd_remote_add(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) porcelain.remote_add('.', args[0], args[1]) class SuperCommand(Command): - subcommands: Dict[str, Command] = {} + subcommands = {} # type: Dict[str, Command] def run(self, args): if not args: - print("Supported subcommands: %s" % ', '.join(self.subcommands.keys())) + print("Supported subcommands: %s" % + ', '.join(self.subcommands.keys())) return False cmd = args[0] try: cmd_kls = self.subcommands[cmd] except KeyError: print('No such subcommand: %s' % args[0]) return False return cmd_kls().run(args[1:]) class cmd_remote(SuperCommand): subcommands = { "add": cmd_remote_add, } class cmd_check_ignore(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) ret = 1 for path in porcelain.check_ignore('.', args): print(path) ret = 0 return ret class cmd_check_mailmap(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) for arg in args: canonical_identity = porcelain.check_mailmap('.', arg) print(canonical_identity) class cmd_stash_list(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) for i, entry in porcelain.stash_list('.'): print("stash@{%d}: %s" % (i, entry.message.rstrip('\n'))) class cmd_stash_push(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) porcelain.stash_push('.') print("Saved working directory and index state") class cmd_stash_pop(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) porcelain.stash_pop('.') print("Restrored working directory and index state") class cmd_stash(SuperCommand): subcommands = { "list": cmd_stash_list, "pop": cmd_stash_pop, "push": cmd_stash_push, } class cmd_ls_files(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) for name in porcelain.ls_files('.'): print(name) class cmd_describe(Command): def run(self, args): parser = optparse.OptionParser() options, args = parser.parse_args(args) print(porcelain.describe('.')) class cmd_help(Command): def run(self, args): parser = optparse.OptionParser() parser.add_option("-a", "--all", dest="all", action="store_true", help="List all commands.") options, args = parser.parse_args(args) if options.all: print('Available commands:') for cmd in sorted(commands): print(' %s' % cmd) else: print("""\ The dulwich command line tool is currently a very basic frontend for the Dulwich python module. For full functionality, please see the API reference. For a list of supported commands, see 'dulwich help -a'. """) commands = { "add": cmd_add, "archive": cmd_archive, "check-ignore": cmd_check_ignore, "check-mailmap": cmd_check_mailmap, "clone": cmd_clone, "commit": cmd_commit, "commit-tree": cmd_commit_tree, "describe": cmd_describe, "daemon": cmd_daemon, "diff": cmd_diff, "diff-tree": cmd_diff_tree, "dump-pack": cmd_dump_pack, "dump-index": cmd_dump_index, "fetch-pack": cmd_fetch_pack, "fetch": cmd_fetch, "fsck": cmd_fsck, "help": cmd_help, "init": cmd_init, "log": cmd_log, "ls-files": cmd_ls_files, "ls-remote": cmd_ls_remote, "ls-tree": cmd_ls_tree, "pack-objects": cmd_pack_objects, "pull": cmd_pull, "push": cmd_push, "receive-pack": cmd_receive_pack, "remote": cmd_remote, "repack": cmd_repack, "reset": cmd_reset, "rev-list": cmd_rev_list, "rm": cmd_rm, "show": cmd_show, "stash": cmd_stash, "status": cmd_status, "symbolic-ref": cmd_symbolic_ref, "tag": cmd_tag, "update-server-info": cmd_update_server_info, "upload-pack": cmd_upload_pack, "web-daemon": cmd_web_daemon, "write-tree": cmd_write_tree, } def main(argv=None): if len(argv) < 1: - print("Usage: %s <%s> [OPTIONS...]" % (sys.argv[0], "|".join(commands.keys()))) + print("Usage: dulwich <%s> [OPTIONS...]" % ("|".join(commands.keys()))) return 1 cmd = argv[0] try: cmd_kls = commands[cmd] except KeyError: print("No such subcommand: %s" % cmd) return 1 # TODO(jelmer): Return non-0 on errors return cmd_kls().run(sys.argv[1:]) if __name__ == '__main__': if 'DULWICH_PDB' in os.environ: signal.signal(signal.SIGQUIT, signal_quit) signal.signal(signal.SIGINT, signal_int) sys.exit(main(sys.argv[1:])) diff --git a/dulwich/tests/compat/test_client.py b/dulwich/tests/compat/test_client.py index b0c3c961..fd62764a 100644 --- a/dulwich/tests/compat/test_client.py +++ b/dulwich/tests/compat/test_client.py @@ -1,612 +1,613 @@ # test_client.py -- Compatibilty tests for git client. # Copyright (C) 2010 Google, Inc. # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Compatibilty tests between the Dulwich client and the cgit server.""" import copy from io import BytesIO import os import select import signal import stat import subprocess import sys import tarfile import tempfile import threading from urllib.parse import unquote import http.server from dulwich import ( client, errors, file, index, protocol, objects, repo, ) from dulwich.tests import ( SkipTest, expectedFailure, ) from dulwich.tests.compat.utils import ( CompatTestCase, check_for_daemon, import_repo_to_dir, rmtree_ro, run_git_or_fail, _DEFAULT_GIT, ) if sys.platform == 'win32': import ctypes class DulwichClientTestBase(object): """Tests for client/server compatibility.""" def setUp(self): self.gitroot = os.path.dirname( import_repo_to_dir('server_new.export').rstrip(os.sep)) 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): rmtree_ro(self.gitroot) def assertDestEqualsSrc(self): repo_dir = os.path.join(self.gitroot, 'server_new.export') dest_repo_dir = os.path.join(self.gitroot, 'dest') with repo.Repo(repo_dir) as src: with repo.Repo(dest_repo_dir) as 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') with repo.Repo(srcpath) as src: sendrefs = dict(src.get_refs()) del sendrefs[b'HEAD'] c.send_pack(self._build_path('/dest'), lambda _: sendrefs, src.generate_pack_data) 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() @staticmethod def _add_file(repo, tree_id, filename, contents): tree = repo[tree_id] blob = objects.Blob() blob.data = contents.encode('utf-8') repo.object_store.add_object(blob) tree.add(filename.encode('utf-8'), stat.S_IFREG | 0o644, blob.id) repo.object_store.add_object(tree) return tree.id def test_send_pack_from_shallow_clone(self): c = self._client() server_new_path = os.path.join(self.gitroot, 'server_new.export') run_git_or_fail(['config', 'http.uploadpack', 'true'], cwd=server_new_path) run_git_or_fail(['config', 'http.receivepack', 'true'], cwd=server_new_path) remote_path = self._build_path('/server_new.export') with repo.Repo(self.dest) as local: result = c.fetch(remote_path, local, depth=1) for r in result.refs.items(): local.refs.set_if_equals(r[0], None, r[1]) tree_id = local[local.head()].tree for filename, contents in [('bar', 'bar contents'), ('zop', 'zop contents')]: tree_id = self._add_file(local, tree_id, filename, contents) commit_id = local.do_commit( message=b"add " + filename.encode('utf-8'), committer=b"Joe Example ", tree=tree_id) sendrefs = dict(local.get_refs()) del sendrefs[b'HEAD'] c.send_pack(remote_path, lambda _: sendrefs, local.generate_pack_data) with repo.Repo(server_new_path) as remote: self.assertEqual(remote.head(), commit_id) def test_send_without_report_status(self): c = self._client() c._send_capabilities.remove(b'report-status') srcpath = os.path.join(self.gitroot, 'server_new.export') with repo.Repo(srcpath) as src: sendrefs = dict(src.get_refs()) del sendrefs[b'HEAD'] c.send_pack(self._build_path('/dest'), lambda _: sendrefs, src.generate_pack_data) self.assertDestEqualsSrc() def make_dummy_commit(self, dest): b = objects.Blob.from_string(b'hi') dest.object_store.add_object(b) t = index.commit_tree(dest.object_store, [(b'hi', b.id, 0o100644)]) c = objects.Commit() c.author = c.committer = b'Foo Bar ' c.author_time = c.commit_time = 0 c.author_timezone = c.commit_timezone = 0 c.message = b'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, src): sendrefs = dict(src.get_refs()) del sendrefs[b'HEAD'] return sendrefs, src.generate_pack_data def test_send_pack_one_error(self): dest, dummy_commit = self.disable_ff_and_make_dummy_commit() dest.refs[b'refs/heads/master'] = dummy_commit repo_dir = os.path.join(self.gitroot, 'server_new.export') with repo.Repo(repo_dir) as src: sendrefs, gen_pack = self.compute_send(src) c = self._client() try: c.send_pack(self._build_path('/dest'), lambda _: sendrefs, gen_pack) except errors.UpdateRefsError as e: self.assertEqual('refs/heads/master failed to update', e.args[0]) self.assertEqual({b'refs/heads/branch': b'ok', b'refs/heads/master': b'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 branch, master = b'refs/heads/branch', b'refs/heads/master' dest.refs[branch] = dest.refs[master] = dummy repo_dir = os.path.join(self.gitroot, 'server_new.export') with repo.Repo(repo_dir) as src: sendrefs, gen_pack = self.compute_send(src) c = self._client() try: c.send_pack(self._build_path('/dest'), lambda _: sendrefs, gen_pack) except errors.UpdateRefsError as e: self.assertIn( str(e), ['{0}, {1} failed to update'.format( branch.decode('ascii'), master.decode('ascii')), '{1}, {0} failed to update'.format( branch.decode('ascii'), master.decode('ascii'))]) self.assertEqual({branch: b'non-fast-forward', master: b'non-fast-forward'}, e.ref_status) def test_archive(self): c = self._client() f = BytesIO() c.archive(self._build_path('/server_new.export'), b'HEAD', f.write) f.seek(0) tf = tarfile.open(fileobj=f) self.assertEqual(['baz', 'foo'], tf.getnames()) def test_fetch_pack(self): c = self._client() with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest: result = c.fetch(self._build_path('/server_new.export'), dest) for r in result.refs.items(): dest.refs.set_if_equals(r[0], None, r[1]) self.assertDestEqualsSrc() def test_fetch_pack_depth(self): c = self._client() with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest: result = c.fetch(self._build_path('/server_new.export'), dest, depth=1) for r in result.refs.items(): dest.refs.set_if_equals(r[0], None, r[1]) self.assertEqual( dest.get_shallow(), set([b'35e0b59e187dd72a0af294aedffc213eaa4d03ff', b'514dc6d3fbfe77361bcaef320c4d21b72bc10be9'])) def test_repeat(self): c = self._client() with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest: result = c.fetch(self._build_path('/server_new.export'), dest) for r in result.refs.items(): dest.refs.set_if_equals(r[0], None, r[1]) self.assertDestEqualsSrc() result = c.fetch(self._build_path('/server_new.export'), dest) for r in result.refs.items(): dest.refs.set_if_equals(r[0], None, r[1]) self.assertDestEqualsSrc() def test_incremental_fetch_pack(self): self.test_fetch_pack() dest, dummy = self.disable_ff_and_make_dummy_commit() dest.refs[b'refs/heads/master'] = dummy c = self._client() repo_dir = os.path.join(self.gitroot, 'server_new.export') with repo.Repo(repo_dir) as dest: result = c.fetch(self._build_path('/dest'), dest) for r in result.refs.items(): dest.refs.set_if_equals(r[0], None, r[1]) self.assertDestEqualsSrc() def test_fetch_pack_no_side_band_64k(self): c = self._client() c._fetch_capabilities.remove(b'side-band-64k') with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest: result = c.fetch(self._build_path('/server_new.export'), dest) for r in result.refs.items(): dest.refs.set_if_equals(r[0], None, r[1]) self.assertDestEqualsSrc() def test_fetch_pack_zero_sha(self): # zero sha1s are already present on the client, and should # be ignored c = self._client() with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest: result = c.fetch( self._build_path('/server_new.export'), dest, lambda refs: [protocol.ZERO_SHA]) for r in result.refs.items(): dest.refs.set_if_equals(r[0], None, r[1]) def test_send_remove_branch(self): with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest: dummy_commit = self.make_dummy_commit(dest) dest.refs[b'refs/heads/master'] = dummy_commit dest.refs[b'refs/heads/abranch'] = dummy_commit sendrefs = dict(dest.refs) sendrefs[b'refs/heads/abranch'] = b"00" * 20 del sendrefs[b'HEAD'] def gen_pack(have, want, ofs_delta=False): return 0, [] c = self._client() self.assertEqual(dest.refs[b"refs/heads/abranch"], dummy_commit) c.send_pack( self._build_path('/dest'), lambda _: sendrefs, gen_pack) self.assertFalse(b"refs/heads/abranch" in dest.refs) def test_send_new_branch_empty_pack(self): with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest: dummy_commit = self.make_dummy_commit(dest) dest.refs[b'refs/heads/master'] = dummy_commit dest.refs[b'refs/heads/abranch'] = dummy_commit sendrefs = {b'refs/heads/bbranch': dummy_commit} + def gen_pack(have, want, ofs_delta=False): return 0, [] c = self._client() self.assertEqual(dest.refs[b"refs/heads/abranch"], dummy_commit) c.send_pack( self._build_path('/dest'), lambda _: sendrefs, gen_pack) self.assertEqual(dummy_commit, dest.refs[b"refs/heads/abranch"]) def test_get_refs(self): c = self._client() refs = c.get_refs(self._build_path('/server_new.export')) repo_dir = os.path.join(self.gitroot, 'server_new.export') with repo.Repo(repo_dir) as dest: self.assertDictEqual(dest.refs.as_dict(), 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() args = [_DEFAULT_GIT, 'daemon', '--verbose', '--export-all', '--pid-file=%s' % self.pidfile, '--base-path=%s' % self.gitroot, '--enable=receive-pack', '--enable=upload-archive', '--listen=localhost', '--reuseaddr', self.gitroot] self.process = subprocess.Popen( args, cwd=self.gitroot, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if not check_for_daemon(): raise SkipTest('git-daemon failed to start') def tearDown(self): with open(self.pidfile) as f: pid = int(f.read().strip()) if sys.platform == 'win32': PROCESS_TERMINATE = 1 handle = ctypes.windll.kernel32.OpenProcess( PROCESS_TERMINATE, False, pid) ctypes.windll.kernel32.TerminateProcess(handle, -1) ctypes.windll.kernel32.CloseHandle(handle) else: try: os.kill(pid, signal.SIGKILL) os.unlink(self.pidfile) except (OSError, IOError): pass self.process.wait() self.process.stdout.close() self.process.stderr.close() DulwichClientTestBase.tearDown(self) CompatTestCase.tearDown(self) def _client(self): return client.TCPGitClient('localhost') def _build_path(self, path): return path if sys.platform == 'win32': @expectedFailure def test_fetch_pack_no_side_band_64k(self): DulwichClientTestBase.test_fetch_pack_no_side_band_64k(self) class TestSSHVendor(object): @staticmethod def run_command(host, command, username=None, port=None, password=None, key_filename=None): cmd, path = command.split(' ') cmd = cmd.split('-', 1) path = path.replace("'", "") p = subprocess.Popen(cmd + [path], bufsize=0, 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) def tearDown(self): DulwichClientTestBase.tearDown(self) CompatTestCase.tearDown(self) def _client(self): return client.SubprocessGitClient() def _build_path(self, path): return self.gitroot + path class GitHTTPRequestHandler(http.server.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 = 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.get("authorization") if authorization: authorization = authorization.split() if len(authorization) == 2: import base64 import 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 content_type = self.headers.get('content-type') if content_type: env['CONTENT_TYPE'] = content_type length = self.headers.get('content-length') if length: env['CONTENT_LENGTH'] = length referer = self.headers.get('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.get('user-agent') if ua: env['HTTP_USER_AGENT'] = ua co = self.headers.get('cookie') if co: env['HTTP_COOKIE'] = 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.wfile.write(b"HTTP/1.1 200 Script output follows\r\n") self.wfile.write( ("Server: %s\r\n" % self.server.server_name).encode('ascii')) self.wfile.write( ("Date: %s\r\n" % self.date_time_string()).encode('ascii')) 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 env['CONTENT_LENGTH'] = '0' # 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(http.server.HTTPServer): allow_reuse_address = True def __init__(self, server_address, root_path): http.server.HTTPServer.__init__( self, server_address, GitHTTPRequestHandler) self.root_path = root_path self.server_name = "localhost" def get_url(self): return 'http://%s:%s/' % (self.server_name, self.server_port) 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) self._httpd.shutdown() self._httpd.socket.close() 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")