diff --git a/dulwich/refs.py b/dulwich/refs.py index 42239642..67ff8d86 100644 --- a/dulwich/refs.py +++ b/dulwich/refs.py @@ -1,782 +1,788 @@ # refs.py -- For dealing with git refs # Copyright (C) 2008-2013 Jelmer Vernooij # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Ref handling. """ import errno import os import sys from dulwich.errors import ( PackedRefsException, RefFormatError, ) from dulwich.objects import ( git_line, valid_hexsha, ZERO_SHA, ) from dulwich.file import ( GitFile, ensure_dir_exists, ) SYMREF = b'ref: ' LOCAL_BRANCH_PREFIX = b'refs/heads/' BAD_REF_CHARS = set(b'\177 ~^:?*[') def check_ref_format(refname): """Check if a refname is correctly formatted. Implements all the same rules as git-check-ref-format[1]. [1] http://www.kernel.org/pub/software/scm/git/docs/git-check-ref-format.html :param refname: The refname to check :return: True if refname is valid, False otherwise """ # These could be combined into one big expression, but are listed separately # to parallel [1]. if b'/.' in refname or refname.startswith(b'.'): return False if b'/' not in refname: return False if b'..' in refname: return False for i, c in enumerate(refname): if ord(refname[i:i+1]) < 0o40 or c in BAD_REF_CHARS: return False if refname[-1] in b'/.': return False if refname.endswith(b'.lock'): return False if b'@{' in refname: return False if b'\\' in refname: return False return True class RefsContainer(object): """A container for refs.""" def set_symbolic_ref(self, name, other): """Make a ref point at another ref. :param name: Name of the ref to set :param other: Name of the ref to point at """ raise NotImplementedError(self.set_symbolic_ref) def get_packed_refs(self): """Get contents of the packed-refs file. :return: Dictionary mapping ref names to SHA1s :note: Will return an empty dictionary when no packed-refs file is present. """ raise NotImplementedError(self.get_packed_refs) 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 the ref may point to a tag, but no cached information is available, None is returned. """ return None def import_refs(self, base, other): for name, value in other.items(): self[b'/'.join((base, name))] = value def allkeys(self): """All refs present in this container.""" raise NotImplementedError(self.allkeys) def keys(self, base=None): """Refs present in this container. :param base: An optional base to return refs under. :return: An unsorted set of valid refs in this container, including packed refs. """ if base is not None: return self.subkeys(base) else: return self.allkeys() def subkeys(self, base): """Refs present in this container under a base. :param base: The base to return refs under. :return: A set of valid refs in this container under the base; the base prefix is stripped from the ref names returned. """ keys = set() base_len = len(base) + 1 for refname in self.allkeys(): if refname.startswith(base): keys.add(refname[base_len:]) return keys def as_dict(self, base=None): """Return the contents of this container as a dictionary. """ ret = {} keys = self.keys(base) if base is None: base = b'' else: base = base.rstrip(b'/') for key in keys: try: ret[key] = self[(base + b'/' + key).strip(b'/')] except KeyError: continue # Unable to resolve return ret def _check_refname(self, name): """Ensure a refname is valid and lives in refs or is HEAD. HEAD is not a valid refname according to git-check-ref-format, but this class needs to be able to touch HEAD. Also, check_ref_format expects refnames without the leading 'refs/', but this class requires that so it cannot touch anything outside the refs dir (or HEAD). :param name: The name of the reference. :raises KeyError: if a refname is not HEAD or is otherwise not valid. """ if name in (b'HEAD', b'refs/stash'): return if not name.startswith(b'refs/') or not check_ref_format(name[5:]): raise RefFormatError(name) def read_ref(self, refname): """Read a reference without following any references. :param refname: The name of the reference :return: The contents of the ref file, or None if it does not exist. """ contents = self.read_loose_ref(refname) if not contents: contents = self.get_packed_refs().get(refname, None) return contents def read_loose_ref(self, name): """Read a loose reference and return its contents. :param name: the refname to read :return: The contents of the ref file, or None if it does not exist. """ raise NotImplementedError(self.read_loose_ref) def follow(self, name): """Follow a reference name. :return: a tuple of (refnames, sha), wheres refnames are the names of references in the chain """ contents = SYMREF + name depth = 0 refnames = [] while contents.startswith(SYMREF): refname = contents[len(SYMREF):] refnames.append(refname) contents = self.read_ref(refname) if not contents: break depth += 1 if depth > 5: raise KeyError(name) return refnames, contents def _follow(self, name): import warnings warnings.warn( "RefsContainer._follow is deprecated. Use RefsContainer.follow instead.", DeprecationWarning) refnames, contents = self.follow(name) if not refnames: return (None, contents) return (refnames[-1], contents) def __contains__(self, refname): if self.read_ref(refname): return True return False def __getitem__(self, name): """Get the SHA1 for a reference name. This method follows all symbolic references. """ _, sha = self.follow(name) if sha is None: raise KeyError(name) return sha def set_if_equals(self, name, old_ref, new_ref): """Set a refname to new_ref only if it currently equals old_ref. This method follows all symbolic references if applicable for the subclass, and can be used to perform an atomic compare-and-swap operation. :param name: The refname to set. :param old_ref: The old sha the refname must refer to, or None to set unconditionally. :param new_ref: The new sha the refname will refer to. :return: True if the set was successful, False otherwise. """ raise NotImplementedError(self.set_if_equals) def add_if_new(self, name, ref): """Add a new reference only if it does not already exist.""" raise NotImplementedError(self.add_if_new) def __setitem__(self, name, ref): """Set a reference name to point to the given SHA1. This method follows all symbolic references if applicable for the subclass. :note: This method unconditionally overwrites the contents of a reference. To update atomically only if the reference has not changed, use set_if_equals(). :param name: The refname to set. :param ref: The new sha the refname will refer to. """ self.set_if_equals(name, None, ref) def remove_if_equals(self, name, old_ref): """Remove a refname only if it currently equals old_ref. This method does not follow symbolic references, even if applicable for the subclass. It can be used to perform an atomic compare-and-delete operation. :param name: The refname to delete. :param old_ref: The old sha the refname must refer to, or None to delete unconditionally. :return: True if the delete was successful, False otherwise. """ raise NotImplementedError(self.remove_if_equals) def __delitem__(self, name): """Remove a refname. This method does not follow symbolic references, even if applicable for the subclass. :note: This method unconditionally deletes the contents of a reference. To delete atomically only if the reference has not changed, use remove_if_equals(). :param name: The refname to delete. """ self.remove_if_equals(name, None) class DictRefsContainer(RefsContainer): """RefsContainer backed by a simple dict. This container does not support symbolic or packed references and is not threadsafe. """ def __init__(self, refs): self._refs = refs self._peeled = {} def allkeys(self): return self._refs.keys() def read_loose_ref(self, name): return self._refs.get(name, None) def get_packed_refs(self): return {} def set_symbolic_ref(self, name, other): self._refs[name] = SYMREF + other def set_if_equals(self, name, old_ref, new_ref): if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref: return False realnames, _ = self.follow(name) for realname in realnames: self._check_refname(realname) self._refs[realname] = new_ref return True def add_if_new(self, name, ref): if name in self._refs: return False self._refs[name] = ref return True def remove_if_equals(self, name, old_ref): if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref: return False try: del self._refs[name] except KeyError: pass return True def get_peeled(self, name): return self._peeled.get(name) def _update(self, refs): """Update multiple refs; intended only for testing.""" # TODO(dborowitz): replace this with a public function that uses # set_if_equal. self._refs.update(refs) def _update_peeled(self, peeled): """Update cached peeled refs; intended only for testing.""" self._peeled.update(peeled) class InfoRefsContainer(RefsContainer): """Refs container that reads refs from a info/refs file.""" def __init__(self, f): self._refs = {} self._peeled = {} for l in f.readlines(): sha, name = l.rstrip(b'\n').split(b'\t') if name.endswith(b'^{}'): name = name[:-3] if not check_ref_format(name): raise ValueError("invalid ref name %r" % name) self._peeled[name] = sha else: if not check_ref_format(name): raise ValueError("invalid ref name %r" % name) self._refs[name] = sha def allkeys(self): return self._refs.keys() def read_loose_ref(self, name): return self._refs.get(name, None) def get_packed_refs(self): return {} def get_peeled(self, name): try: return self._peeled[name] except KeyError: return self._refs[name] class DiskRefsContainer(RefsContainer): """Refs container that reads refs from disk.""" - def __init__(self, path): + def __init__(self, path, **kwargs): self.path = path + self.worktree_path = kwargs.get('worktree', path) self._packed_refs = None self._peeled_refs = None def __repr__(self): return "%s(%r)" % (self.__class__.__name__, self.path) def subkeys(self, base): subkeys = set() path = self.refpath(base) for root, dirs, files in os.walk(path): dir = root[len(path):].strip(os.path.sep).replace(os.path.sep, "/") for filename in files: refname = (("%s/%s" % (dir, filename)) .strip("/").encode(sys.getfilesystemencoding())) # check_ref_format requires at least one /, so we prepend the # base before calling it. if check_ref_format(base + b'/' + refname): subkeys.add(refname) for key in self.get_packed_refs(): if key.startswith(base): subkeys.add(key[len(base):].strip(b'/')) return subkeys def allkeys(self): allkeys = set() if os.path.exists(self.refpath(b'HEAD')): allkeys.add(b'HEAD') path = self.refpath(b'') for root, dirs, files in os.walk(self.refpath(b'refs')): dir = root[len(path):].strip(os.path.sep).replace(os.path.sep, "/") for filename in files: refname = ("%s/%s" % (dir, filename)).encode(sys.getfilesystemencoding()) if check_ref_format(refname): allkeys.add(refname) allkeys.update(self.get_packed_refs()) return allkeys def refpath(self, name): """Return the disk path of a ref. """ if getattr(self.path, "encode", None) and getattr(name, "decode", None): name = name.decode(sys.getfilesystemencoding()) if os.path.sep != "/": name = name.replace("/", os.path.sep) - return os.path.join(self.path, name) + #TODO: as the 'HEAD' reference is working tree specific, it + # should actually not be a part of RefsContainer + if(name == b'HEAD'): + return os.path.join(self.worktree_path, name) + else: + return os.path.join(self.path, name) def get_packed_refs(self): """Get contents of the packed-refs file. :return: Dictionary mapping ref names to SHA1s :note: Will return an empty dictionary when no packed-refs file is present. """ # TODO: invalidate the cache on repacking if self._packed_refs is None: # set both to empty because we want _peeled_refs to be # None if and only if _packed_refs is also None. self._packed_refs = {} self._peeled_refs = {} path = os.path.join(self.path, 'packed-refs') try: f = GitFile(path, 'rb') except IOError as e: if e.errno == errno.ENOENT: return {} raise with f: first_line = next(iter(f)).rstrip() if (first_line.startswith(b'# pack-refs') and b' peeled' in first_line): for sha, name, peeled in read_packed_refs_with_peeled(f): self._packed_refs[name] = sha if peeled: self._peeled_refs[name] = peeled else: f.seek(0) for sha, name in read_packed_refs(f): self._packed_refs[name] = sha return self._packed_refs 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 the ref may point to a tag, but no cached information is available, None is returned. """ self.get_packed_refs() if self._peeled_refs is None or name not in self._packed_refs: # No cache: no peeled refs were read, or this ref is loose return None if name in self._peeled_refs: return self._peeled_refs[name] else: # Known not peelable return self[name] def read_loose_ref(self, name): """Read a reference file and return its contents. If the reference file a symbolic reference, only read the first line of the file. Otherwise, only read the first 40 bytes. :param name: the refname to read, relative to refpath :return: The contents of the ref file, or None if the file does not exist. :raises IOError: if any other error occurs """ filename = self.refpath(name) try: with GitFile(filename, 'rb') as f: header = f.read(len(SYMREF)) if header == SYMREF: # Read only the first line return header + next(iter(f)).rstrip(b'\r\n') else: # Read only the first 40 bytes return header + f.read(40 - len(SYMREF)) except IOError as e: if e.errno == errno.ENOENT: return None raise def _remove_packed_ref(self, name): if self._packed_refs is None: return filename = os.path.join(self.path, 'packed-refs') # reread cached refs from disk, while holding the lock f = GitFile(filename, 'wb') try: self._packed_refs = None self.get_packed_refs() if name not in self._packed_refs: return del self._packed_refs[name] if name in self._peeled_refs: del self._peeled_refs[name] write_packed_refs(f, self._packed_refs, self._peeled_refs) f.close() finally: f.abort() def set_symbolic_ref(self, name, other): """Make a ref point at another ref. :param name: Name of the ref to set :param other: Name of the ref to point at """ self._check_refname(name) self._check_refname(other) filename = self.refpath(name) try: f = GitFile(filename, 'wb') try: f.write(SYMREF + other + b'\n') except (IOError, OSError): f.abort() raise finally: f.close() def set_if_equals(self, name, old_ref, new_ref): """Set a refname to new_ref only if it currently equals old_ref. This method follows all symbolic references, and can be used to perform an atomic compare-and-swap operation. :param name: The refname to set. :param old_ref: The old sha the refname must refer to, or None to set unconditionally. :param new_ref: The new sha the refname will refer to. :return: True if the set was successful, False otherwise. """ self._check_refname(name) try: realnames, _ = self.follow(name) realname = realnames[-1] except (KeyError, IndexError): realname = name filename = self.refpath(realname) ensure_dir_exists(os.path.dirname(filename)) with GitFile(filename, 'wb') as f: if old_ref is not None: try: # read again while holding the lock orig_ref = self.read_loose_ref(realname) if orig_ref is None: orig_ref = self.get_packed_refs().get(realname, ZERO_SHA) if orig_ref != old_ref: f.abort() return False except (OSError, IOError): f.abort() raise try: f.write(new_ref + b'\n') except (OSError, IOError): f.abort() raise return True def add_if_new(self, name, ref): """Add a new reference only if it does not already exist. This method follows symrefs, and only ensures that the last ref in the chain does not exist. :param name: The refname to set. :param ref: The new sha the refname will refer to. :return: True if the add was successful, False otherwise. """ try: realnames, contents = self.follow(name) if contents is not None: return False realname = realnames[-1] except (KeyError, IndexError): realname = name self._check_refname(realname) filename = self.refpath(realname) ensure_dir_exists(os.path.dirname(filename)) with GitFile(filename, 'wb') as f: if os.path.exists(filename) or name in self.get_packed_refs(): f.abort() return False try: f.write(ref + b'\n') except (OSError, IOError): f.abort() raise return True def remove_if_equals(self, name, old_ref): """Remove a refname only if it currently equals old_ref. This method does not follow symbolic references. It can be used to perform an atomic compare-and-delete operation. :param name: The refname to delete. :param old_ref: The old sha the refname must refer to, or None to delete unconditionally. :return: True if the delete was successful, False otherwise. """ self._check_refname(name) filename = self.refpath(name) ensure_dir_exists(os.path.dirname(filename)) f = GitFile(filename, 'wb') try: if old_ref is not None: orig_ref = self.read_loose_ref(name) if orig_ref is None: orig_ref = self.get_packed_refs().get(name, ZERO_SHA) if orig_ref != old_ref: return False # may only be packed try: os.remove(filename) except OSError as e: if e.errno != errno.ENOENT: raise self._remove_packed_ref(name) finally: # never write, we just wanted the lock f.abort() return True def _split_ref_line(line): """Split a single ref line into a tuple of SHA1 and name.""" fields = line.rstrip(b'\n').split(b' ') if len(fields) != 2: raise PackedRefsException("invalid ref line %r" % line) sha, name = fields if not valid_hexsha(sha): raise PackedRefsException("Invalid hex sha %r" % sha) if not check_ref_format(name): raise PackedRefsException("invalid ref name %r" % name) return (sha, name) def read_packed_refs(f): """Read a packed refs file. :param f: file-like object to read from :return: Iterator over tuples with SHA1s and ref names. """ for l in f: if l.startswith(b'#'): # Comment continue if l.startswith(b'^'): raise PackedRefsException( "found peeled ref in packed-refs without peeled") yield _split_ref_line(l) def read_packed_refs_with_peeled(f): """Read a packed refs file including peeled refs. Assumes the "# pack-refs with: peeled" line was already read. Yields tuples with ref names, SHA1s, and peeled SHA1s (or None). :param f: file-like object to read from, seek'ed to the second line """ last = None for l in f: if l[0] == b'#': continue l = l.rstrip(b'\r\n') if l.startswith(b'^'): if not last: raise PackedRefsException("unexpected peeled ref line") if not valid_hexsha(l[1:]): raise PackedRefsException("Invalid hex sha %r" % l[1:]) sha, name = _split_ref_line(last) last = None yield (sha, name, l[1:]) else: if last: sha, name = _split_ref_line(last) yield (sha, name, None) last = l if last: sha, name = _split_ref_line(last) yield (sha, name, None) def write_packed_refs(f, packed_refs, peeled_refs=None): """Write a packed refs file. :param f: empty file-like object to write to :param packed_refs: dict of refname to sha of packed refs to write :param peeled_refs: dict of refname to peeled value of sha """ if peeled_refs is None: peeled_refs = {} else: f.write(b'# pack-refs with: peeled\n') for refname in sorted(packed_refs.keys()): f.write(git_line(packed_refs[refname], refname)) if refname in peeled_refs: f.write(b'^' + peeled_refs[refname] + b'\n') def read_info_refs(f): ret = {} for l in f.readlines(): (sha, name) = l.rstrip(b"\r\n").split(b"\t", 1) ret[name] = sha return ret def write_info_refs(refs, store): """Generate info refs.""" for name, sha in sorted(refs.items()): # get_refs() includes HEAD as a special case, but we don't want to # advertise it if name == b'HEAD': continue try: o = store[sha] except KeyError: continue peeled = store.peel_sha(sha) yield o.id + b'\t' + name + b'\n' if o.id != peeled.id: yield peeled.id + b'\t' + name + b'^{}\n' is_local_branch = lambda x: x.startswith(b'refs/heads/') diff --git a/dulwich/repo.py b/dulwich/repo.py index 210af72a..3a945af5 100644 --- a/dulwich/repo.py +++ b/dulwich/repo.py @@ -1,1027 +1,1043 @@ # repo.py -- For dealing with git repositories. # Copyright (C) 2007 James Westby # Copyright (C) 2008-2013 Jelmer Vernooij # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Repository access. This module contains the base class for git repositories (BaseRepo) and an implementation which uses a repository on local disk (Repo). """ from io import BytesIO import errno import os import sys from dulwich.errors import ( NoIndexPresent, NotBlobError, NotCommitError, NotGitRepository, NotTreeError, NotTagError, CommitError, RefFormatError, HookError, ) from dulwich.file import ( GitFile, ) from dulwich.object_store import ( DiskObjectStore, MemoryObjectStore, ObjectStoreGraphWalker, ) from dulwich.objects import ( check_hexsha, Blob, Commit, ShaFile, Tag, Tree, ) from dulwich.hooks import ( PreCommitShellHook, PostCommitShellHook, CommitMsgShellHook, ) from dulwich.refs import ( check_ref_format, RefsContainer, DictRefsContainer, InfoRefsContainer, DiskRefsContainer, read_packed_refs, read_packed_refs_with_peeled, write_packed_refs, SYMREF, ) import warnings CONTROLDIR = '.git' OBJECTDIR = 'objects' REFSDIR = 'refs' REFSDIR_TAGS = 'tags' REFSDIR_HEADS = 'heads' INDEX_FILENAME = "index" +COMMONDIR = 'commondir' BASE_DIRECTORIES = [ ["branches"], [REFSDIR], [REFSDIR, REFSDIR_TAGS], [REFSDIR, REFSDIR_HEADS], ["hooks"], ["info"] ] def parse_graftpoints(graftpoints): """Convert a list of graftpoints into a dict :param graftpoints: Iterator of graftpoint lines Each line is formatted as: []* Resulting dictionary is: : [*] https://git.wiki.kernel.org/index.php/GraftPoint """ grafts = {} for l in graftpoints: raw_graft = l.split(None, 1) commit = raw_graft[0] if len(raw_graft) == 2: parents = raw_graft[1].split() else: parents = [] for sha in [commit] + parents: check_hexsha(sha, 'Invalid graftpoint') grafts[commit] = parents return grafts def serialize_graftpoints(graftpoints): """Convert a dictionary of grafts into string The graft dictionary is: : [*] Each line is formatted as: []* https://git.wiki.kernel.org/index.php/GraftPoint """ graft_lines = [] for commit, parents in graftpoints.items(): if parents: graft_lines.append(commit + b' ' + b' '.join(parents)) else: graft_lines.append(commit) return b'\n'.join(graft_lines) class BaseRepo(object): """Base class for a git repository. :ivar object_store: Dictionary-like object for accessing the objects :ivar refs: Dictionary-like object with the refs in this repository """ def __init__(self, object_store, refs): """Open a repository. This shouldn't be called directly, but rather through one of the base classes, such as MemoryRepo or Repo. :param object_store: Object store to use :param refs: Refs container to use """ self.object_store = object_store self.refs = refs self._graftpoints = {} self.hooks = {} def _init_files(self, bare): """Initialize a default set of named files.""" from dulwich.config import ConfigFile self._put_named_file('description', b"Unnamed repository") f = BytesIO() cf = ConfigFile() cf.set(b"core", b"repositoryformatversion", b"0") cf.set(b"core", b"filemode", b"true") cf.set(b"core", b"bare", bare) cf.set(b"core", b"logallrefupdates", True) cf.write_to_file(f) self._put_named_file('config', f.getvalue()) self._put_named_file(os.path.join('info', 'exclude'), b'') def get_named_file(self, path): """Get a file from the control dir with a specific name. Although the filename should be interpreted as a filename relative to the control dir in a disk-based Repo, the object returned need not be pointing to a file in that location. :param path: The path to the file, relative to the control dir. :return: An open file object, or None if the file does not exist. """ raise NotImplementedError(self.get_named_file) def _put_named_file(self, path, contents): """Write a file to the control dir with the given name and contents. :param path: The path to the file, relative to the control dir. :param contents: A string to write to the file. """ raise NotImplementedError(self._put_named_file) def open_index(self): """Open the index for this repository. :raise NoIndexPresent: If no index is present :return: The matching `Index` """ raise NotImplementedError(self.open_index) def fetch(self, target, determine_wants=None, progress=None): """Fetch objects into another repository. :param target: The target repository :param determine_wants: Optional function to determine what refs to fetch. :param progress: Optional progress function :return: The local refs """ if determine_wants is None: determine_wants = target.object_store.determine_wants_all target.object_store.add_objects( self.fetch_objects(determine_wants, target.get_graph_walker(), progress)) return self.get_refs() def fetch_objects(self, determine_wants, graph_walker, progress, get_tagged=None): """Fetch the missing objects required for a set of revisions. :param determine_wants: Function that takes a dictionary with heads and returns the list of heads to fetch. :param graph_walker: Object that can iterate over the list of revisions to fetch and has an "ack" method that will be called to acknowledge that a revision is present. :param progress: Simple progress function that will be called with updated progress strings. :param get_tagged: Function that returns a dict of pointed-to sha -> tag sha for including tags. :return: iterator over objects, with __len__ implemented """ wants = determine_wants(self.get_refs()) if not isinstance(wants, list): raise TypeError("determine_wants() did not return a list") shallows = getattr(graph_walker, 'shallow', frozenset()) unshallows = getattr(graph_walker, 'unshallow', frozenset()) if wants == []: # TODO(dborowitz): find a way to short-circuit that doesn't change # this interface. if shallows or unshallows: # Do not send a pack in shallow short-circuit path return None return [] # If the graph walker is set up with an implementation that can # ACK/NAK to the wire, it will write data to the client through # this call as a side-effect. haves = self.object_store.find_common_revisions(graph_walker) # Deal with shallow requests separately because the haves do # not reflect what objects are missing if shallows or unshallows: haves = [] # TODO: filter the haves commits from iter_shas. # the specific commits aren't missing. def get_parents(commit): if commit.id in shallows: return [] return self.get_parents(commit.id, commit) return self.object_store.iter_shas( self.object_store.find_missing_objects( haves, wants, progress, get_tagged, get_parents=get_parents)) def get_graph_walker(self, heads=None): """Retrieve a graph walker. A graph walker is used by a remote repository (or proxy) to find out which objects are present in this repository. :param heads: Repository heads to use (optional) :return: A graph walker object """ if heads is None: heads = self.refs.as_dict(b'refs/heads').values() return ObjectStoreGraphWalker(heads, self.get_parents) def get_refs(self): """Get dictionary with all refs. :return: A ``dict`` mapping ref names to SHA1s """ return self.refs.as_dict() def head(self): """Return the SHA1 pointed at by HEAD.""" return self.refs[b'HEAD'] def _get_object(self, sha, cls): assert len(sha) in (20, 40) ret = self.get_object(sha) if not isinstance(ret, cls): if cls is Commit: raise NotCommitError(ret) elif cls is Blob: raise NotBlobError(ret) elif cls is Tree: raise NotTreeError(ret) elif cls is Tag: raise NotTagError(ret) else: raise Exception("Type invalid: %r != %r" % ( ret.type_name, cls.type_name)) return ret def get_object(self, sha): """Retrieve the object with the specified SHA. :param sha: SHA to retrieve :return: A ShaFile object :raise KeyError: when the object can not be found """ return self.object_store[sha] def get_parents(self, sha, commit=None): """Retrieve the parents of a specific commit. If the specific commit is a graftpoint, the graft parents will be returned instead. :param sha: SHA of the commit for which to retrieve the parents :param commit: Optional commit matching the sha :return: List of parents """ try: return self._graftpoints[sha] except KeyError: if commit is None: commit = self[sha] return commit.parents def get_config(self): """Retrieve the config object. :return: `ConfigFile` object for the ``.git/config`` file. """ raise NotImplementedError(self.get_config) def get_description(self): """Retrieve the description for this repository. :return: String with the description of the repository as set by the user. """ raise NotImplementedError(self.get_description) def set_description(self, description): """Set the description for this repository. :param description: Text to set as description for this repository. """ raise NotImplementedError(self.set_description) def get_config_stack(self): """Return a config stack for this repository. This stack accesses the configuration for both this repository itself (.git/config) and the global configuration, which usually lives in ~/.gitconfig. :return: `Config` instance for this repository """ from dulwich.config import StackedConfig backends = [self.get_config()] + StackedConfig.default_backends() return StackedConfig(backends, writable=backends[0]) def get_peeled(self, ref): """Get the peeled value of a ref. :param ref: The refname to peel. :return: The fully-peeled SHA1 of a tag object, after peeling all intermediate tags; if the original ref does not point to a tag, this will equal the original SHA1. """ cached = self.refs.get_peeled(ref) if cached is not None: return cached return self.object_store.peel_sha(self.refs[ref]).id def get_walker(self, include=None, *args, **kwargs): """Obtain a walker for this repository. :param include: Iterable of SHAs of commits to include along with their ancestors. Defaults to [HEAD] :param exclude: Iterable of SHAs of commits to exclude along with their ancestors, overriding includes. :param order: ORDER_* constant specifying the order of results. Anything other than ORDER_DATE may result in O(n) memory usage. :param reverse: If True, reverse the order of output, requiring O(n) memory. :param max_entries: The maximum number of entries to yield, or None for no limit. :param paths: Iterable of file or subtree paths to show entries for. :param rename_detector: diff.RenameDetector object for detecting renames. :param follow: If True, follow path across renames/copies. Forces a default rename_detector. :param since: Timestamp to list commits after. :param until: Timestamp to list commits before. :param queue_cls: A class to use for a queue of commits, supporting the iterator protocol. The constructor takes a single argument, the Walker. :return: A `Walker` object """ from dulwich.walk import Walker if include is None: include = [self.head()] if isinstance(include, str): include = [include] kwargs['get_parents'] = lambda commit: self.get_parents(commit.id, commit) return Walker(self.object_store, include, *args, **kwargs) def __getitem__(self, name): """Retrieve a Git object by SHA1 or ref. :param name: A Git object SHA1 or a ref name :return: A `ShaFile` object, such as a Commit or Blob :raise KeyError: when the specified ref or object does not exist """ if not isinstance(name, bytes): raise TypeError("'name' must be bytestring, not %.80s" % type(name).__name__) if len(name) in (20, 40): try: return self.object_store[name] except (KeyError, ValueError): pass try: return self.object_store[self.refs[name]] except RefFormatError: raise KeyError(name) def __contains__(self, name): """Check if a specific Git object or ref is present. :param name: Git object SHA1 or ref name """ if len(name) in (20, 40): return name in self.object_store or name in self.refs else: return name in self.refs def __setitem__(self, name, value): """Set a ref. :param name: ref name :param value: Ref value - either a ShaFile object, or a hex sha """ if name.startswith(b"refs/") or name == b'HEAD': if isinstance(value, ShaFile): self.refs[name] = value.id elif isinstance(value, bytes): self.refs[name] = value else: raise TypeError(value) else: raise ValueError(name) def __delitem__(self, name): """Remove a ref. :param name: Name of the ref to remove """ if name.startswith(b"refs/") or name == b"HEAD": del self.refs[name] else: raise ValueError(name) def _get_user_identity(self): """Determine the identity to use for new commits. """ config = self.get_config_stack() return (config.get((b"user", ), b"name") + b" <" + config.get((b"user", ), b"email") + b">") def _add_graftpoints(self, updated_graftpoints): """Add or modify graftpoints :param updated_graftpoints: Dict of commit shas to list of parent shas """ # Simple validation for commit, parents in updated_graftpoints.items(): for sha in [commit] + parents: check_hexsha(sha, 'Invalid graftpoint') self._graftpoints.update(updated_graftpoints) def _remove_graftpoints(self, to_remove=[]): """Remove graftpoints :param to_remove: List of commit shas """ for sha in to_remove: del self._graftpoints[sha] def do_commit(self, message=None, committer=None, author=None, commit_timestamp=None, commit_timezone=None, author_timestamp=None, author_timezone=None, tree=None, encoding=None, ref=b'HEAD', merge_heads=None): """Create a new commit. :param message: Commit message :param committer: Committer fullname :param author: Author fullname (defaults to committer) :param commit_timestamp: Commit timestamp (defaults to now) :param commit_timezone: Commit timestamp timezone (defaults to GMT) :param author_timestamp: Author timestamp (defaults to commit timestamp) :param author_timezone: Author timestamp timezone (defaults to commit timestamp timezone) :param tree: SHA1 of the tree root to use (if not specified the current index will be committed). :param encoding: Encoding :param ref: Optional ref to commit to (defaults to current branch) :param merge_heads: Merge heads (defaults to .git/MERGE_HEADS) :return: New commit SHA1 """ import time c = Commit() if tree is None: index = self.open_index() c.tree = index.commit(self.object_store) else: if len(tree) != 40: raise ValueError("tree must be a 40-byte hex sha string") c.tree = tree try: self.hooks['pre-commit'].execute() except HookError as e: raise CommitError(e) except KeyError: # no hook defined, silent fallthrough pass if merge_heads is None: # FIXME: Read merge heads from .git/MERGE_HEADS merge_heads = [] if committer is None: # FIXME: Support GIT_COMMITTER_NAME/GIT_COMMITTER_EMAIL environment # variables committer = self._get_user_identity() c.committer = committer if commit_timestamp is None: # FIXME: Support GIT_COMMITTER_DATE environment variable commit_timestamp = time.time() c.commit_time = int(commit_timestamp) if commit_timezone is None: # FIXME: Use current user timezone rather than UTC commit_timezone = 0 c.commit_timezone = commit_timezone if author is None: # FIXME: Support GIT_AUTHOR_NAME/GIT_AUTHOR_EMAIL environment # variables author = committer c.author = author if author_timestamp is None: # FIXME: Support GIT_AUTHOR_DATE environment variable author_timestamp = commit_timestamp c.author_time = int(author_timestamp) if author_timezone is None: author_timezone = commit_timezone c.author_timezone = author_timezone if encoding is not None: c.encoding = encoding if message is None: # FIXME: Try to read commit message from .git/MERGE_MSG raise ValueError("No commit message specified") try: c.message = self.hooks['commit-msg'].execute(message) if c.message is None: c.message = message except HookError as e: raise CommitError(e) except KeyError: # no hook defined, message not modified c.message = message if ref is None: # Create a dangling commit c.parents = merge_heads self.object_store.add_object(c) else: try: old_head = self.refs[ref] c.parents = [old_head] + merge_heads self.object_store.add_object(c) ok = self.refs.set_if_equals(ref, old_head, c.id) except KeyError: c.parents = merge_heads self.object_store.add_object(c) ok = self.refs.add_if_new(ref, c.id) if not ok: # Fail if the atomic compare-and-swap failed, leaving the commit and # all its objects as garbage. raise CommitError("%s changed during commit" % (ref,)) try: self.hooks['post-commit'].execute() except HookError as e: # silent failure warnings.warn("post-commit hook failed: %s" % e, UserWarning) except KeyError: # no hook defined, silent fallthrough pass return c.id def read_gitfile(f): """Read a ``.git`` file. The first line of the file should start with "gitdir: " :param f: File-like object to read from :return: A path """ cs = f.read() if not cs.startswith("gitdir: "): raise ValueError("Expected file to start with 'gitdir: '") return cs[len("gitdir: "):].rstrip("\n") class Repo(BaseRepo): """A git repository backed by local disk. To open an existing repository, call the contructor with the path of the repository. To create a new repository, use the Repo.init class method. """ def __init__(self, root): hidden_path = os.path.join(root, CONTROLDIR) if os.path.isdir(os.path.join(hidden_path, OBJECTDIR)): self.bare = False self._controldir = hidden_path elif (os.path.isdir(os.path.join(root, OBJECTDIR)) and os.path.isdir(os.path.join(root, REFSDIR))): self.bare = True self._controldir = root elif os.path.isfile(hidden_path): self.bare = False with open(hidden_path, 'r') as f: path = read_gitfile(f) self.bare = False self._controldir = os.path.join(root, path) else: raise NotGitRepository( "No git repository was found at %(path)s" % dict(path=root) ) + commondir = self.get_named_file(COMMONDIR) + if commondir: + self._commondir = os.path.join(self.controldir(), commondir.read().rstrip("\n")) + else: + self._commondir = self._controldir self.path = root - object_store = DiskObjectStore(os.path.join(self.controldir(), + object_store = DiskObjectStore(os.path.join(self.commondir(), OBJECTDIR)) - refs = DiskRefsContainer(self.controldir()) + refs = DiskRefsContainer(self.commondir(), worktree=self._controldir) BaseRepo.__init__(self, object_store, refs) self._graftpoints = {} graft_file = self.get_named_file(os.path.join("info", "grafts")) if graft_file: with graft_file: self._graftpoints.update(parse_graftpoints(graft_file)) graft_file = self.get_named_file("shallow") if graft_file: with graft_file: self._graftpoints.update(parse_graftpoints(graft_file)) self.hooks['pre-commit'] = PreCommitShellHook(self.controldir()) self.hooks['commit-msg'] = CommitMsgShellHook(self.controldir()) self.hooks['post-commit'] = PostCommitShellHook(self.controldir()) @classmethod def discover(cls, start='.'): """Iterate parent directories to discover a repository Return a Repo object for the first parent directory that looks like a Git repository. :param start: The directory to start discovery from (defaults to '.') """ remaining = True path = os.path.abspath(start) while remaining: try: return cls(path) except NotGitRepository: path, remaining = os.path.split(path) raise NotGitRepository( "No git repository was found at %(path)s" % dict(path=start) ) def controldir(self): """Return the path of the control directory.""" return self._controldir + def commondir(self): + """Return the path of the common directory. + + For a main working tree, it is identical to `controldir()`. + + For a linked working tree, it is the control directory of the + main working tree.""" + + return self._commondir + def _put_named_file(self, path, contents): """Write a file to the control dir with the given name and contents. :param path: The path to the file, relative to the control dir. :param contents: A string to write to the file. """ path = path.lstrip(os.path.sep) with GitFile(os.path.join(self.controldir(), path), 'wb') as f: f.write(contents) def get_named_file(self, path): """Get a file from the control dir with a specific name. Although the filename should be interpreted as a filename relative to the control dir in a disk-based Repo, the object returned need not be pointing to a file in that location. :param path: The path to the file, relative to the control dir. :return: An open file object, or None if the file does not exist. """ # TODO(dborowitz): sanitize filenames, since this is used directly by # the dumb web serving code. path = path.lstrip(os.path.sep) try: return open(os.path.join(self.controldir(), path), 'rb') except (IOError, OSError) as e: if e.errno == errno.ENOENT: return None raise def index_path(self): """Return path to the index file.""" return os.path.join(self.controldir(), INDEX_FILENAME) def open_index(self): """Open the index for this repository. :raise NoIndexPresent: If no index is present :return: The matching `Index` """ from dulwich.index import Index if not self.has_index(): raise NoIndexPresent() return Index(self.index_path()) def has_index(self): """Check if an index is present.""" # Bare repos must never have index files; non-bare repos may have a # missing index file, which is treated as empty. return not self.bare def stage(self, fs_paths): """Stage a set of paths. :param fs_paths: List of paths, relative to the repository path """ root_path_bytes = self.path.encode(sys.getfilesystemencoding()) if not isinstance(fs_paths, list): fs_paths = [fs_paths] from dulwich.index import ( blob_from_path_and_stat, index_entry_from_stat, _fs_to_tree_path, ) index = self.open_index() for fs_path in fs_paths: if not isinstance(fs_path, bytes): fs_path = fs_path.encode(sys.getfilesystemencoding()) tree_path = _fs_to_tree_path(fs_path) full_path = os.path.join(root_path_bytes, fs_path) try: st = os.lstat(full_path) except OSError: # File no longer exists try: del index[tree_path] except KeyError: pass # already removed else: blob = blob_from_path_and_stat(full_path, st) self.object_store.add_object(blob) index[tree_path] = index_entry_from_stat(st, blob.id, 0) index.write() def clone(self, target_path, mkdir=True, bare=False, origin=b"origin"): """Clone this repository. :param target_path: Target path :param mkdir: Create the target directory :param bare: Whether to create a bare repository :param origin: Base name for refs in target repository cloned from this repository :return: Created repository as `Repo` """ if not bare: target = self.init(target_path, mkdir=mkdir) else: target = self.init_bare(target_path) self.fetch(target) target.refs.import_refs( b'refs/remotes/' + origin, self.refs.as_dict(b'refs/heads')) target.refs.import_refs( b'refs/tags', self.refs.as_dict(b'refs/tags')) try: target.refs.add_if_new( b'refs/heads/master', self.refs[b'refs/heads/master']) except KeyError: pass # Update target head head_chain, head_sha = self.refs.follow(b'HEAD') if head_chain and head_sha is not None: target.refs.set_symbolic_ref(b'HEAD', head_chain[-1]) target[b'HEAD'] = head_sha if not bare: # Checkout HEAD to target dir target.reset_index() return target def reset_index(self, tree=None): """Reset the index back to a specific tree. :param tree: Tree SHA to reset to, None for current HEAD tree. """ from dulwich.index import ( build_index_from_tree, validate_path_element_default, validate_path_element_ntfs, ) if tree is None: tree = self[b'HEAD'].tree config = self.get_config() honor_filemode = config.get_boolean('core', 'filemode', os.name != "nt") if config.get_boolean('core', 'core.protectNTFS', os.name == "nt"): validate_path_element = validate_path_element_ntfs else: validate_path_element = validate_path_element_default return build_index_from_tree(self.path, self.index_path(), self.object_store, tree, honor_filemode=honor_filemode, validate_path_element=validate_path_element) def get_config(self): """Retrieve the config object. :return: `ConfigFile` object for the ``.git/config`` file. """ from dulwich.config import ConfigFile path = os.path.join(self._controldir, 'config') try: return ConfigFile.from_path(path) except (IOError, OSError) as e: if e.errno != errno.ENOENT: raise ret = ConfigFile() ret.path = path return ret def get_description(self): """Retrieve the description of this repository. :return: A string describing the repository or None. """ path = os.path.join(self._controldir, 'description') try: with GitFile(path, 'rb') as f: return f.read() except (IOError, OSError) as e: if e.errno != errno.ENOENT: raise return None def __repr__(self): return "" % self.path def set_description(self, description): """Set the description for this repository. :param description: Text to set as description for this repository. """ self._put_named_file('description', description) @classmethod def _init_maybe_bare(cls, path, bare): for d in BASE_DIRECTORIES: os.mkdir(os.path.join(path, *d)) DiskObjectStore.init(os.path.join(path, OBJECTDIR)) ret = cls(path) ret.refs.set_symbolic_ref(b'HEAD', b"refs/heads/master") ret._init_files(bare) return ret @classmethod def init(cls, path, mkdir=False): """Create a new repository. :param path: Path in which to create the repository :param mkdir: Whether to create the directory :return: `Repo` instance """ if mkdir: os.mkdir(path) controldir = os.path.join(path, CONTROLDIR) os.mkdir(controldir) cls._init_maybe_bare(controldir, False) return cls(path) @classmethod def init_bare(cls, path): """Create a new bare repository. ``path`` should already exist and be an emty directory. :param path: Path to create bare repository in :return: a `Repo` instance """ return cls._init_maybe_bare(path, True) create = init_bare def close(self): """Close any files opened by this repository.""" self.object_store.close() class MemoryRepo(BaseRepo): """Repo that stores refs, objects, and named files in memory. MemoryRepos are always bare: they have no working tree and no index, since those have a stronger dependency on the filesystem. """ def __init__(self): from dulwich.config import ConfigFile BaseRepo.__init__(self, MemoryObjectStore(), DictRefsContainer({})) self._named_files = {} self.bare = True self._config = ConfigFile() def _put_named_file(self, path, contents): """Write a file to the control dir with the given name and contents. :param path: The path to the file, relative to the control dir. :param contents: A string to write to the file. """ self._named_files[path] = contents def get_named_file(self, path): """Get a file from the control dir with a specific name. Although the filename should be interpreted as a filename relative to the control dir in a disk-baked Repo, the object returned need not be pointing to a file in that location. :param path: The path to the file, relative to the control dir. :return: An open file object, or None if the file does not exist. """ contents = self._named_files.get(path, None) if contents is None: return None return BytesIO(contents) def open_index(self): """Fail to open index for this repo, since it is bare. :raise NoIndexPresent: Raised when no index is present """ raise NoIndexPresent() def get_config(self): """Retrieve the config object. :return: `ConfigFile` object. """ return self._config def get_description(self): """Retrieve the repository description. This defaults to None, for no description. """ return None @classmethod def init_bare(cls, objects, refs): """Create a new bare repository in memory. :param objects: Objects for the new repository, as iterable :param refs: Refs as dictionary, mapping names to object SHA1s """ ret = cls() for obj in objects: ret.object_store.add_object(obj) for refname, sha in refs.items(): ret.refs[refname] = sha ret._init_files(bare=True) return ret diff --git a/dulwich/tests/compat/test_repository.py b/dulwich/tests/compat/test_repository.py index c881a72f..cf3d2065 100644 --- a/dulwich/tests/compat/test_repository.py +++ b/dulwich/tests/compat/test_repository.py @@ -1,121 +1,158 @@ # test_repo.py -- Git repo compatibility tests # Copyright (C) 2010 Google, Inc. # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Compatibility tests for dulwich repositories.""" from io import BytesIO from itertools import chain import os from dulwich.objects import ( hex_to_sha, ) from dulwich.repo import ( check_ref_format, + Repo, ) +from dulwich.tests import skipIf + from dulwich.tests.compat.utils import ( run_git_or_fail, CompatTestCase, - ) + rmtree_ro, + git_version +) class ObjectStoreTestCase(CompatTestCase): """Tests for git repository compatibility.""" def setUp(self): super(ObjectStoreTestCase, self).setUp() self._repo = self.import_repo('server_new.export') + def repo_path(self): + return self._repo.path + def _run_git(self, args): return run_git_or_fail(args, cwd=self._repo.path) def _parse_refs(self, output): refs = {} for line in BytesIO(output): fields = line.rstrip(b'\n').split(b' ') self.assertEqual(3, len(fields)) refname, type_name, sha = fields check_ref_format(refname[5:]) hex_to_sha(sha) refs[refname] = (type_name, sha) return refs def _parse_objects(self, output): return set(s.rstrip(b'\n').split(b' ')[0] for s in BytesIO(output)) def test_bare(self): self.assertTrue(self._repo.bare) self.assertFalse(os.path.exists(os.path.join(self._repo.path, '.git'))) def test_head(self): output = self._run_git(['rev-parse', 'HEAD']) head_sha = output.rstrip(b'\n') hex_to_sha(head_sha) self.assertEqual(head_sha, self._repo.refs[b'HEAD']) def test_refs(self): output = self._run_git( ['for-each-ref', '--format=%(refname) %(objecttype) %(objectname)']) expected_refs = self._parse_refs(output) actual_refs = {} for refname, sha in self._repo.refs.as_dict().items(): if refname == b'HEAD': continue # handled in test_head obj = self._repo[sha] self.assertEqual(sha, obj.id) actual_refs[refname] = (obj.type_name, obj.id) self.assertEqual(expected_refs, actual_refs) # TODO(dborowitz): peeled ref tests def _get_loose_shas(self): output = self._run_git(['rev-list', '--all', '--objects', '--unpacked']) return self._parse_objects(output) def _get_all_shas(self): output = self._run_git(['rev-list', '--all', '--objects']) return self._parse_objects(output) def assertShasMatch(self, expected_shas, actual_shas_iter): actual_shas = set() for sha in actual_shas_iter: obj = self._repo[sha] self.assertEqual(sha, obj.id) actual_shas.add(sha) self.assertEqual(expected_shas, actual_shas) def test_loose_objects(self): # TODO(dborowitz): This is currently not very useful since fast-imported # repos only contained packed objects. expected_shas = self._get_loose_shas() self.assertShasMatch(expected_shas, self._repo.object_store._iter_loose_objects()) def test_packed_objects(self): expected_shas = self._get_all_shas() - self._get_loose_shas() self.assertShasMatch(expected_shas, chain(*self._repo.object_store.packs)) def test_all_objects(self): expected_shas = self._get_all_shas() self.assertShasMatch(expected_shas, iter(self._repo.object_store)) + + +@skipIf(git_version() < (2, 5), 'Git version must be >= 2.5') +class WorkingTreeTestCase(ObjectStoreTestCase): + """Test for compatibility with git-worktree.""" + + def setUp(self): + super(WorkingTreeTestCase, self).setUp() + self._worktree_path = self.create_new_worktree(self.repo_path()) + self._worktree_repo = Repo(self._worktree_path) + self._mainworktree_repo = self._repo + self._repo = self._worktree_repo + + def tearDown(self): + self._worktree_repo.close() + rmtree_ro(self._worktree_path) + self._repo = self._mainworktree_repo + super(WorkingTreeTestCase, self).tearDown() + + def test_refs(self): + super(WorkingTreeTestCase, self).test_refs() + self.assertTrue(self._mainworktree_repo.refs.allkeys()==\ + self._repo.refs.allkeys()) + self.assertFalse(self._repo.refs[b'HEAD']==\ + self._mainworktree_repo.refs[b'HEAD']) + + def test_bare(self): + self.assertFalse(self._repo.bare) + self.assertTrue(os.path.isfile(os.path.join(self._repo.path, '.git'))) diff --git a/dulwich/tests/compat/utils.py b/dulwich/tests/compat/utils.py index 6fd7e30f..8b4d6fef 100644 --- a/dulwich/tests/compat/utils.py +++ b/dulwich/tests/compat/utils.py @@ -1,254 +1,265 @@ # utils.py -- Git compatibility utilities # Copyright (C) 2010 Google, Inc. # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Utilities for interacting with cgit.""" import errno import functools import os import shutil import socket import stat import subprocess import sys import tempfile import time from dulwich.repo import Repo from dulwich.protocol import TCP_GIT_PORT from dulwich.tests import ( SkipTest, TestCase, ) _DEFAULT_GIT = 'git' _VERSION_LEN = 4 _REPOS_DATA_DIR = os.path.abspath(os.path.join( os.path.dirname(__file__), os.pardir, 'data', 'repos')) def git_version(git_path=_DEFAULT_GIT): """Attempt to determine the version of git currently installed. :param git_path: Path to the git executable; defaults to the version in the system path. :return: A tuple of ints of the form (major, minor, point, sub-point), or None if no git installation was found. """ try: output = run_git_or_fail(['--version'], git_path=git_path) except OSError: return None version_prefix = b'git version ' if not output.startswith(version_prefix): return None parts = output[len(version_prefix):].split(b'.') nums = [] for part in parts: try: nums.append(int(part)) except ValueError: break while len(nums) < _VERSION_LEN: nums.append(0) return tuple(nums[:_VERSION_LEN]) def require_git_version(required_version, git_path=_DEFAULT_GIT): """Require git version >= version, or skip the calling test. :param required_version: A tuple of ints of the form (major, minor, point, sub-point); ommitted components default to 0. :param git_path: Path to the git executable; defaults to the version in the system path. :raise ValueError: if the required version tuple has too many parts. :raise SkipTest: if no suitable git version was found at the given path. """ found_version = git_version(git_path=git_path) if found_version is None: raise SkipTest('Test requires git >= %s, but c git not found' % (required_version, )) if len(required_version) > _VERSION_LEN: raise ValueError('Invalid version tuple %s, expected %i parts' % (required_version, _VERSION_LEN)) required_version = list(required_version) while len(found_version) < len(required_version): required_version.append(0) required_version = tuple(required_version) if found_version < required_version: required_version = '.'.join(map(str, required_version)) found_version = '.'.join(map(str, found_version)) raise SkipTest('Test requires git >= %s, found %s' % (required_version, found_version)) def run_git(args, git_path=_DEFAULT_GIT, input=None, capture_stdout=False, **popen_kwargs): """Run a git command. Input is piped from the input parameter and output is sent to the standard streams, unless capture_stdout is set. :param args: A list of args to the git command. :param git_path: Path to to the git executable. :param input: Input data to be sent to stdin. :param capture_stdout: Whether to capture and return stdout. :param popen_kwargs: Additional kwargs for subprocess.Popen; stdin/stdout args are ignored. :return: A tuple of (returncode, stdout contents). If capture_stdout is False, None will be returned as stdout contents. :raise OSError: if the git executable was not found. """ env = popen_kwargs.pop('env', {}) env['LC_ALL'] = env['LANG'] = 'C' args = [git_path] + args popen_kwargs['stdin'] = subprocess.PIPE if capture_stdout: popen_kwargs['stdout'] = subprocess.PIPE else: popen_kwargs.pop('stdout', None) p = subprocess.Popen(args, env=env, **popen_kwargs) stdout, stderr = p.communicate(input=input) return (p.returncode, stdout) def run_git_or_fail(args, git_path=_DEFAULT_GIT, input=None, **popen_kwargs): """Run a git command, capture stdout/stderr, and fail if git fails.""" if 'stderr' not in popen_kwargs: popen_kwargs['stderr'] = subprocess.STDOUT returncode, stdout = run_git(args, git_path=git_path, input=input, capture_stdout=True, **popen_kwargs) if returncode != 0: raise AssertionError("git with args %r failed with %d: %r" % ( args, returncode, stdout)) return stdout def import_repo_to_dir(name): """Import a repo from a fast-export file in a temporary directory. These are used rather than binary repos for compat tests because they are more compact and human-editable, and we already depend on git. :param name: The name of the repository export file, relative to dulwich/tests/data/repos. :returns: The path to the imported repository. """ temp_dir = tempfile.mkdtemp() export_path = os.path.join(_REPOS_DATA_DIR, name) temp_repo_dir = os.path.join(temp_dir, name) export_file = open(export_path, 'rb') run_git_or_fail(['init', '--quiet', '--bare', temp_repo_dir]) run_git_or_fail(['fast-import'], input=export_file.read(), cwd=temp_repo_dir) export_file.close() return temp_repo_dir def check_for_daemon(limit=10, delay=0.1, timeout=0.1, port=TCP_GIT_PORT): """Check for a running TCP daemon. Defaults to checking 10 times with a delay of 0.1 sec between tries. :param limit: Number of attempts before deciding no daemon is running. :param delay: Delay between connection attempts. :param timeout: Socket timeout for connection attempts. :param port: Port on which we expect the daemon to appear. :returns: A boolean, true if a daemon is running on the specified port, false if not. """ for _ in range(limit): time.sleep(delay) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(delay) try: s.connect(('localhost', port)) return True except socket.timeout: pass except socket.error as e: if getattr(e, 'errno', False) and e.errno != errno.ECONNREFUSED: raise elif e.args[0] != errno.ECONNREFUSED: raise finally: s.close() return False class CompatTestCase(TestCase): """Test case that requires git for compatibility checks. Subclasses can change the git version required by overriding min_git_version. """ min_git_version = (1, 5, 0) def setUp(self): super(CompatTestCase, self).setUp() require_git_version(self.min_git_version) def assertObjectStoreEqual(self, store1, store2): self.assertEqual(sorted(set(store1)), sorted(set(store2))) def assertReposEqual(self, repo1, repo2): self.assertEqual(repo1.get_refs(), repo2.get_refs()) self.assertObjectStoreEqual(repo1.object_store, repo2.object_store) def assertReposNotEqual(self, repo1, repo2): refs1 = repo1.get_refs() objs1 = set(repo1.object_store) refs2 = repo2.get_refs() objs2 = set(repo2.object_store) self.assertFalse(refs1 == refs2 and objs1 == objs2) def import_repo(self, name): """Import a repo from a fast-export file in a temporary directory. :param name: The name of the repository export file, relative to dulwich/tests/data/repos. :returns: An initialized Repo object that lives in a temporary directory. """ path = import_repo_to_dir(name) repo = Repo(path) def cleanup(): repo.close() rmtree_ro(os.path.dirname(path.rstrip(os.sep))) self.addCleanup(cleanup) return repo + def create_new_worktree(self, repo_dir): + """Create a new worktree using git-worktree. + + :param repo_dir: The directory of the main working tree. + + :returns: The path to the new working tree. + """ + temp_dir = tempfile.mkdtemp() + run_git_or_fail(['worktree', 'add', temp_dir, 'branch'],\ + cwd=repo_dir) + return temp_dir if sys.platform == 'win32': def remove_ro(action, name, exc): os.chmod(name, stat.S_IWRITE) os.remove(name) rmtree_ro = functools.partial(shutil.rmtree, onerror=remove_ro) else: rmtree_ro = shutil.rmtree