diff --git a/dulwich/config.py b/dulwich/config.py index 082b6e60..db82e0f2 100644 --- a/dulwich/config.py +++ b/dulwich/config.py @@ -1,434 +1,426 @@ # config.py - Reading and writing Git config files # Copyright (C) 2011-2013 Jelmer Vernooij # # 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) a later version. # # 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. """Reading and writing Git configuration files. TODO: * preserve formatting when updating configuration files * treat subsection names as case-insensitive for [branch.foo] style subsections """ import errno import os from collections import ( OrderedDict, MutableMapping, ) from dulwich.file import GitFile class Config(object): """A Git configuration.""" def get(self, section, name): """Retrieve the contents of a configuration setting. :param section: Tuple with section name and optional subsection namee :param subsection: Subsection name :return: Contents of the setting :raise KeyError: if the value is not set """ raise NotImplementedError(self.get) def get_boolean(self, section, name, default=None): """Retrieve a configuration setting as boolean. :param section: Tuple with section name and optional subsection namee :param name: Name of the setting, including section and possible subsection. :return: Contents of the setting :raise KeyError: if the value is not set """ try: value = self.get(section, name) except KeyError: return default if value.lower() == b"true": return True elif value.lower() == b"false": return False raise ValueError("not a valid boolean string: %r" % value) def set(self, section, name, value): """Set a configuration value. :param section: Tuple with section name and optional subsection namee :param name: Name of the configuration value, including section and optional subsection :param: Value of the setting """ raise NotImplementedError(self.set) def iteritems(self, section): """Iterate over the configuration pairs for a specific section. :param section: Tuple with section name and optional subsection namee :return: Iterator over (name, value) pairs """ raise NotImplementedError(self.iteritems) def itersections(self): """Iterate over the sections. :return: Iterator over section tuples """ raise NotImplementedError(self.itersections) class ConfigDict(Config, MutableMapping): """Git configuration stored in a dictionary.""" def __init__(self, values=None): """Create a new ConfigDict.""" if values is None: values = OrderedDict() self._values = values def __repr__(self): return "%s(%r)" % (self.__class__.__name__, self._values) def __eq__(self, other): return ( isinstance(other, self.__class__) and other._values == self._values) def __getitem__(self, key): return self._values.__getitem__(key) def __setitem__(self, key, value): return self._values.__setitem__(key, value) def __delitem__(self, key): return self._values.__delitem__(key) def __iter__(self): return self._values.__iter__() def __len__(self): return self._values.__len__() @classmethod def _parse_setting(cls, name): parts = name.split(".") if len(parts) == 3: return (parts[0], parts[1], parts[2]) else: return (parts[0], None, parts[1]) def get(self, section, name): if not isinstance(section, tuple): section = (section, ) if len(section) > 1: try: return self._values[section][name] except KeyError: pass return self._values[(section[0],)][name] def set(self, section, name, value): if not isinstance(section, tuple): section = (section, ) self._values.setdefault(section, OrderedDict())[name] = value def iteritems(self, section): return self._values.get(section, OrderedDict()).items() def itersections(self): return self._values.keys() def _format_string(value): if (value.startswith(b" ") or value.startswith(b"\t") or value.endswith(b" ") or value.endswith(b"\t")): return b'"' + _escape_value(value) + b'"' return _escape_value(value) _ESCAPE_TABLE = { ord(b"\\"): ord(b"\\"), ord(b"\""): ord(b"\""), ord(b"n"): ord(b"\n"), ord(b"t"): ord(b"\t"), ord(b"b"): ord(b"\b"), } _COMMENT_CHARS = [ord(b"#"), ord(b";")] _WHITESPACE_CHARS = [ord(b"\t"), ord(b" ")] def _parse_string(value): value = bytearray(value.strip()) ret = bytearray() whitespace = bytearray() in_quotes = False i = 0 while i < len(value): c = value[i] if c == ord(b"\\"): i += 1 try: v = _ESCAPE_TABLE[value[i]] except IndexError: raise ValueError( "escape character in %r at %d before end of string" % (value, i)) except KeyError: raise ValueError( "escape character followed by unknown character %s at %d in %r" % (value[i], i, value)) if whitespace: ret.extend(whitespace) whitespace = bytearray() ret.append(v) elif c == ord(b"\""): in_quotes = (not in_quotes) elif c in _COMMENT_CHARS and not in_quotes: # the rest of the line is a comment break elif c in _WHITESPACE_CHARS: whitespace.append(c) else: if whitespace: ret.extend(whitespace) whitespace = bytearray() ret.append(c) i += 1 if in_quotes: raise ValueError("missing end quote") return bytes(ret) -def _unescape_value(value): - """Unescape a value.""" - ret = bytearray() - i = 0 - - return ret - - def _escape_value(value): """Escape a value.""" return value.replace(b"\\", b"\\\\").replace(b"\n", b"\\n").replace(b"\t", b"\\t").replace(b"\"", b"\\\"") def _check_variable_name(name): for i in range(len(name)): c = name[i:i+1] if not c.isalnum() and c != b'-': return False return True def _check_section_name(name): for i in range(len(name)): c = name[i:i+1] if not c.isalnum() and c not in (b'-', b'.'): return False return True def _strip_comments(line): line = line.split(b"#")[0] line = line.split(b";")[0] return line class ConfigFile(ConfigDict): """A Git configuration file, like .git/config or ~/.gitconfig. """ @classmethod def from_file(cls, f): """Read configuration from a file-like object.""" ret = cls() section = None setting = None for lineno, line in enumerate(f.readlines()): line = line.lstrip() if setting is None: # Parse section header ("[bla]") if len(line) > 0 and line[:1] == b"[": line = _strip_comments(line).rstrip() last = line.index(b"]") if last == -1: raise ValueError("expected trailing ]") pts = line[1:last].split(b" ", 1) line = line[last+1:] pts[0] = pts[0].lower() if len(pts) == 2: if pts[1][:1] != b"\"" or pts[1][-1:] != b"\"": raise ValueError( "Invalid subsection %r" % pts[1]) else: pts[1] = pts[1][1:-1] if not _check_section_name(pts[0]): raise ValueError("invalid section name %r" % pts[0]) section = (pts[0], pts[1]) else: if not _check_section_name(pts[0]): raise ValueError("invalid section name %r" % pts[0]) pts = pts[0].split(b".", 1) if len(pts) == 2: section = (pts[0], pts[1]) else: section = (pts[0], ) ret._values[section] = OrderedDict() if _strip_comments(line).strip() == b"": continue if section is None: raise ValueError("setting %r without section" % line) try: setting, value = line.split(b"=", 1) except ValueError: setting = line value = b"true" setting = setting.strip().lower() if not _check_variable_name(setting): raise ValueError("invalid variable name %s" % setting) if value.endswith(b"\\\n"): value = value[:-2] continuation = True else: continuation = False value = _parse_string(value) ret._values[section][setting] = value if not continuation: setting = None else: # continuation line if line.endswith(b"\\\n"): line = line[:-2] continuation = True else: continuation = False value = _parse_string(line) ret._values[section][setting] += value if not continuation: setting = None return ret @classmethod def from_path(cls, path): """Read configuration from a file on disk.""" with GitFile(path, 'rb') as f: ret = cls.from_file(f) ret.path = path return ret def write_to_path(self, path=None): """Write configuration to a file on disk.""" if path is None: path = self.path with GitFile(path, 'wb') as f: self.write_to_file(f) def write_to_file(self, f): """Write configuration to a file-like object.""" for section, values in self._values.items(): try: section_name, subsection_name = section except ValueError: (section_name, ) = section subsection_name = None if subsection_name is None: f.write(b"[" + section_name + b"]\n") else: f.write(b"[" + section_name + b" \"" + subsection_name + b"\"]\n") for key, value in values.items(): if value is True: value = b"true" elif value is False: value = b"false" else: value = _escape_value(value) f.write(b"\t" + key + b" = " + value + b"\n") class StackedConfig(Config): """Configuration which reads from multiple config files..""" def __init__(self, backends, writable=None): self.backends = backends self.writable = writable def __repr__(self): return "<%s for %r>" % (self.__class__.__name__, self.backends) @classmethod def default_backends(cls): """Retrieve the default configuration. See git-config(1) for details on the files searched. """ paths = [] paths.append(os.path.expanduser("~/.gitconfig")) xdg_config_home = os.environ.get( "XDG_CONFIG_HOME", os.path.expanduser("~/.config/"), ) paths.append(os.path.join(xdg_config_home, "git", "config")) if "GIT_CONFIG_NOSYSTEM" not in os.environ: paths.append("/etc/gitconfig") backends = [] for path in paths: try: cf = ConfigFile.from_path(path) except (IOError, OSError) as e: if e.errno != errno.ENOENT: raise else: continue backends.append(cf) return backends def get(self, section, name): for backend in self.backends: try: return backend.get(section, name) except KeyError: pass raise KeyError(name) def set(self, section, name, value): if self.writable is None: raise NotImplementedError(self.set) return self.writable.set(section, name, value) def parse_submodules(config): """Parse a gitmodules GitConfig file, returning submodules. :param config: A `ConfigFile` :return: list of tuples (submodule path, url, name), where name is quoted part of the section's name. """ for section in config.keys(): section_kind, section_name = section if section_kind == b'submodule': sm_path = config.get(section, b'path') sm_url = config.get(section, b'url') yield (sm_path, sm_url, section_name) diff --git a/dulwich/contrib/paramiko_vendor.py b/dulwich/contrib/paramiko_vendor.py index 8d7ee153..e8716ad4 100644 --- a/dulwich/contrib/paramiko_vendor.py +++ b/dulwich/contrib/paramiko_vendor.py @@ -1,138 +1,137 @@ # paramiko_vendor.py -- paramiko implementation of the SSHVendor interface # Copyright (C) 2013 Aaron O'Mullan # # 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; either version 2 # or (at your option) a 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. """Paramiko SSH support for Dulwich. To use this implementation as the SSH implementation in Dulwich, override the dulwich.client.get_ssh_vendor attribute: >>> from dulwich import client as _mod_client >>> from dulwich.contrib.paramiko_vendor import ParamikoSSHVendor >>> _mod_client.get_ssh_vendor = ParamikoSSHVendor This implementation is experimental and does not have any tests. """ import paramiko import paramiko.client -import subprocess import threading class _ParamikoWrapper(object): STDERR_READ_N = 2048 # 2k def __init__(self, client, channel, progress_stderr=None): self.client = client self.channel = channel self.progress_stderr = progress_stderr self.should_monitor = bool(progress_stderr) or True self.monitor_thread = None self.stderr = b'' # Channel must block self.channel.setblocking(True) # Start if self.should_monitor: self.monitor_thread = threading.Thread( target=self.monitor_stderr) self.monitor_thread.start() def monitor_stderr(self): while self.should_monitor: # Block and read data = self.read_stderr(self.STDERR_READ_N) # Socket closed if not data: self.should_monitor = False break # Emit data if self.progress_stderr: self.progress_stderr(data) # Append to buffer self.stderr += data def stop_monitoring(self): # Stop StdErr thread if self.should_monitor: self.should_monitor = False self.monitor_thread.join() # Get left over data data = self.channel.in_stderr_buffer.empty() self.stderr += data def can_read(self): return self.channel.recv_ready() def write(self, data): return self.channel.sendall(data) def read_stderr(self, n): return self.channel.recv_stderr(n) def read(self, n=None): data = self.channel.recv(n) data_len = len(data) # Closed socket if not data: return # Read more if needed if n and data_len < n: diff_len = n - data_len return data + self.read(diff_len) return data def close(self): self.channel.close() self.stop_monitoring() class ParamikoSSHVendor(object): def __init__(self): self.ssh_kwargs = {} def run_command(self, host, command, username=None, port=None, progress_stderr=None): if not isinstance(command, bytes): raise TypeError(command) # Paramiko needs an explicit port. None is not valid if port is None: port = 22 client = paramiko.SSHClient() policy = paramiko.client.MissingHostKeyPolicy() client.set_missing_host_key_policy(policy) client.connect(host, username=username, port=port, **self.ssh_kwargs) # Open SSH session channel = client.get_transport().open_session() # Run commands channel.exec_command(command) return _ParamikoWrapper( client, channel, progress_stderr=progress_stderr) diff --git a/dulwich/server.py b/dulwich/server.py index 861b498b..3ce81939 100644 --- a/dulwich/server.py +++ b/dulwich/server.py @@ -1,1108 +1,1107 @@ # server.py -- Implementation of the server side git protocols # Copyright (C) 2008 John Carr # Coprygith (C) 2011-2012 Jelmer Vernooij # # 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 # 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. """Git smart network protocol server implementation. For more detailed implementation on the network protocol, see the Documentation/technical directory in the cgit distribution, and in particular: * Documentation/technical/protocol-capabilities.txt * Documentation/technical/pack-protocol.txt Currently supported capabilities: * include-tag * thin-pack * multi_ack_detailed * multi_ack * side-band-64k * ofs-delta * no-progress * report-status * delete-refs * shallow """ import collections import os import socket import sys import zlib try: import SocketServer except ImportError: import socketserver as SocketServer from dulwich.errors import ( ApplyDeltaError, ChecksumMismatch, GitProtocolError, NotGitRepository, UnexpectedCommandError, ObjectFormatException, ) from dulwich import log_utils from dulwich.objects import ( Commit, valid_hexsha, ) from dulwich.pack import ( write_pack_objects, ) from dulwich.protocol import ( BufferedPktLineWriter, capability_agent, CAPABILITIES_REF, CAPABILITY_DELETE_REFS, CAPABILITY_INCLUDE_TAG, CAPABILITY_MULTI_ACK_DETAILED, CAPABILITY_MULTI_ACK, CAPABILITY_NO_DONE, CAPABILITY_NO_PROGRESS, CAPABILITY_OFS_DELTA, CAPABILITY_QUIET, CAPABILITY_REPORT_STATUS, CAPABILITY_SHALLOW, CAPABILITY_SIDE_BAND_64K, CAPABILITY_THIN_PACK, COMMAND_DEEPEN, COMMAND_DONE, COMMAND_HAVE, COMMAND_SHALLOW, COMMAND_UNSHALLOW, COMMAND_WANT, MULTI_ACK, MULTI_ACK_DETAILED, Protocol, ProtocolFile, ReceivableProtocol, SIDE_BAND_CHANNEL_DATA, SIDE_BAND_CHANNEL_PROGRESS, SIDE_BAND_CHANNEL_FATAL, SINGLE_ACK, TCP_GIT_PORT, ZERO_SHA, ack_type, extract_capabilities, extract_want_line_capabilities, ) from dulwich.refs import ( write_info_refs, ) from dulwich.repo import ( Repo, ) logger = log_utils.getLogger(__name__) class Backend(object): """A backend for the Git smart server implementation.""" def open_repository(self, path): """Open the repository at a path. :param path: Path to the repository :raise NotGitRepository: no git repository was found at path :return: Instance of BackendRepo """ raise NotImplementedError(self.open_repository) class BackendRepo(object): """Repository abstraction used by the Git server. The methods required here are a subset of those provided by dulwich.repo.Repo. """ object_store = None refs = None def get_refs(self): """ Get all the refs in the repository :return: dict of name -> sha """ raise NotImplementedError def get_peeled(self, name): """Return the cached peeled value of a ref, if available. :param name: Name of the ref to peel :return: The peeled value of the ref. If the ref is known not point to a tag, this will be the SHA the ref refers to. If no cached information about a tag is available, this method may return None, but it should attempt to peel the tag if possible. """ return None def fetch_objects(self, determine_wants, graph_walker, progress, get_tagged=None): """ Yield the objects required for a list of commits. :param progress: is a callback to send progress messages to the client :param get_tagged: Function that returns a dict of pointed-to sha -> tag sha for including tags. """ raise NotImplementedError class DictBackend(Backend): """Trivial backend that looks up Git repositories in a dictionary.""" def __init__(self, repos): self.repos = repos def open_repository(self, path): logger.debug('Opening repository at %s', path) try: return self.repos[path] except KeyError: raise NotGitRepository( "No git repository was found at %(path)s" % dict(path=path) ) class FileSystemBackend(Backend): """Simple backend that looks up Git repositories in the local file system.""" def __init__(self, root=os.sep): super(FileSystemBackend, self).__init__() self.root = (os.path.abspath(root) + os.sep).replace(os.sep * 2, os.sep) def open_repository(self, path): logger.debug('opening repository at %s', path) abspath = os.path.abspath(os.path.join(self.root, path)) + os.sep normcase_abspath = os.path.normcase(abspath) normcase_root = os.path.normcase(self.root) if not normcase_abspath.startswith(normcase_root): raise NotGitRepository("Path %r not inside root %r" % (path, self.root)) return Repo(abspath) class Handler(object): """Smart protocol command handler base class.""" def __init__(self, backend, proto, http_req=None): self.backend = backend self.proto = proto self.http_req = http_req def handle(self): raise NotImplementedError(self.handle) class PackHandler(Handler): """Protocol handler for packs.""" def __init__(self, backend, proto, http_req=None): super(PackHandler, self).__init__(backend, proto, http_req) self._client_capabilities = None # Flags needed for the no-done capability self._done_received = False @classmethod def capability_line(cls): return b"".join([b" " + c for c in cls.capabilities()]) @classmethod def capabilities(cls): raise NotImplementedError(cls.capabilities) @classmethod def innocuous_capabilities(cls): return (CAPABILITY_INCLUDE_TAG, CAPABILITY_THIN_PACK, CAPABILITY_NO_PROGRESS, CAPABILITY_OFS_DELTA, capability_agent()) @classmethod def required_capabilities(cls): """Return a list of capabilities that we require the client to have.""" return [] def set_client_capabilities(self, caps): allowable_caps = set(self.innocuous_capabilities()) allowable_caps.update(self.capabilities()) for cap in caps: if cap not in allowable_caps: raise GitProtocolError('Client asked for capability %s that ' 'was not advertised.' % cap) for cap in self.required_capabilities(): if cap not in caps: raise GitProtocolError('Client does not support required ' 'capability %s.' % cap) self._client_capabilities = set(caps) logger.info('Client capabilities: %s', caps) def has_capability(self, cap): if self._client_capabilities is None: raise GitProtocolError('Server attempted to access capability %s ' 'before asking client' % cap) return cap in self._client_capabilities def notify_done(self): self._done_received = True class UploadPackHandler(PackHandler): """Protocol handler for uploading a pack to the client.""" def __init__(self, backend, args, proto, http_req=None, advertise_refs=False): super(UploadPackHandler, self).__init__(backend, proto, http_req=http_req) self.repo = backend.open_repository(args[0]) self._graph_walker = None self.advertise_refs = advertise_refs # A state variable for denoting that the have list is still # being processed, and the client is not accepting any other # data (such as side-band, see the progress method here). self._processing_have_lines = False @classmethod def capabilities(cls): return (CAPABILITY_MULTI_ACK_DETAILED, CAPABILITY_MULTI_ACK, CAPABILITY_SIDE_BAND_64K, CAPABILITY_THIN_PACK, CAPABILITY_OFS_DELTA, CAPABILITY_NO_PROGRESS, CAPABILITY_INCLUDE_TAG, CAPABILITY_SHALLOW, CAPABILITY_NO_DONE) @classmethod def required_capabilities(cls): return (CAPABILITY_SIDE_BAND_64K, CAPABILITY_THIN_PACK, CAPABILITY_OFS_DELTA) def progress(self, message): if self.has_capability(CAPABILITY_NO_PROGRESS) or self._processing_have_lines: return self.proto.write_sideband(SIDE_BAND_CHANNEL_PROGRESS, message) def get_tagged(self, refs=None, repo=None): """Get a dict of peeled values of tags to their original tag shas. :param refs: dict of refname -> sha of possible tags; defaults to all of the backend's refs. :param repo: optional Repo instance for getting peeled refs; defaults to the backend's repo, if available :return: dict of peeled_sha -> tag_sha, where tag_sha is the sha of a tag whose peeled value is peeled_sha. """ if not self.has_capability(CAPABILITY_INCLUDE_TAG): return {} if refs is None: refs = self.repo.get_refs() if repo is None: repo = getattr(self.repo, "repo", None) if repo is None: # Bail if we don't have a Repo available; this is ok since # clients must be able to handle if the server doesn't include # all relevant tags. # TODO: fix behavior when missing return {} tagged = {} for name, sha in refs.items(): peeled_sha = repo.get_peeled(name) if peeled_sha != sha: tagged[peeled_sha] = sha return tagged def handle(self): write = lambda x: self.proto.write_sideband(SIDE_BAND_CHANNEL_DATA, x) graph_walker = ProtocolGraphWalker(self, self.repo.object_store, self.repo.get_peeled) objects_iter = self.repo.fetch_objects( graph_walker.determine_wants, graph_walker, self.progress, get_tagged=self.get_tagged) # Note the fact that client is only processing responses related # to the have lines it sent, and any other data (including side- # band) will be be considered a fatal error. self._processing_have_lines = True # Did the process short-circuit (e.g. in a stateless RPC call)? Note # that the client still expects a 0-object pack in most cases. # Also, if it also happens that the object_iter is instantiated # with a graph walker with an implementation that talks over the # wire (which is this instance of this class) this will actually # iterate through everything and write things out to the wire. if len(objects_iter) == 0: return # The provided haves are processed, and it is safe to send side- # band data now. self._processing_have_lines = False if not graph_walker.handle_done( not self.has_capability(CAPABILITY_NO_DONE), self._done_received): return self.progress(b"dul-daemon says what\n") self.progress(("counting objects: %d, done.\n" % len(objects_iter)).encode('ascii')) write_pack_objects(ProtocolFile(None, write), objects_iter) self.progress(b"how was that, then?\n") # we are done self.proto.write_pkt_line(None) def _split_proto_line(line, allowed): """Split a line read from the wire. :param line: The line read from the wire. :param allowed: An iterable of command names that should be allowed. Command names not listed below as possible return values will be ignored. If None, any commands from the possible return values are allowed. :return: a tuple having one of the following forms: ('want', obj_id) ('have', obj_id) ('done', None) (None, None) (for a flush-pkt) :raise UnexpectedCommandError: if the line cannot be parsed into one of the allowed return values. """ if not line: fields = [None] else: fields = line.rstrip(b'\n').split(b' ', 1) command = fields[0] if allowed is not None and command not in allowed: raise UnexpectedCommandError(command) if len(fields) == 1 and command in (COMMAND_DONE, None): return (command, None) elif len(fields) == 2: if command in (COMMAND_WANT, COMMAND_HAVE, COMMAND_SHALLOW, COMMAND_UNSHALLOW): if not valid_hexsha(fields[1]): raise GitProtocolError("Invalid sha") return tuple(fields) elif command == COMMAND_DEEPEN: return command, int(fields[1]) raise GitProtocolError('Received invalid line from client: %r' % line) def _find_shallow(store, heads, depth): """Find shallow commits according to a given depth. :param store: An ObjectStore for looking up objects. :param heads: Iterable of head SHAs to start walking from. :param depth: The depth of ancestors to include. A depth of one includes only the heads themselves. :return: A tuple of (shallow, not_shallow), sets of SHAs that should be considered shallow and unshallow according to the arguments. Note that these sets may overlap if a commit is reachable along multiple paths. """ parents = {} def get_parents(sha): result = parents.get(sha, None) if not result: result = store[sha].parents parents[sha] = result return result todo = [] # stack of (sha, depth) for head_sha in heads: obj = store.peel_sha(head_sha) if isinstance(obj, Commit): todo.append((obj.id, 1)) not_shallow = set() shallow = set() while todo: sha, cur_depth = todo.pop() if cur_depth < depth: not_shallow.add(sha) new_depth = cur_depth + 1 todo.extend((p, new_depth) for p in get_parents(sha)) else: shallow.add(sha) return shallow, not_shallow def _want_satisfied(store, haves, want, earliest): o = store[want] pending = collections.deque([o]) while pending: commit = pending.popleft() if commit.id in haves: return True if commit.type_name != b"commit": # non-commit wants are assumed to be satisfied continue for parent in commit.parents: parent_obj = store[parent] # TODO: handle parents with later commit times than children if parent_obj.commit_time >= earliest: pending.append(parent_obj) return False def _all_wants_satisfied(store, haves, wants): """Check whether all the current wants are satisfied by a set of haves. :param store: Object store to retrieve objects from :param haves: A set of commits we know the client has. :param wants: A set of commits the client wants :note: Wants are specified with set_wants rather than passed in since in the current interface they are determined outside this class. """ haves = set(haves) if haves: earliest = min([store[h].commit_time for h in haves]) else: earliest = 0 - unsatisfied_wants = set() for want in wants: if not _want_satisfied(store, haves, want, earliest): return False return True class ProtocolGraphWalker(object): """A graph walker that knows the git protocol. As a graph walker, this class implements ack(), next(), and reset(). It also contains some base methods for interacting with the wire and walking the commit tree. The work of determining which acks to send is passed on to the implementation instance stored in _impl. The reason for this is that we do not know at object creation time what ack level the protocol requires. A call to set_ack_level() is required to set up the implementation, before any calls to next() or ack() are made. """ def __init__(self, handler, object_store, get_peeled): self.handler = handler self.store = object_store self.get_peeled = get_peeled self.proto = handler.proto self.http_req = handler.http_req self.advertise_refs = handler.advertise_refs self._wants = [] self.shallow = set() self.client_shallow = set() self.unshallow = set() self._cached = False self._cache = [] self._cache_index = 0 self._impl = None def determine_wants(self, heads): """Determine the wants for a set of heads. The given heads are advertised to the client, who then specifies which refs he wants using 'want' lines. This portion of the protocol is the same regardless of ack type, and in fact is used to set the ack type of the ProtocolGraphWalker. If the client has the 'shallow' capability, this method also reads and responds to the 'shallow' and 'deepen' lines from the client. These are not part of the wants per se, but they set up necessary state for walking the graph. Additionally, later code depends on this method consuming everything up to the first 'have' line. :param heads: a dict of refname->SHA1 to advertise :return: a list of SHA1s requested by the client """ values = set(heads.values()) if self.advertise_refs or not self.http_req: for i, (ref, sha) in enumerate(sorted(heads.items())): line = sha + b' ' + ref if not i: line += b'\x00' + self.handler.capability_line() self.proto.write_pkt_line(line + b'\n') peeled_sha = self.get_peeled(ref) if peeled_sha != sha: self.proto.write_pkt_line(peeled_sha + b' ' + ref + b'^{}\n') # i'm done.. self.proto.write_pkt_line(None) if self.advertise_refs: return [] # Now client will sending want want want commands want = self.proto.read_pkt_line() if not want: return [] line, caps = extract_want_line_capabilities(want) self.handler.set_client_capabilities(caps) self.set_ack_type(ack_type(caps)) allowed = (COMMAND_WANT, COMMAND_SHALLOW, COMMAND_DEEPEN, None) command, sha = _split_proto_line(line, allowed) want_revs = [] while command == COMMAND_WANT: if sha not in values: raise GitProtocolError( 'Client wants invalid object %s' % sha) want_revs.append(sha) command, sha = self.read_proto_line(allowed) self.set_wants(want_revs) if command in (COMMAND_SHALLOW, COMMAND_DEEPEN): self.unread_proto_line(command, sha) self._handle_shallow_request(want_revs) if self.http_req and self.proto.eof(): # The client may close the socket at this point, expecting a # flush-pkt from the server. We might be ready to send a packfile at # this point, so we need to explicitly short-circuit in this case. return [] return want_revs def unread_proto_line(self, command, value): if isinstance(value, int): value = str(value).encode('ascii') self.proto.unread_pkt_line(command + b' ' + value) def ack(self, have_ref): if len(have_ref) != 40: raise ValueError("invalid sha %r" % have_ref) return self._impl.ack(have_ref) def reset(self): self._cached = True self._cache_index = 0 def next(self): if not self._cached: if not self._impl and self.http_req: return None return next(self._impl) self._cache_index += 1 if self._cache_index > len(self._cache): return None return self._cache[self._cache_index] __next__ = next def read_proto_line(self, allowed): """Read a line from the wire. :param allowed: An iterable of command names that should be allowed. :return: A tuple of (command, value); see _split_proto_line. :raise UnexpectedCommandError: If an error occurred reading the line. """ return _split_proto_line(self.proto.read_pkt_line(), allowed) def _handle_shallow_request(self, wants): while True: command, val = self.read_proto_line((COMMAND_DEEPEN, COMMAND_SHALLOW)) if command == COMMAND_DEEPEN: depth = val break self.client_shallow.add(val) self.read_proto_line((None,)) # consume client's flush-pkt shallow, not_shallow = _find_shallow(self.store, wants, depth) # Update self.shallow instead of reassigning it since we passed a # reference to it before this method was called. self.shallow.update(shallow - not_shallow) new_shallow = self.shallow - self.client_shallow unshallow = self.unshallow = not_shallow & self.client_shallow for sha in sorted(new_shallow): self.proto.write_pkt_line(COMMAND_SHALLOW + b' ' + sha) for sha in sorted(unshallow): self.proto.write_pkt_line(COMMAND_UNSHALLOW + b' ' + sha) self.proto.write_pkt_line(None) def notify_done(self): # relay the message down to the handler. self.handler.notify_done() def send_ack(self, sha, ack_type=b''): if ack_type: ack_type = b' ' + ack_type self.proto.write_pkt_line(b'ACK ' + sha + ack_type + b'\n') def send_nak(self): self.proto.write_pkt_line(b'NAK\n') def handle_done(self, done_required, done_received): # Delegate this to the implementation. return self._impl.handle_done(done_required, done_received) def set_wants(self, wants): self._wants = wants def all_wants_satisfied(self, haves): """Check whether all the current wants are satisfied by a set of haves. :param haves: A set of commits we know the client has. :note: Wants are specified with set_wants rather than passed in since in the current interface they are determined outside this class. """ return _all_wants_satisfied(self.store, haves, self._wants) def set_ack_type(self, ack_type): impl_classes = { MULTI_ACK: MultiAckGraphWalkerImpl, MULTI_ACK_DETAILED: MultiAckDetailedGraphWalkerImpl, SINGLE_ACK: SingleAckGraphWalkerImpl, } self._impl = impl_classes[ack_type](self) _GRAPH_WALKER_COMMANDS = (COMMAND_HAVE, COMMAND_DONE, None) class SingleAckGraphWalkerImpl(object): """Graph walker implementation that speaks the single-ack protocol.""" def __init__(self, walker): self.walker = walker self._common = [] def ack(self, have_ref): if not self._common: self.walker.send_ack(have_ref) self._common.append(have_ref) def next(self): command, sha = self.walker.read_proto_line(_GRAPH_WALKER_COMMANDS) if command in (None, COMMAND_DONE): # defer the handling of done self.walker.notify_done() return None elif command == COMMAND_HAVE: return sha __next__ = next def handle_done(self, done_required, done_received): if not self._common: self.walker.send_nak() if done_required and not done_received: # we are not done, especially when done is required; skip # the pack for this request and especially do not handle # the done. return False if not done_received and not self._common: # Okay we are not actually done then since the walker picked # up no haves. This is usually triggered when client attempts # to pull from a source that has no common base_commit. # See: test_server.MultiAckDetailedGraphWalkerImplTestCase.\ # test_multi_ack_stateless_nodone return False return True class MultiAckGraphWalkerImpl(object): """Graph walker implementation that speaks the multi-ack protocol.""" def __init__(self, walker): self.walker = walker self._found_base = False self._common = [] def ack(self, have_ref): self._common.append(have_ref) if not self._found_base: self.walker.send_ack(have_ref, b'continue') if self.walker.all_wants_satisfied(self._common): self._found_base = True # else we blind ack within next def next(self): while True: command, sha = self.walker.read_proto_line(_GRAPH_WALKER_COMMANDS) if command is None: self.walker.send_nak() # in multi-ack mode, a flush-pkt indicates the client wants to # flush but more have lines are still coming continue elif command == COMMAND_DONE: self.walker.notify_done() return None elif command == COMMAND_HAVE: if self._found_base: # blind ack self.walker.send_ack(sha, b'continue') return sha __next__ = next def handle_done(self, done_required, done_received): if done_required and not done_received: # we are not done, especially when done is required; skip # the pack for this request and especially do not handle # the done. return False if not done_received and not self._common: # Okay we are not actually done then since the walker picked # up no haves. This is usually triggered when client attempts # to pull from a source that has no common base_commit. # See: test_server.MultiAckDetailedGraphWalkerImplTestCase.\ # test_multi_ack_stateless_nodone return False # don't nak unless no common commits were found, even if not # everything is satisfied if self._common: self.walker.send_ack(self._common[-1]) else: self.walker.send_nak() return True class MultiAckDetailedGraphWalkerImpl(object): """Graph walker implementation speaking the multi-ack-detailed protocol.""" def __init__(self, walker): self.walker = walker self._common = [] def ack(self, have_ref): # Should only be called iff have_ref is common self._common.append(have_ref) self.walker.send_ack(have_ref, b'common') def next(self): while True: command, sha = self.walker.read_proto_line(_GRAPH_WALKER_COMMANDS) if command is None: if self.walker.all_wants_satisfied(self._common): self.walker.send_ack(self._common[-1], b'ready') self.walker.send_nak() if self.walker.http_req: # The HTTP version of this request a flush-pkt always # signifies an end of request, so we also return # nothing here as if we are done (but not really, as # it depends on whether no-done capability was # specified and that's handled in handle_done which # may or may not call post_nodone_check depending on # that). return None elif command == COMMAND_DONE: # Let the walker know that we got a done. self.walker.notify_done() break elif command == COMMAND_HAVE: # return the sha and let the caller ACK it with the # above ack method. return sha # don't nak unless no common commits were found, even if not # everything is satisfied __next__ = next def handle_done(self, done_required, done_received): if done_required and not done_received: # we are not done, especially when done is required; skip # the pack for this request and especially do not handle # the done. return False if not done_received and not self._common: # Okay we are not actually done then since the walker picked # up no haves. This is usually triggered when client attempts # to pull from a source that has no common base_commit. # See: test_server.MultiAckDetailedGraphWalkerImplTestCase.\ # test_multi_ack_stateless_nodone return False # don't nak unless no common commits were found, even if not # everything is satisfied if self._common: self.walker.send_ack(self._common[-1]) else: self.walker.send_nak() return True class ReceivePackHandler(PackHandler): """Protocol handler for downloading a pack from the client.""" def __init__(self, backend, args, proto, http_req=None, advertise_refs=False): super(ReceivePackHandler, self).__init__(backend, proto, http_req=http_req) self.repo = backend.open_repository(args[0]) self.advertise_refs = advertise_refs @classmethod def capabilities(cls): return (CAPABILITY_REPORT_STATUS, CAPABILITY_DELETE_REFS, CAPABILITY_QUIET, CAPABILITY_OFS_DELTA, CAPABILITY_SIDE_BAND_64K, CAPABILITY_NO_DONE) def _apply_pack(self, refs): all_exceptions = (IOError, OSError, ChecksumMismatch, ApplyDeltaError, AssertionError, socket.error, zlib.error, ObjectFormatException) status = [] will_send_pack = False for command in refs: if command[1] != ZERO_SHA: will_send_pack = True if will_send_pack: # TODO: more informative error messages than just the exception string try: recv = getattr(self.proto, "recv", None) self.repo.object_store.add_thin_pack(self.proto.read, recv) status.append((b'unpack', b'ok')) except all_exceptions as e: status.append((b'unpack', str(e).replace('\n', ''))) # The pack may still have been moved in, but it may contain broken # objects. We trust a later GC to clean it up. else: # The git protocol want to find a status entry related to unpack process # even if no pack data has been sent. status.append((b'unpack', b'ok')) for oldsha, sha, ref in refs: ref_status = b'ok' try: if sha == ZERO_SHA: if not CAPABILITY_DELETE_REFS in self.capabilities(): raise GitProtocolError( 'Attempted to delete refs without delete-refs ' 'capability.') try: self.repo.refs.remove_if_equals(ref, oldsha) except all_exceptions: ref_status = b'failed to delete' else: try: self.repo.refs.set_if_equals(ref, oldsha, sha) except all_exceptions: ref_status = b'failed to write' except KeyError as e: ref_status = b'bad ref' status.append((ref, ref_status)) return status def _report_status(self, status): if self.has_capability(CAPABILITY_SIDE_BAND_64K): writer = BufferedPktLineWriter( lambda d: self.proto.write_sideband(SIDE_BAND_CHANNEL_DATA, d)) write = writer.write def flush(): writer.flush() self.proto.write_pkt_line(None) else: write = self.proto.write_pkt_line flush = lambda: None for name, msg in status: if name == b'unpack': write(b'unpack ' + msg + b'\n') elif msg == b'ok': write(b'ok ' + name + b'\n') else: write(b'ng ' + name + b' ' + msg + b'\n') write(None) flush() def handle(self): if self.advertise_refs or not self.http_req: refs = sorted(self.repo.get_refs().items()) if not refs: refs = [(CAPABILITIES_REF, ZERO_SHA)] self.proto.write_pkt_line( refs[0][1] + b' ' + refs[0][0] + b'\0' + self.capability_line() + b'\n') for i in range(1, len(refs)): ref = refs[i] self.proto.write_pkt_line(ref[1] + b' ' + ref[0] + b'\n') self.proto.write_pkt_line(None) if self.advertise_refs: return client_refs = [] ref = self.proto.read_pkt_line() # if ref is none then client doesnt want to send us anything.. if ref is None: return ref, caps = extract_capabilities(ref) self.set_client_capabilities(caps) # client will now send us a list of (oldsha, newsha, ref) while ref: client_refs.append(ref.split()) ref = self.proto.read_pkt_line() # backend can now deal with this refs and read a pack using self.read status = self._apply_pack(client_refs) # when we have read all the pack from the client, send a status report # if the client asked for it if self.has_capability(CAPABILITY_REPORT_STATUS): self._report_status(status) class UploadArchiveHandler(Handler): def __init__(self, backend, proto, http_req=None): super(UploadArchiveHandler, self).__init__(backend, proto, http_req) def handle(self): # TODO(jelmer) raise NotImplementedError(self.handle) # Default handler classes for git services. DEFAULT_HANDLERS = { b'git-upload-pack': UploadPackHandler, b'git-receive-pack': ReceivePackHandler, # b'git-upload-archive': UploadArchiveHandler, } class TCPGitRequestHandler(SocketServer.StreamRequestHandler): def __init__(self, handlers, *args, **kwargs): self.handlers = handlers SocketServer.StreamRequestHandler.__init__(self, *args, **kwargs) def handle(self): proto = ReceivableProtocol(self.connection.recv, self.wfile.write) command, args = proto.read_cmd() logger.info('Handling %s request, args=%s', command, args) cls = self.handlers.get(command, None) if not callable(cls): raise GitProtocolError('Invalid service %s' % command) h = cls(self.server.backend, args, proto) h.handle() class TCPGitServer(SocketServer.TCPServer): allow_reuse_address = True serve = SocketServer.TCPServer.serve_forever def _make_handler(self, *args, **kwargs): return TCPGitRequestHandler(self.handlers, *args, **kwargs) def __init__(self, backend, listen_addr, port=TCP_GIT_PORT, handlers=None): self.handlers = dict(DEFAULT_HANDLERS) if handlers is not None: self.handlers.update(handlers) self.backend = backend logger.info('Listening for TCP connections on %s:%d', listen_addr, port) SocketServer.TCPServer.__init__(self, (listen_addr, port), self._make_handler) def verify_request(self, request, client_address): logger.info('Handling request from %s', client_address) return True def handle_error(self, request, client_address): logger.exception('Exception happened during processing of request ' 'from %s', client_address) def main(argv=sys.argv): """Entry point for starting a TCP git server.""" import optparse 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(argv) log_utils.default_logging_config() if len(args) > 1: gitdir = args[1] else: gitdir = '.' from dulwich import porcelain porcelain.daemon(gitdir, address=options.listen_address, port=options.port) def serve_command(handler_cls, argv=sys.argv, backend=None, inf=sys.stdin, outf=sys.stdout): """Serve a single command. This is mostly useful for the implementation of commands used by e.g. git+ssh. :param handler_cls: `Handler` class to use for the request :param argv: execv-style command-line arguments. Defaults to sys.argv. :param backend: `Backend` to use :param inf: File-like object to read from, defaults to standard input. :param outf: File-like object to write to, defaults to standard output. :return: Exit code for use with sys.exit. 0 on success, 1 on failure. """ if backend is None: backend = FileSystemBackend() def send_fn(data): outf.write(data) outf.flush() proto = Protocol(inf.read, send_fn) handler = handler_cls(backend, argv[1:], proto) # FIXME: Catch exceptions and write a single-line summary to outf. handler.handle() return 0 def generate_info_refs(repo): """Generate an info refs file.""" refs = repo.get_refs() return write_info_refs(refs, repo.object_store) def generate_objects_info_packs(repo): """Generate an index for for packs.""" for pack in repo.object_store.packs: yield b'P ' + pack.data.filename.encode(sys.getfilesystemencoding()) + b'\n' def update_server_info(repo): """Generate server info for dumb file access. This generates info/refs and objects/info/packs, similar to "git update-server-info". """ repo._put_named_file(os.path.join('info', 'refs'), b"".join(generate_info_refs(repo))) repo._put_named_file(os.path.join('objects', 'info', 'packs'), b"".join(generate_objects_info_packs(repo))) if __name__ == '__main__': main() diff --git a/dulwich/tests/compat/test_pack.py b/dulwich/tests/compat/test_pack.py index a62d20bc..f0cc772d 100644 --- a/dulwich/tests/compat/test_pack.py +++ b/dulwich/tests/compat/test_pack.py @@ -1,155 +1,154 @@ # test_pack.py -- Compatibility tests for git packs. # 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. """Compatibility tests for git packs.""" import binascii import os import re import shutil import tempfile from dulwich.pack import ( write_pack, ) from dulwich.objects import ( Blob, ) from dulwich.tests import ( SkipTest, ) from dulwich.tests.test_pack import ( a_sha, pack1_sha, PackTests, ) from dulwich.tests.compat.utils import ( require_git_version, run_git_or_fail, ) _NON_DELTA_RE = re.compile(b'non delta: (?P\d+) objects') def _git_verify_pack_object_list(output): pack_shas = set() for line in output.splitlines(): sha = line[:40] try: binascii.unhexlify(sha) except (TypeError, binascii.Error): continue # non-sha line pack_shas.add(sha) return pack_shas class TestPack(PackTests): """Compatibility tests for reading and writing pack files.""" def setUp(self): require_git_version((1, 5, 0)) super(TestPack, self).setUp() self._tempdir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, self._tempdir) def test_copy(self): with self.get_pack(pack1_sha) as origpack: self.assertSucceeds(origpack.index.check) pack_path = os.path.join(self._tempdir, "Elch") write_pack(pack_path, origpack.pack_tuples()) output = run_git_or_fail(['verify-pack', '-v', pack_path]) orig_shas = set(o.id for o in origpack.iterobjects()) self.assertEqual(orig_shas, _git_verify_pack_object_list(output)) def test_deltas_work(self): with self.get_pack(pack1_sha) as orig_pack: orig_blob = orig_pack[a_sha] new_blob = Blob() new_blob.data = orig_blob.data + b'x' all_to_pack = list(orig_pack.pack_tuples()) + [(new_blob, None)] pack_path = os.path.join(self._tempdir, 'pack_with_deltas') write_pack(pack_path, all_to_pack, deltify=True) output = run_git_or_fail(['verify-pack', '-v', pack_path]) self.assertEqual(set(x[0].id for x in all_to_pack), _git_verify_pack_object_list(output)) # We specifically made a new blob that should be a delta # against the blob a_sha, so make sure we really got only 3 # non-delta objects: got_non_delta = int(_NON_DELTA_RE.search(output).group('non_delta')) self.assertEqual( 3, got_non_delta, 'Expected 3 non-delta objects, got %d' % got_non_delta) def test_delta_medium_object(self): # This tests an object set that will have a copy operation # 2**20 in size. with self.get_pack(pack1_sha) as orig_pack: orig_blob = orig_pack[a_sha] new_blob = Blob() new_blob.data = orig_blob.data + (b'x' * 2 ** 20) new_blob_2 = Blob() new_blob_2.data = new_blob.data + b'y' all_to_pack = list(orig_pack.pack_tuples()) + [(new_blob, None), (new_blob_2, None)] pack_path = os.path.join(self._tempdir, 'pack_with_deltas') write_pack(pack_path, all_to_pack, deltify=True) output = run_git_or_fail(['verify-pack', '-v', pack_path]) self.assertEqual(set(x[0].id for x in all_to_pack), _git_verify_pack_object_list(output)) # We specifically made a new blob that should be a delta # against the blob a_sha, so make sure we really got only 3 # non-delta objects: got_non_delta = int(_NON_DELTA_RE.search(output).group('non_delta')) self.assertEqual( 3, got_non_delta, 'Expected 3 non-delta objects, got %d' % got_non_delta) # We expect one object to have a delta chain length of two # (new_blob_2), so let's verify that actually happens: self.assertIn(b'chain length = 2', output) # This test is SUPER slow: over 80 seconds on a 2012-era # laptop. This is because SequenceMatcher is worst-case quadratic # on the input size. It's impractical to produce deltas for # objects this large, but it's still worth doing the right thing # when it happens. def test_delta_large_object(self): # This tests an object set that will have a copy operation # 2**25 in size. This is a copy large enough that it requires # two copy operations in git's binary delta format. raise SkipTest('skipping slow, large test') with self.get_pack(pack1_sha) as orig_pack: - orig_blob = orig_pack[a_sha] new_blob = Blob() new_blob.data = 'big blob' + ('x' * 2 ** 25) new_blob_2 = Blob() new_blob_2.data = new_blob.data + 'y' all_to_pack = list(orig_pack.pack_tuples()) + [(new_blob, None), (new_blob_2, None)] pack_path = os.path.join(self._tempdir, "pack_with_deltas") write_pack(pack_path, all_to_pack, deltify=True) output = run_git_or_fail(['verify-pack', '-v', pack_path]) self.assertEqual(set(x[0].id for x in all_to_pack), _git_verify_pack_object_list(output)) # We specifically made a new blob that should be a delta # against the blob a_sha, so make sure we really got only 4 # non-delta objects: got_non_delta = int(_NON_DELTA_RE.search(output).group('non_delta')) self.assertEqual( 4, got_non_delta, 'Expected 4 non-delta objects, got %d' % got_non_delta) diff --git a/dulwich/tests/test_config.py b/dulwich/tests/test_config.py index be4d42ef..bd6e05c7 100644 --- a/dulwich/tests/test_config.py +++ b/dulwich/tests/test_config.py @@ -1,308 +1,307 @@ # test_config.py -- Tests for reading and writing configuration files # Copyright (C) 2011 Jelmer Vernooij # # 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; either version 2 # or (at your option) a 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 reading and writing configuration files.""" from io import BytesIO from dulwich.config import ( ConfigDict, ConfigFile, StackedConfig, _check_section_name, _check_variable_name, _format_string, _escape_value, _parse_string, - _unescape_value, parse_submodules, ) from dulwich.tests import ( TestCase, ) class ConfigFileTests(TestCase): def from_file(self, text): return ConfigFile.from_file(BytesIO(text)) def test_empty(self): ConfigFile() def test_eq(self): self.assertEqual(ConfigFile(), ConfigFile()) def test_default_config(self): cf = self.from_file(b"""[core] repositoryformatversion = 0 filemode = true bare = false logallrefupdates = true """) self.assertEqual(ConfigFile({(b"core", ): { b"repositoryformatversion": b"0", b"filemode": b"true", b"bare": b"false", b"logallrefupdates": b"true"}}), cf) def test_from_file_empty(self): cf = self.from_file(b"") self.assertEqual(ConfigFile(), cf) def test_empty_line_before_section(self): cf = self.from_file(b"\n[section]\n") self.assertEqual(ConfigFile({(b"section", ): {}}), cf) def test_comment_before_section(self): cf = self.from_file(b"# foo\n[section]\n") self.assertEqual(ConfigFile({(b"section", ): {}}), cf) def test_comment_after_section(self): cf = self.from_file(b"[section] # foo\n") self.assertEqual(ConfigFile({(b"section", ): {}}), cf) def test_comment_after_variable(self): cf = self.from_file(b"[section]\nbar= foo # a comment\n") self.assertEqual(ConfigFile({(b"section", ): {b"bar": b"foo"}}), cf) def test_from_file_section(self): cf = self.from_file(b"[core]\nfoo = bar\n") self.assertEqual(b"bar", cf.get((b"core", ), b"foo")) self.assertEqual(b"bar", cf.get((b"core", b"foo"), b"foo")) def test_from_file_section_case_insensitive(self): cf = self.from_file(b"[cOre]\nfOo = bar\n") self.assertEqual(b"bar", cf.get((b"core", ), b"foo")) self.assertEqual(b"bar", cf.get((b"core", b"foo"), b"foo")) def test_from_file_with_mixed_quoted(self): cf = self.from_file(b"[core]\nfoo = \"bar\"la\n") self.assertEqual(b"barla", cf.get((b"core", ), b"foo")) def test_from_file_with_open_quoted(self): self.assertRaises(ValueError, self.from_file, b"[core]\nfoo = \"bar\n") def test_from_file_with_quotes(self): cf = self.from_file( b"[core]\n" b'foo = " bar"\n') self.assertEqual(b" bar", cf.get((b"core", ), b"foo")) def test_from_file_with_interrupted_line(self): cf = self.from_file( b"[core]\n" b'foo = bar\\\n' b' la\n') self.assertEqual(b"barla", cf.get((b"core", ), b"foo")) def test_from_file_with_boolean_setting(self): cf = self.from_file( b"[core]\n" b'foo\n') self.assertEqual(b"true", cf.get((b"core", ), b"foo")) def test_from_file_subsection(self): cf = self.from_file(b"[branch \"foo\"]\nfoo = bar\n") self.assertEqual(b"bar", cf.get((b"branch", b"foo"), b"foo")) def test_from_file_subsection_invalid(self): self.assertRaises(ValueError, self.from_file, b"[branch \"foo]\nfoo = bar\n") def test_from_file_subsection_not_quoted(self): cf = self.from_file(b"[branch.foo]\nfoo = bar\n") self.assertEqual(b"bar", cf.get((b"branch", b"foo"), b"foo")) def test_write_to_file_empty(self): c = ConfigFile() f = BytesIO() c.write_to_file(f) self.assertEqual(b"", f.getvalue()) def test_write_to_file_section(self): c = ConfigFile() c.set((b"core", ), b"foo", b"bar") f = BytesIO() c.write_to_file(f) self.assertEqual(b"[core]\n\tfoo = bar\n", f.getvalue()) def test_write_to_file_subsection(self): c = ConfigFile() c.set((b"branch", b"blie"), b"foo", b"bar") f = BytesIO() c.write_to_file(f) self.assertEqual(b"[branch \"blie\"]\n\tfoo = bar\n", f.getvalue()) def test_same_line(self): cf = self.from_file(b"[branch.foo] foo = bar\n") self.assertEqual(b"bar", cf.get((b"branch", b"foo"), b"foo")) #@expectedFailure def test_quoted(self): cf = self.from_file(b"""[gui] fontdiff = -family \\\"Ubuntu Mono\\\" -size 11 -weight normal -slant roman -underline 0 -overstrike 0 """) self.assertEqual(ConfigFile({(b'gui', ): { b'fontdiff': b'-family "Ubuntu Mono" -size 11 -weight normal -slant roman -underline 0 -overstrike 0', }}), cf) class ConfigDictTests(TestCase): def test_get_set(self): cd = ConfigDict() self.assertRaises(KeyError, cd.get, b"foo", b"core") cd.set((b"core", ), b"foo", b"bla") self.assertEqual(b"bla", cd.get((b"core", ), b"foo")) cd.set((b"core", ), b"foo", b"bloe") self.assertEqual(b"bloe", cd.get((b"core", ), b"foo")) def test_get_boolean(self): cd = ConfigDict() cd.set((b"core", ), b"foo", b"true") self.assertTrue(cd.get_boolean((b"core", ), b"foo")) cd.set((b"core", ), b"foo", b"false") self.assertFalse(cd.get_boolean((b"core", ), b"foo")) cd.set((b"core", ), b"foo", b"invalid") self.assertRaises(ValueError, cd.get_boolean, (b"core", ), b"foo") def test_dict(self): cd = ConfigDict() cd.set((b"core", ), b"foo", b"bla") cd.set((b"core2", ), b"foo", b"bloe") self.assertEqual([(b"core", ), (b"core2", )], list(cd.keys())) self.assertEqual(cd[(b"core", )], {b'foo': b'bla'}) cd[b'a'] = b'b' self.assertEqual(cd[b'a'], b'b') def test_iteritems(self): cd = ConfigDict() cd.set((b"core", ), b"foo", b"bla") cd.set((b"core2", ), b"foo", b"bloe") self.assertEqual( [(b'foo', b'bla')], list(cd.iteritems((b"core", )))) def test_iteritems_nonexistant(self): cd = ConfigDict() cd.set((b"core2", ), b"foo", b"bloe") self.assertEqual([], list(cd.iteritems((b"core", )))) def test_itersections(self): cd = ConfigDict() cd.set((b"core2", ), b"foo", b"bloe") self.assertEqual([(b"core2", )], list(cd.itersections())) class StackedConfigTests(TestCase): def test_default_backends(self): StackedConfig.default_backends() class EscapeValueTests(TestCase): def test_nothing(self): self.assertEqual(b"foo", _escape_value(b"foo")) def test_backslash(self): self.assertEqual(b"foo\\\\", _escape_value(b"foo\\")) def test_newline(self): self.assertEqual(b"foo\\n", _escape_value(b"foo\n")) class FormatStringTests(TestCase): def test_quoted(self): self.assertEqual(b'" foo"', _format_string(b" foo")) self.assertEqual(b'"\\tfoo"', _format_string(b"\tfoo")) def test_not_quoted(self): self.assertEqual(b'foo', _format_string(b"foo")) self.assertEqual(b'foo bar', _format_string(b"foo bar")) class ParseStringTests(TestCase): def test_quoted(self): self.assertEqual(b' foo', _parse_string(b'" foo"')) self.assertEqual(b'\tfoo', _parse_string(b'"\\tfoo"')) def test_not_quoted(self): self.assertEqual(b'foo', _parse_string(b"foo")) self.assertEqual(b'foo bar', _parse_string(b"foo bar")) def test_nothing(self): self.assertEqual(b"", _parse_string(b'')) def test_tab(self): self.assertEqual(b"\tbar\t", _parse_string(b"\\tbar\\t")) def test_newline(self): self.assertEqual(b"\nbar\t", _parse_string(b"\\nbar\\t\t")) def test_quote(self): self.assertEqual(b"\"foo\"", _parse_string(b"\\\"foo\\\"")) class CheckVariableNameTests(TestCase): def test_invalid(self): self.assertFalse(_check_variable_name(b"foo ")) self.assertFalse(_check_variable_name(b"bar,bar")) self.assertFalse(_check_variable_name(b"bar.bar")) def test_valid(self): self.assertTrue(_check_variable_name(b"FOO")) self.assertTrue(_check_variable_name(b"foo")) self.assertTrue(_check_variable_name(b"foo-bar")) class CheckSectionNameTests(TestCase): def test_invalid(self): self.assertFalse(_check_section_name(b"foo ")) self.assertFalse(_check_section_name(b"bar,bar")) def test_valid(self): self.assertTrue(_check_section_name(b"FOO")) self.assertTrue(_check_section_name(b"foo")) self.assertTrue(_check_section_name(b"foo-bar")) self.assertTrue(_check_section_name(b"bar.bar")) class SubmodulesTests(TestCase): def testSubmodules(self): cf = ConfigFile.from_file(BytesIO(b"""\ [submodule "core/lib"] path = core/lib url = https://github.com/phhusson/QuasselC.git """)) got = list(parse_submodules(cf)) self.assertEqual([ (b'core/lib', b'https://github.com/phhusson/QuasselC.git', b'core/lib')], got) diff --git a/dulwich/tests/test_hooks.py b/dulwich/tests/test_hooks.py index f10cfc34..bd885dc3 100644 --- a/dulwich/tests/test_hooks.py +++ b/dulwich/tests/test_hooks.py @@ -1,131 +1,130 @@ # test_hooks.py -- Tests for executing hooks # # 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; either version 2 # or (at your option) a 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 executing hooks.""" import os import stat import shutil -import sys import tempfile from dulwich import errors from dulwich.hooks import ( PreCommitShellHook, PostCommitShellHook, CommitMsgShellHook, ) from dulwich.tests import TestCase class ShellHookTests(TestCase): def setUp(self): super(ShellHookTests, self).setUp() if os.name != 'posix': self.skipTest('shell hook tests requires POSIX shell') def test_hook_pre_commit(self): pre_commit_fail = """#!/bin/sh exit 1 """ pre_commit_success = """#!/bin/sh exit 0 """ repo_dir = os.path.join(tempfile.mkdtemp()) os.mkdir(os.path.join(repo_dir, 'hooks')) self.addCleanup(shutil.rmtree, repo_dir) pre_commit = os.path.join(repo_dir, 'hooks', 'pre-commit') hook = PreCommitShellHook(repo_dir) with open(pre_commit, 'w') as f: f.write(pre_commit_fail) os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC) self.assertRaises(errors.HookError, hook.execute) with open(pre_commit, 'w') as f: f.write(pre_commit_success) os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC) hook.execute() def test_hook_commit_msg(self): commit_msg_fail = """#!/bin/sh exit 1 """ commit_msg_success = """#!/bin/sh exit 0 """ repo_dir = os.path.join(tempfile.mkdtemp()) os.mkdir(os.path.join(repo_dir, 'hooks')) self.addCleanup(shutil.rmtree, repo_dir) commit_msg = os.path.join(repo_dir, 'hooks', 'commit-msg') hook = CommitMsgShellHook(repo_dir) with open(commit_msg, 'w') as f: f.write(commit_msg_fail) os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC) self.assertRaises(errors.HookError, hook.execute, b'failed commit') with open(commit_msg, 'w') as f: f.write(commit_msg_success) os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC) hook.execute(b'empty commit') def test_hook_post_commit(self): (fd, path) = tempfile.mkstemp() os.close(fd) post_commit_msg = """#!/bin/sh rm """ + path + "\n" post_commit_msg_fail = """#!/bin/sh exit 1 """ repo_dir = os.path.join(tempfile.mkdtemp()) os.mkdir(os.path.join(repo_dir, 'hooks')) self.addCleanup(shutil.rmtree, repo_dir) post_commit = os.path.join(repo_dir, 'hooks', 'post-commit') hook = PostCommitShellHook(repo_dir) with open(post_commit, 'w') as f: f.write(post_commit_msg_fail) os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC) self.assertRaises(errors.HookError, hook.execute) with open(post_commit, 'w') as f: f.write(post_commit_msg) os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC) hook.execute() self.assertFalse(os.path.exists(path)) diff --git a/dulwich/tests/test_porcelain.py b/dulwich/tests/test_porcelain.py index 6390ed57..e6065abc 100644 --- a/dulwich/tests/test_porcelain.py +++ b/dulwich/tests/test_porcelain.py @@ -1,805 +1,804 @@ # test_porcelain.py -- porcelain tests # Copyright (C) 2013 Jelmer Vernooij # # 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) a later version. # # 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.porcelain.""" from contextlib import closing from io import BytesIO try: from StringIO import StringIO except ImportError: from io import StringIO import os import shutil import tarfile import tempfile import time from dulwich import porcelain from dulwich.diff_tree import tree_changes from dulwich.objects import ( Blob, Tag, Tree, ZERO_SHA, ) from dulwich.repo import Repo from dulwich.tests import ( TestCase, ) from dulwich.tests.utils import ( build_commit_graph, make_object, ) class PorcelainTestCase(TestCase): def setUp(self): super(PorcelainTestCase, self).setUp() repo_dir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, repo_dir) self.repo = Repo.init(repo_dir) def tearDown(self): super(PorcelainTestCase, self).tearDown() self.repo.close() class ArchiveTests(PorcelainTestCase): """Tests for the archive command.""" def test_simple(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"refs/heads/master"] = c3.id out = BytesIO() err = BytesIO() porcelain.archive(self.repo.path, b"refs/heads/master", outstream=out, errstream=err) self.assertEqual(b"", err.getvalue()) tf = tarfile.TarFile(fileobj=out) self.addCleanup(tf.close) self.assertEqual([], tf.getnames()) class UpdateServerInfoTests(PorcelainTestCase): def test_simple(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"refs/heads/foo"] = c3.id porcelain.update_server_info(self.repo.path) self.assertTrue(os.path.exists(os.path.join(self.repo.controldir(), 'info', 'refs'))) class CommitTests(PorcelainTestCase): def test_custom_author(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"refs/heads/foo"] = c3.id sha = porcelain.commit(self.repo.path, message=b"Some message", author=b"Joe ", committer=b"Bob ") self.assertTrue(isinstance(sha, bytes)) self.assertEqual(len(sha), 40) class CloneTests(PorcelainTestCase): def test_simple_local(self): f1_1 = make_object(Blob, data=b'f1') commit_spec = [[1], [2, 1], [3, 1, 2]] trees = {1: [(b'f1', f1_1), (b'f2', f1_1)], 2: [(b'f1', f1_1), (b'f2', f1_1)], 3: [(b'f1', f1_1), (b'f2', f1_1)], } c1, c2, c3 = build_commit_graph(self.repo.object_store, commit_spec, trees) self.repo.refs[b"refs/heads/master"] = c3.id target_path = tempfile.mkdtemp() errstream = BytesIO() self.addCleanup(shutil.rmtree, target_path) r = porcelain.clone(self.repo.path, target_path, checkout=False, errstream=errstream) self.assertEqual(r.path, target_path) self.assertEqual(Repo(target_path).head(), c3.id) self.assertTrue(b'f1' not in os.listdir(target_path)) self.assertTrue(b'f2' not in os.listdir(target_path)) def test_simple_local_with_checkout(self): f1_1 = make_object(Blob, data=b'f1') commit_spec = [[1], [2, 1], [3, 1, 2]] trees = {1: [(b'f1', f1_1), (b'f2', f1_1)], 2: [(b'f1', f1_1), (b'f2', f1_1)], 3: [(b'f1', f1_1), (b'f2', f1_1)], } c1, c2, c3 = build_commit_graph(self.repo.object_store, commit_spec, trees) self.repo.refs[b"refs/heads/master"] = c3.id target_path = tempfile.mkdtemp() errstream = BytesIO() self.addCleanup(shutil.rmtree, target_path) with closing(porcelain.clone(self.repo.path, target_path, checkout=True, errstream=errstream)) as r: self.assertEqual(r.path, target_path) with closing(Repo(target_path)) as r: self.assertEqual(r.head(), c3.id) self.assertTrue('f1' in os.listdir(target_path)) self.assertTrue('f2' in os.listdir(target_path)) def test_bare_local_with_checkout(self): f1_1 = make_object(Blob, data=b'f1') commit_spec = [[1], [2, 1], [3, 1, 2]] trees = {1: [(b'f1', f1_1), (b'f2', f1_1)], 2: [(b'f1', f1_1), (b'f2', f1_1)], 3: [(b'f1', f1_1), (b'f2', f1_1)], } c1, c2, c3 = build_commit_graph(self.repo.object_store, commit_spec, trees) self.repo.refs[b"refs/heads/master"] = c3.id target_path = tempfile.mkdtemp() errstream = BytesIO() self.addCleanup(shutil.rmtree, target_path) r = porcelain.clone(self.repo.path, target_path, bare=True, errstream=errstream) self.assertEqual(r.path, target_path) self.assertEqual(Repo(target_path).head(), c3.id) self.assertFalse(b'f1' in os.listdir(target_path)) self.assertFalse(b'f2' in os.listdir(target_path)) def test_no_checkout_with_bare(self): f1_1 = make_object(Blob, data=b'f1') commit_spec = [[1]] trees = {1: [(b'f1', f1_1), (b'f2', f1_1)]} (c1, ) = build_commit_graph(self.repo.object_store, commit_spec, trees) self.repo.refs[b"refs/heads/master"] = c1.id target_path = tempfile.mkdtemp() errstream = BytesIO() self.addCleanup(shutil.rmtree, target_path) self.assertRaises(ValueError, porcelain.clone, self.repo.path, target_path, checkout=True, bare=True, errstream=errstream) class InitTests(TestCase): def test_non_bare(self): repo_dir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, repo_dir) porcelain.init(repo_dir) def test_bare(self): repo_dir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, repo_dir) porcelain.init(repo_dir, bare=True) class AddTests(PorcelainTestCase): def test_add_default_paths(self): # create a file for initial commit with open(os.path.join(self.repo.path, 'blah'), 'w') as f: f.write("\n") porcelain.add(repo=self.repo.path, paths=['blah']) porcelain.commit(repo=self.repo.path, message=b'test', author=b'test', committer=b'test') # Add a second test file and a file in a directory with open(os.path.join(self.repo.path, 'foo'), 'w') as f: f.write("\n") os.mkdir(os.path.join(self.repo.path, 'adir')) with open(os.path.join(self.repo.path, 'adir', 'afile'), 'w') as f: f.write("\n") porcelain.add(self.repo.path) # Check that foo was added and nothing in .git was modified index = self.repo.open_index() self.assertEqual(sorted(index), [b'adir/afile', b'blah', b'foo']) def test_add_file(self): with open(os.path.join(self.repo.path, 'foo'), 'w') as f: f.write("BAR") porcelain.add(self.repo.path, paths=["foo"]) class RemoveTests(PorcelainTestCase): def test_remove_file(self): with open(os.path.join(self.repo.path, 'foo'), 'w') as f: f.write("BAR") porcelain.add(self.repo.path, paths=["foo"]) porcelain.rm(self.repo.path, paths=["foo"]) class LogTests(PorcelainTestCase): def test_simple(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id outstream = StringIO() porcelain.log(self.repo.path, outstream=outstream) self.assertEqual(3, outstream.getvalue().count("-" * 50)) def test_max_entries(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id outstream = StringIO() porcelain.log(self.repo.path, outstream=outstream, max_entries=1) self.assertEqual(1, outstream.getvalue().count("-" * 50)) class ShowTests(PorcelainTestCase): def test_nolist(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id outstream = StringIO() porcelain.show(self.repo.path, objects=c3.id, outstream=outstream) self.assertTrue(outstream.getvalue().startswith("-" * 50)) def test_simple(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id outstream = StringIO() porcelain.show(self.repo.path, objects=[c3.id], outstream=outstream) self.assertTrue(outstream.getvalue().startswith("-" * 50)) def test_blob(self): b = Blob.from_string(b"The Foo\n") self.repo.object_store.add_object(b) outstream = StringIO() porcelain.show(self.repo.path, objects=[b.id], outstream=outstream) self.assertEqual(outstream.getvalue(), "The Foo\n") class SymbolicRefTests(PorcelainTestCase): def test_set_wrong_symbolic_ref(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id self.assertRaises(ValueError, porcelain.symbolic_ref, self.repo.path, b'foobar') def test_set_force_wrong_symbolic_ref(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id porcelain.symbolic_ref(self.repo.path, b'force_foobar', force=True) #test if we actually changed the file with self.repo.get_named_file('HEAD') as f: new_ref = f.read() self.assertEqual(new_ref, b'ref: refs/heads/force_foobar\n') def test_set_symbolic_ref(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id porcelain.symbolic_ref(self.repo.path, b'master') def test_set_symbolic_ref_other_than_master(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]], attrs=dict(refs='develop')) self.repo.refs[b"HEAD"] = c3.id self.repo.refs[b"refs/heads/develop"] = c3.id porcelain.symbolic_ref(self.repo.path, b'develop') #test if we actually changed the file with self.repo.get_named_file('HEAD') as f: new_ref = f.read() self.assertEqual(new_ref, b'ref: refs/heads/develop\n') class DiffTreeTests(PorcelainTestCase): def test_empty(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id outstream = BytesIO() porcelain.diff_tree(self.repo.path, c2.tree, c3.tree, outstream=outstream) self.assertEqual(outstream.getvalue(), b"") class CommitTreeTests(PorcelainTestCase): def test_simple(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) b = Blob() b.data = b"foo the bar" t = Tree() t.add(b"somename", 0o100644, b.id) self.repo.object_store.add_object(t) self.repo.object_store.add_object(b) sha = porcelain.commit_tree( self.repo.path, t.id, message=b"Withcommit.", author=b"Joe ", committer=b"Jane ") self.assertTrue(isinstance(sha, bytes)) self.assertEqual(len(sha), 40) class RevListTests(PorcelainTestCase): def test_simple(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) outstream = BytesIO() porcelain.rev_list( self.repo.path, [c3.id], outstream=outstream) self.assertEqual( c3.id + b"\n" + c2.id + b"\n" + c1.id + b"\n", outstream.getvalue()) class TagCreateTests(PorcelainTestCase): def test_annotated(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id porcelain.tag_create(self.repo.path, b"tryme", b'foo ', b'bar', annotated=True) tags = self.repo.refs.as_dict(b"refs/tags") self.assertEqual(list(tags.keys()), [b"tryme"]) tag = self.repo[b'refs/tags/tryme'] self.assertTrue(isinstance(tag, Tag)) self.assertEqual(b"foo ", tag.tagger) self.assertEqual(b"bar", tag.message) self.assertLess(time.time() - tag.tag_time, 5) def test_unannotated(self): c1, c2, c3 = build_commit_graph(self.repo.object_store, [[1], [2, 1], [3, 1, 2]]) self.repo.refs[b"HEAD"] = c3.id porcelain.tag_create(self.repo.path, b"tryme", annotated=False) tags = self.repo.refs.as_dict(b"refs/tags") self.assertEqual(list(tags.keys()), [b"tryme"]) self.repo[b'refs/tags/tryme'] self.assertEqual(list(tags.values()), [self.repo.head()]) class TagListTests(PorcelainTestCase): def test_empty(self): tags = porcelain.tag_list(self.repo.path) self.assertEqual([], tags) def test_simple(self): self.repo.refs[b"refs/tags/foo"] = b"aa" * 20 self.repo.refs[b"refs/tags/bar/bla"] = b"bb" * 20 tags = porcelain.tag_list(self.repo.path) self.assertEqual([b"bar/bla", b"foo"], tags) class TagDeleteTests(PorcelainTestCase): def test_simple(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo[b"HEAD"] = c1.id porcelain.tag_create(self.repo, b'foo') self.assertTrue(b"foo" in porcelain.tag_list(self.repo)) porcelain.tag_delete(self.repo, b'foo') self.assertFalse(b"foo" in porcelain.tag_list(self.repo)) class ResetTests(PorcelainTestCase): def test_hard_head(self): with open(os.path.join(self.repo.path, 'foo'), 'w') as f: f.write("BAR") porcelain.add(self.repo.path, paths=["foo"]) porcelain.commit(self.repo.path, message=b"Some message", committer=b"Jane ", author=b"John ") with open(os.path.join(self.repo.path, 'foo'), 'wb') as f: f.write(b"OOH") porcelain.reset(self.repo, "hard", b"HEAD") index = self.repo.open_index() changes = list(tree_changes(self.repo, index.commit(self.repo.object_store), self.repo[b'HEAD'].tree)) self.assertEqual([], changes) class PushTests(PorcelainTestCase): def test_simple(self): """ Basic test of porcelain push where self.repo is the remote. First clone the remote, commit a file to the clone, then push the changes back to the remote. """ outstream = BytesIO() errstream = BytesIO() porcelain.commit(repo=self.repo.path, message=b'init', author=b'', committer=b'') # Setup target repo cloned from temp test repo clone_path = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, clone_path) target_repo = porcelain.clone(self.repo.path, target=clone_path, errstream=errstream) try: self.assertEqual(target_repo[b'HEAD'], self.repo[b'HEAD']) finally: target_repo.close() # create a second file to be pushed back to origin handle, fullpath = tempfile.mkstemp(dir=clone_path) os.close(handle) porcelain.add(repo=clone_path, paths=[os.path.basename(fullpath)]) porcelain.commit(repo=clone_path, message=b'push', author=b'', committer=b'') # Setup a non-checked out branch in the remote refs_path = b"refs/heads/foo" new_id = self.repo[b'HEAD'].id self.assertNotEqual(new_id, ZERO_SHA) self.repo.refs[refs_path] = new_id # Push to the remote porcelain.push(clone_path, self.repo.path, b"HEAD:" + refs_path, outstream=outstream, errstream=errstream) # Check that the target and source with closing(Repo(clone_path)) as r_clone: self.assertEqual({ b'HEAD': new_id, b'refs/heads/foo': r_clone[b'HEAD'].id, b'refs/heads/master': new_id, }, self.repo.get_refs()) self.assertEqual(r_clone[b'HEAD'].id, self.repo.refs[refs_path]) # Get the change in the target repo corresponding to the add # this will be in the foo branch. change = list(tree_changes(self.repo, self.repo[b'HEAD'].tree, self.repo[b'refs/heads/foo'].tree))[0] self.assertEqual(os.path.basename(fullpath), change.new.path.decode('ascii')) def test_delete(self): """Basic test of porcelain push, removing a branch. """ outstream = BytesIO() errstream = BytesIO() porcelain.commit(repo=self.repo.path, message=b'init', author=b'', committer=b'') # Setup target repo cloned from temp test repo clone_path = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, clone_path) target_repo = porcelain.clone(self.repo.path, target=clone_path, errstream=errstream) target_repo.close() # Setup a non-checked out branch in the remote refs_path = b"refs/heads/foo" new_id = self.repo[b'HEAD'].id self.assertNotEqual(new_id, ZERO_SHA) self.repo.refs[refs_path] = new_id # Push to the remote porcelain.push(clone_path, self.repo.path, b":" + refs_path, outstream=outstream, errstream=errstream) - with closing(Repo(clone_path)) as r_clone: - self.assertEqual({ - b'HEAD': new_id, - b'refs/heads/master': new_id, - }, self.repo.get_refs()) + self.assertEqual({ + b'HEAD': new_id, + b'refs/heads/master': new_id, + }, self.repo.get_refs()) class PullTests(PorcelainTestCase): def test_simple(self): outstream = BytesIO() errstream = BytesIO() # create a file for initial commit handle, fullpath = tempfile.mkstemp(dir=self.repo.path) os.close(handle) filename = os.path.basename(fullpath) porcelain.add(repo=self.repo.path, paths=filename) porcelain.commit(repo=self.repo.path, message=b'test', author=b'test', committer=b'test') # Setup target repo target_path = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, target_path) target_repo = porcelain.clone(self.repo.path, target=target_path, errstream=errstream) target_repo.close() # create a second file to be pushed handle, fullpath = tempfile.mkstemp(dir=self.repo.path) os.close(handle) filename = os.path.basename(fullpath) porcelain.add(repo=self.repo.path, paths=filename) porcelain.commit(repo=self.repo.path, message=b'test2', author=b'test2', committer=b'test2') self.assertTrue(b'refs/heads/master' in self.repo.refs) self.assertTrue(b'refs/heads/master' in target_repo.refs) # Pull changes into the cloned repo porcelain.pull(target_path, self.repo.path, b'refs/heads/master', outstream=outstream, errstream=errstream) # Check the target repo for pushed changes with closing(Repo(target_path)) as r: self.assertEqual(r[b'HEAD'].id, self.repo[b'HEAD'].id) class StatusTests(PorcelainTestCase): def test_empty(self): results = porcelain.status(self.repo) self.assertEqual( {'add': [], 'delete': [], 'modify': []}, results.staged) self.assertEqual([], results.unstaged) def test_status(self): """Integration test for `status` functionality.""" # Commit a dummy file then modify it fullpath = os.path.join(self.repo.path, 'foo') with open(fullpath, 'w') as f: f.write('origstuff') porcelain.add(repo=self.repo.path, paths=['foo']) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'', committer=b'') # modify access and modify time of path os.utime(fullpath, (0, 0)) with open(fullpath, 'wb') as f: f.write(b'stuff') # Make a dummy file and stage it filename_add = 'bar' fullpath = os.path.join(self.repo.path, filename_add) with open(fullpath, 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=filename_add) results = porcelain.status(self.repo) self.assertEqual(results.staged['add'][0], filename_add.encode('ascii')) self.assertEqual(results.unstaged, [b'foo']) def test_get_tree_changes_add(self): """Unit test for get_tree_changes add.""" # Make a dummy file, stage filename = 'bar' with open(os.path.join(self.repo.path, filename), 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=filename) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'', committer=b'') filename = 'foo' with open(os.path.join(self.repo.path, filename), 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=filename) changes = porcelain.get_tree_changes(self.repo.path) self.assertEqual(changes['add'][0], filename.encode('ascii')) self.assertEqual(len(changes['add']), 1) self.assertEqual(len(changes['modify']), 0) self.assertEqual(len(changes['delete']), 0) def test_get_tree_changes_modify(self): """Unit test for get_tree_changes modify.""" # Make a dummy file, stage, commit, modify filename = 'foo' fullpath = os.path.join(self.repo.path, filename) with open(fullpath, 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=filename) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'', committer=b'') with open(fullpath, 'w') as f: f.write('otherstuff') porcelain.add(repo=self.repo.path, paths=filename) changes = porcelain.get_tree_changes(self.repo.path) self.assertEqual(changes['modify'][0], filename.encode('ascii')) self.assertEqual(len(changes['add']), 0) self.assertEqual(len(changes['modify']), 1) self.assertEqual(len(changes['delete']), 0) def test_get_tree_changes_delete(self): """Unit test for get_tree_changes delete.""" # Make a dummy file, stage, commit, remove filename = 'foo' with open(os.path.join(self.repo.path, filename), 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=filename) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'', committer=b'') porcelain.rm(repo=self.repo.path, paths=[filename]) changes = porcelain.get_tree_changes(self.repo.path) self.assertEqual(changes['delete'][0], filename.encode('ascii')) self.assertEqual(len(changes['add']), 0) self.assertEqual(len(changes['modify']), 0) self.assertEqual(len(changes['delete']), 1) # TODO(jelmer): Add test for dulwich.porcelain.daemon class UploadPackTests(PorcelainTestCase): """Tests for upload_pack.""" def test_upload_pack(self): outf = BytesIO() exitcode = porcelain.upload_pack(self.repo.path, BytesIO(b"0000"), outf) outlines = outf.getvalue().splitlines() self.assertEqual([b"0000"], outlines) self.assertEqual(0, exitcode) class ReceivePackTests(PorcelainTestCase): """Tests for receive_pack.""" def test_receive_pack(self): filename = 'foo' with open(os.path.join(self.repo.path, filename), 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=filename) self.repo.do_commit(message=b'test status', author=b'', committer=b'', author_timestamp=1402354300, commit_timestamp=1402354300, author_timezone=0, commit_timezone=0) outf = BytesIO() exitcode = porcelain.receive_pack(self.repo.path, BytesIO(b"0000"), outf) outlines = outf.getvalue().splitlines() self.assertEqual([ b'00739e65bdcf4a22cdd4f3700604a275cd2aaf146b23 HEAD\x00 report-status ' b'delete-refs quiet ofs-delta side-band-64k no-done', b'003f9e65bdcf4a22cdd4f3700604a275cd2aaf146b23 refs/heads/master', b'0000'], outlines) self.assertEqual(0, exitcode) class BranchListTests(PorcelainTestCase): def test_standard(self): self.assertEqual(set([]), set(porcelain.branch_list(self.repo))) def test_new_branch(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo[b"HEAD"] = c1.id porcelain.branch_create(self.repo, b"foo") self.assertEqual( set([b"master", b"foo"]), set(porcelain.branch_list(self.repo))) class BranchCreateTests(PorcelainTestCase): def test_branch_exists(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo[b"HEAD"] = c1.id porcelain.branch_create(self.repo, b"foo") self.assertRaises(KeyError, porcelain.branch_create, self.repo, b"foo") porcelain.branch_create(self.repo, b"foo", force=True) def test_new_branch(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo[b"HEAD"] = c1.id porcelain.branch_create(self.repo, b"foo") self.assertEqual( set([b"master", b"foo"]), set(porcelain.branch_list(self.repo))) class BranchDeleteTests(PorcelainTestCase): def test_simple(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo[b"HEAD"] = c1.id porcelain.branch_create(self.repo, b'foo') self.assertTrue(b"foo" in porcelain.branch_list(self.repo)) porcelain.branch_delete(self.repo, b'foo') self.assertFalse(b"foo" in porcelain.branch_list(self.repo)) class FetchTests(PorcelainTestCase): def test_simple(self): outstream = BytesIO() errstream = BytesIO() # create a file for initial commit handle, fullpath = tempfile.mkstemp(dir=self.repo.path) os.close(handle) filename = os.path.basename(fullpath) porcelain.add(repo=self.repo.path, paths=filename) porcelain.commit(repo=self.repo.path, message=b'test', author=b'test', committer=b'test') # Setup target repo target_path = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, target_path) target_repo = porcelain.clone(self.repo.path, target=target_path, errstream=errstream) # create a second file to be pushed handle, fullpath = tempfile.mkstemp(dir=self.repo.path) os.close(handle) filename = os.path.basename(fullpath) porcelain.add(repo=self.repo.path, paths=filename) porcelain.commit(repo=self.repo.path, message=b'test2', author=b'test2', committer=b'test2') self.assertFalse(self.repo[b'HEAD'].id in target_repo) target_repo.close() # Fetch changes into the cloned repo porcelain.fetch(target_path, self.repo.path, outstream=outstream, errstream=errstream) # Check the target repo for pushed changes with closing(Repo(target_path)) as r: self.assertTrue(self.repo[b'HEAD'].id in r) class RepackTests(PorcelainTestCase): def test_empty(self): porcelain.repack(self.repo) def test_simple(self): handle, fullpath = tempfile.mkstemp(dir=self.repo.path) os.close(handle) filename = os.path.basename(fullpath) porcelain.add(repo=self.repo.path, paths=filename) porcelain.repack(self.repo) diff --git a/dulwich/tests/utils.py b/dulwich/tests/utils.py index 4d8cf1f2..17b74b99 100644 --- a/dulwich/tests/utils.py +++ b/dulwich/tests/utils.py @@ -1,362 +1,361 @@ # utils.py -- Test utilities for Dulwich. # 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. """Utility functions common to Dulwich tests.""" import datetime import os import shutil -import sys import tempfile import time import types import warnings from dulwich.index import ( commit_tree, ) from dulwich.objects import ( FixedSha, Commit, Tag, object_class, ) from dulwich.pack import ( OFS_DELTA, REF_DELTA, DELTA_TYPES, obj_sha, SHA1Writer, write_pack_header, write_pack_object, create_delta, ) from dulwich.repo import Repo from dulwich.tests import ( SkipTest, skipIf, ) # Plain files are very frequently used in tests, so let the mode be very short. F = 0o100644 # Shorthand mode for Files. def open_repo(name, temp_dir=None): """Open a copy of a repo in a temporary directory. Use this function for accessing repos in dulwich/tests/data/repos to avoid accidentally or intentionally modifying those repos in place. Use tear_down_repo to delete any temp files created. :param name: The name of the repository, relative to dulwich/tests/data/repos :param temp_dir: temporary directory to initialize to. If not provided, a temporary directory will be created. :returns: An initialized Repo object that lives in a temporary directory. """ if temp_dir is None: temp_dir = tempfile.mkdtemp() repo_dir = os.path.join(os.path.dirname(__file__), 'data', 'repos', name) temp_repo_dir = os.path.join(temp_dir, name) shutil.copytree(repo_dir, temp_repo_dir, symlinks=True) return Repo(temp_repo_dir) def tear_down_repo(repo): """Tear down a test repository.""" repo.close() temp_dir = os.path.dirname(repo.path.rstrip(os.sep)) shutil.rmtree(temp_dir) def make_object(cls, **attrs): """Make an object for testing and assign some members. This method creates a new subclass to allow arbitrary attribute reassignment, which is not otherwise possible with objects having __slots__. :param attrs: dict of attributes to set on the new object. :return: A newly initialized object of type cls. """ class TestObject(cls): """Class that inherits from the given class, but without __slots__. Note that classes with __slots__ can't have arbitrary attributes monkey- patched in, so this is a class that is exactly the same only with a __dict__ instead of __slots__. """ pass TestObject.__name__ = 'TestObject_' + cls.__name__ obj = TestObject() for name, value in attrs.items(): if name == 'id': # id property is read-only, so we overwrite sha instead. sha = FixedSha(value) obj.sha = lambda: sha else: setattr(obj, name, value) return obj def make_commit(**attrs): """Make a Commit object with a default set of members. :param attrs: dict of attributes to overwrite from the default values. :return: A newly initialized Commit object. """ default_time = int(time.mktime(datetime.datetime(2010, 1, 1).timetuple())) all_attrs = {'author': b'Test Author ', 'author_time': default_time, 'author_timezone': 0, 'committer': b'Test Committer ', 'commit_time': default_time, 'commit_timezone': 0, 'message': b'Test message.', 'parents': [], 'tree': b'0' * 40} all_attrs.update(attrs) return make_object(Commit, **all_attrs) def make_tag(target, **attrs): """Make a Tag object with a default set of values. :param target: object to be tagged (Commit, Blob, Tree, etc) :param attrs: dict of attributes to overwrite from the default values. :return: A newly initialized Tag object. """ target_id = target.id target_type = object_class(target.type_name) default_time = int(time.mktime(datetime.datetime(2010, 1, 1).timetuple())) all_attrs = {'tagger': b'Test Author ', 'tag_time': default_time, 'tag_timezone': 0, 'message': b'Test message.', 'object': (target_type, target_id), 'name': b'Test Tag', } all_attrs.update(attrs) return make_object(Tag, **all_attrs) def functest_builder(method, func): """Generate a test method that tests the given function.""" def do_test(self): method(self, func) return do_test def ext_functest_builder(method, func): """Generate a test method that tests the given extension function. This is intended to generate test methods that test both a pure-Python version and an extension version using common test code. The extension test will raise SkipTest if the extension is not found. Sample usage: class MyTest(TestCase); def _do_some_test(self, func_impl): self.assertEqual('foo', func_impl()) test_foo = functest_builder(_do_some_test, foo_py) test_foo_extension = ext_functest_builder(_do_some_test, _foo_c) :param method: The method to run. It must must two parameters, self and the function implementation to test. :param func: The function implementation to pass to method. """ def do_test(self): if not isinstance(func, types.BuiltinFunctionType): raise SkipTest("%s extension not found" % func) method(self, func) return do_test def build_pack(f, objects_spec, store=None): """Write test pack data from a concise spec. :param f: A file-like object to write the pack to. :param objects_spec: A list of (type_num, obj). For non-delta types, obj is the string of that object's data. For delta types, obj is a tuple of (base, data), where: * base can be either an index in objects_spec of the base for that * delta; or for a ref delta, a SHA, in which case the resulting pack * will be thin and the base will be an external ref. * data is a string of the full, non-deltified data for that object. Note that offsets/refs and deltas are computed within this function. :param store: An optional ObjectStore for looking up external refs. :return: A list of tuples in the order specified by objects_spec: (offset, type num, data, sha, CRC32) """ sf = SHA1Writer(f) num_objects = len(objects_spec) write_pack_header(sf, num_objects) full_objects = {} offsets = {} crc32s = {} while len(full_objects) < num_objects: for i, (type_num, data) in enumerate(objects_spec): if type_num not in DELTA_TYPES: full_objects[i] = (type_num, data, obj_sha(type_num, [data])) continue base, data = data if isinstance(base, int): if base not in full_objects: continue base_type_num, _, _ = full_objects[base] else: base_type_num, _ = store.get_raw(base) full_objects[i] = (base_type_num, data, obj_sha(base_type_num, [data])) for i, (type_num, obj) in enumerate(objects_spec): offset = f.tell() if type_num == OFS_DELTA: base_index, data = obj base = offset - offsets[base_index] _, base_data, _ = full_objects[base_index] obj = (base, create_delta(base_data, data)) elif type_num == REF_DELTA: base_ref, data = obj if isinstance(base_ref, int): _, base_data, base = full_objects[base_ref] else: base_type_num, base_data = store.get_raw(base_ref) base = obj_sha(base_type_num, base_data) obj = (base, create_delta(base_data, data)) crc32 = write_pack_object(sf, type_num, obj) offsets[i] = offset crc32s[i] = crc32 expected = [] for i in range(num_objects): type_num, data, sha = full_objects[i] assert len(sha) == 20 expected.append((offsets[i], type_num, data, sha, crc32s[i])) sf.write_sha() f.seek(0) return expected def build_commit_graph(object_store, commit_spec, trees=None, attrs=None): """Build a commit graph from a concise specification. Sample usage: >>> c1, c2, c3 = build_commit_graph(store, [[1], [2, 1], [3, 1, 2]]) >>> store[store[c3].parents[0]] == c1 True >>> store[store[c3].parents[1]] == c2 True If not otherwise specified, commits will refer to the empty tree and have commit times increasing in the same order as the commit spec. :param object_store: An ObjectStore to commit objects to. :param commit_spec: An iterable of iterables of ints defining the commit graph. Each entry defines one commit, and entries must be in topological order. The first element of each entry is a commit number, and the remaining elements are its parents. The commit numbers are only meaningful for the call to make_commits; since real commit objects are created, they will get created with real, opaque SHAs. :param trees: An optional dict of commit number -> tree spec for building trees for commits. The tree spec is an iterable of (path, blob, mode) or (path, blob) entries; if mode is omitted, it defaults to the normal file mode (0100644). :param attrs: A dict of commit number -> (dict of attribute -> value) for assigning additional values to the commits. :return: The list of commit objects created. :raise ValueError: If an undefined commit identifier is listed as a parent. """ if trees is None: trees = {} if attrs is None: attrs = {} commit_time = 0 nums = {} commits = [] for commit in commit_spec: commit_num = commit[0] try: parent_ids = [nums[pn] for pn in commit[1:]] except KeyError as e: missing_parent, = e.args raise ValueError('Unknown parent %i' % missing_parent) blobs = [] for entry in trees.get(commit_num, []): if len(entry) == 2: path, blob = entry entry = (path, blob, F) path, blob, mode = entry blobs.append((path, blob.id, mode)) object_store.add_object(blob) tree_id = commit_tree(object_store, blobs) commit_attrs = { 'message': ('Commit %i' % commit_num).encode('ascii'), 'parents': parent_ids, 'tree': tree_id, 'commit_time': commit_time, } commit_attrs.update(attrs.get(commit_num, {})) commit_obj = make_commit(**commit_attrs) # By default, increment the time by a lot. Out-of-order commits should # be closer together than this because their main cause is clock skew. commit_time = commit_attrs['commit_time'] + 100 nums[commit_num] = commit_obj.id object_store.add_object(commit_obj) commits.append(commit_obj) return commits def setup_warning_catcher(): """Wrap warnings.showwarning with code that records warnings.""" caught_warnings = [] original_showwarning = warnings.showwarning def custom_showwarning(*args, **kwargs): caught_warnings.append(args[0]) warnings.showwarning = custom_showwarning def restore_showwarning(): warnings.showwarning = original_showwarning return caught_warnings, restore_showwarning