diff --git a/dulwich/porcelain.py b/dulwich/porcelain.py index 715ead9b..0c055d46 100644 --- a/dulwich/porcelain.py +++ b/dulwich/porcelain.py @@ -1,1216 +1,1222 @@ # porcelain.py -- Porcelain-like layer on top of Dulwich # Copyright (C) 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. # """Simple wrapper that provides porcelain-like functions on top of Dulwich. Currently implemented: * archive * add * branch{_create,_delete,_list} * check-ignore * checkout * clone * commit * commit-tree * daemon * diff-tree * fetch * init * ls-remote * ls-tree * pull * push * rm * remote{_add} * receive-pack * reset * rev-list * tag{_create,_delete,_list} * upload-pack * update-server-info * status * symbolic-ref These functions are meant to behave similarly to the git subcommands. Differences in behaviour are considered bugs. Functions should generally accept both unicode strings and bytestrings """ from collections import namedtuple from contextlib import ( closing, contextmanager, ) from io import BytesIO import os import posixpath import stat import sys import time from dulwich.archive import ( tar_stream, ) from dulwich.client import ( get_transport_and_path, ) from dulwich.config import ( StackedConfig, ) from dulwich.diff_tree import ( CHANGE_ADD, CHANGE_DELETE, CHANGE_MODIFY, CHANGE_RENAME, CHANGE_COPY, RENAME_CHANGE_TYPES, ) from dulwich.errors import ( SendPackError, UpdateRefsError, ) from dulwich.ignore import IgnoreFilterManager from dulwich.index import ( blob_from_path_and_stat, get_unstaged_changes, ) from dulwich.object_store import ( tree_lookup_path, ) from dulwich.objects import ( Commit, Tag, format_timezone, parse_timezone, pretty_format_tree_entry, ) from dulwich.objectspec import ( parse_commit, parse_object, parse_ref, parse_reftuples, parse_tree, ) from dulwich.pack import ( write_pack_index, write_pack_objects, ) from dulwich.patch import write_tree_diff from dulwich.protocol import ( Protocol, ZERO_SHA, ) -from dulwich.refs import ANNOTATED_TAG_SUFFIX +from dulwich.refs import ( + ANNOTATED_TAG_SUFFIX, + strip_peeled_refs, +) from dulwich.repo import (BaseRepo, Repo) from dulwich.server import ( FileSystemBackend, TCPGitServer, ReceivePackHandler, UploadPackHandler, update_server_info as server_update_server_info, ) # Module level tuple definition for status output GitStatus = namedtuple('GitStatus', 'staged unstaged untracked') default_bytes_out_stream = getattr(sys.stdout, 'buffer', sys.stdout) default_bytes_err_stream = getattr(sys.stderr, 'buffer', sys.stderr) DEFAULT_ENCODING = 'utf-8' class RemoteExists(Exception): """Raised when the remote already exists.""" def open_repo(path_or_repo): """Open an argument that can be a repository or a path for a repository.""" if isinstance(path_or_repo, BaseRepo): return path_or_repo return Repo(path_or_repo) @contextmanager def _noop_context_manager(obj): """Context manager that has the same api as closing but does nothing.""" yield obj def open_repo_closing(path_or_repo): """Open an argument that can be a repository or a path for a repository. returns a context manager that will close the repo on exit if the argument is a path, else does nothing if the argument is a repo. """ if isinstance(path_or_repo, BaseRepo): return _noop_context_manager(path_or_repo) return closing(Repo(path_or_repo)) def path_to_tree_path(repopath, path): """Convert a path to a path usable in e.g. an index. :param repo: Repository :param path: A path :return: A path formatted for use in e.g. an index """ os.path.relpath(path, repopath) if os.path.sep != '/': path = path.replace(os.path.sep, '/') return path.encode(sys.getfilesystemencoding()) def archive(repo, committish=None, outstream=default_bytes_out_stream, errstream=default_bytes_err_stream): """Create an archive. :param repo: Path of repository for which to generate an archive. :param committish: Commit SHA1 or ref to use :param outstream: Output stream (defaults to stdout) :param errstream: Error stream (defaults to stderr) """ if committish is None: committish = "HEAD" with open_repo_closing(repo) as repo_obj: c = repo_obj[committish] for chunk in tar_stream( repo_obj.object_store, repo_obj.object_store[c.tree], c.commit_time): outstream.write(chunk) def update_server_info(repo="."): """Update server info files for a repository. :param repo: path to the repository """ with open_repo_closing(repo) as r: server_update_server_info(r) def symbolic_ref(repo, ref_name, force=False): """Set git symbolic ref into HEAD. :param repo: path to the repository :param ref_name: short name of the new ref :param force: force settings without checking if it exists in refs/heads """ with open_repo_closing(repo) as repo_obj: ref_path = _make_branch_ref(ref_name) if not force and ref_path not in repo_obj.refs.keys(): raise ValueError('fatal: ref `%s` is not a ref' % ref_name) repo_obj.refs.set_symbolic_ref(b'HEAD', ref_path) def commit(repo=".", message=None, author=None, committer=None, encoding=None): """Create a new commit. :param repo: Path to repository :param message: Optional commit message :param author: Optional author name and email :param committer: Optional committer name and email :return: SHA1 of the new commit """ # FIXME: Support --all argument # FIXME: Support --signoff argument if getattr(message, 'encode', None): message = message.encode(encoding or DEFAULT_ENCODING) if getattr(author, 'encode', None): author = author.encode(encoding or DEFAULT_ENCODING) if getattr(committer, 'encode', None): committer = committer.encode(encoding or DEFAULT_ENCODING) with open_repo_closing(repo) as r: return r.do_commit( message=message, author=author, committer=committer, encoding=encoding) def commit_tree(repo, tree, message=None, author=None, committer=None): """Create a new commit object. :param repo: Path to repository :param tree: An existing tree object :param author: Optional author name and email :param committer: Optional committer name and email """ with open_repo_closing(repo) as r: return r.do_commit( message=message, tree=tree, committer=committer, author=author) def init(path=".", bare=False): """Create a new git repository. :param path: Path to repository. :param bare: Whether to create a bare repository. :return: A Repo instance """ if not os.path.exists(path): os.mkdir(path) if bare: return Repo.init_bare(path) else: return Repo.init(path) def clone(source, target=None, bare=False, checkout=None, errstream=default_bytes_err_stream, outstream=None, origin=b"origin", **kwargs): """Clone a local or remote git repository. :param source: Path or URL for source repository :param target: Path to target repository (optional) :param bare: Whether or not to create a bare repository :param checkout: Whether or not to check-out HEAD after cloning :param errstream: Optional stream to write progress to :param outstream: Optional stream to write progress to (deprecated) :param origin: Name of remote from the repository used to clone :return: The new repository """ if outstream is not None: import warnings warnings.warn( "outstream= has been deprecated in favour of errstream=.", DeprecationWarning, stacklevel=3) errstream = outstream if checkout is None: checkout = (not bare) if checkout and bare: raise ValueError("checkout and bare are incompatible") config = StackedConfig.default() client, host_path = get_transport_and_path(source, config=config, **kwargs) if target is None: target = host_path.split("/")[-1] if not os.path.exists(target): os.mkdir(target) if bare: r = Repo.init_bare(target) else: r = Repo.init(target) try: fetch_result = client.fetch( host_path, r, determine_wants=r.object_store.determine_wants_all, progress=errstream.write) ref_message = b"clone: from " + source.encode('utf-8') r.refs.import_refs( b'refs/remotes/' + origin, {n[len(b'refs/heads/'):]: v for (n, v) in fetch_result.refs.items() if n.startswith(b'refs/heads/')}, message=ref_message) r.refs.import_refs( b'refs/tags', {n[len(b'refs/tags/'):]: v for (n, v) in fetch_result.refs.items() if n.startswith(b'refs/tags/') and not n.endswith(ANNOTATED_TAG_SUFFIX)}, message=ref_message) target_config = r.get_config() if not isinstance(source, bytes): source = source.encode(DEFAULT_ENCODING) target_config.set((b'remote', origin), b'url', source) target_config.set( (b'remote', origin), b'fetch', b'+refs/heads/*:refs/remotes/' + origin + b'/*') target_config.write_to_path() # TODO(jelmer): Support symref capability, # https://github.com/jelmer/dulwich/issues/485 try: head = r[fetch_result.refs[b"HEAD"]] except KeyError: head = None else: r[b'HEAD'] = head.id if checkout and not bare and head is not None: errstream.write(b'Checking out ' + head.id + b'\n') r.reset_index(head.tree) except BaseException: r.close() raise return r def add(repo=".", paths=None): """Add files to the staging area. :param repo: Repository for the files :param paths: Paths to add. No value passed stages all modified files. :return: Tuple with set of added files and ignored files """ ignored = set() with open_repo_closing(repo) as r: ignore_manager = IgnoreFilterManager.from_repo(r) if not paths: paths = list( get_untracked_paths(os.getcwd(), r.path, r.open_index())) relpaths = [] if not isinstance(paths, list): paths = [paths] for p in paths: relpath = os.path.relpath(p, r.path) if relpath.startswith('../'): raise ValueError('path %r is not in repo' % relpath) # FIXME: Support patterns, directories. if ignore_manager.is_ignored(relpath): ignored.add(relpath) continue relpaths.append(relpath) r.stage(relpaths) return (relpaths, ignored) def remove(repo=".", paths=None, cached=False): """Remove files from the staging area. :param repo: Repository for the files :param paths: Paths to remove """ with open_repo_closing(repo) as r: index = r.open_index() for p in paths: full_path = os.path.abspath(p).encode(sys.getfilesystemencoding()) tree_path = path_to_tree_path(r.path, p) try: index_sha = index[tree_path].sha except KeyError: raise Exception('%s did not match any files' % p) if not cached: try: st = os.lstat(full_path) except OSError: pass else: try: blob = blob_from_path_and_stat(full_path, st) except IOError: pass else: try: committed_sha = tree_lookup_path( r.__getitem__, r[r.head()].tree, tree_path)[1] except KeyError: committed_sha = None if blob.id != index_sha and index_sha != committed_sha: raise Exception( 'file has staged content differing ' 'from both the file and head: %s' % p) if index_sha != committed_sha: raise Exception( 'file has staged changes: %s' % p) os.remove(full_path) del index[tree_path] index.write() rm = remove def commit_decode(commit, contents, default_encoding=DEFAULT_ENCODING): if commit.encoding is not None: return contents.decode(commit.encoding, "replace") return contents.decode(default_encoding, "replace") def print_commit(commit, decode, outstream=sys.stdout): """Write a human-readable commit log entry. :param commit: A `Commit` object :param outstream: A stream file to write to """ outstream.write("-" * 50 + "\n") outstream.write("commit: " + commit.id.decode('ascii') + "\n") if len(commit.parents) > 1: outstream.write( "merge: " + "...".join([c.decode('ascii') for c in commit.parents[1:]]) + "\n") outstream.write("Author: " + decode(commit.author) + "\n") if commit.author != commit.committer: outstream.write("Committer: " + decode(commit.committer) + "\n") time_tuple = time.gmtime(commit.author_time + commit.author_timezone) time_str = time.strftime("%a %b %d %Y %H:%M:%S", time_tuple) timezone_str = format_timezone(commit.author_timezone).decode('ascii') outstream.write("Date: " + time_str + " " + timezone_str + "\n") outstream.write("\n") outstream.write(decode(commit.message) + "\n") outstream.write("\n") def print_tag(tag, decode, outstream=sys.stdout): """Write a human-readable tag. :param tag: A `Tag` object :param decode: Function for decoding bytes to unicode string :param outstream: A stream to write to """ outstream.write("Tagger: " + decode(tag.tagger) + "\n") outstream.write("Date: " + decode(tag.tag_time) + "\n") outstream.write("\n") outstream.write(decode(tag.message) + "\n") outstream.write("\n") def show_blob(repo, blob, decode, outstream=sys.stdout): """Write a blob to a stream. :param repo: A `Repo` object :param blob: A `Blob` object :param decode: Function for decoding bytes to unicode string :param outstream: A stream file to write to """ outstream.write(decode(blob.data)) def show_commit(repo, commit, decode, outstream=sys.stdout): """Show a commit to a stream. :param repo: A `Repo` object :param commit: A `Commit` object :param decode: Function for decoding bytes to unicode string :param outstream: Stream to write to """ print_commit(commit, decode=decode, outstream=outstream) if commit.parents: parent_commit = repo[commit.parents[0]] base_tree = parent_commit.tree else: base_tree = None diffstream = BytesIO() write_tree_diff( diffstream, repo.object_store, base_tree, commit.tree) diffstream.seek(0) outstream.write( diffstream.getvalue().decode( commit.encoding or DEFAULT_ENCODING, 'replace')) def show_tree(repo, tree, decode, outstream=sys.stdout): """Print a tree to a stream. :param repo: A `Repo` object :param tree: A `Tree` object :param decode: Function for decoding bytes to unicode string :param outstream: Stream to write to """ for n in tree: outstream.write(decode(n) + "\n") def show_tag(repo, tag, decode, outstream=sys.stdout): """Print a tag to a stream. :param repo: A `Repo` object :param tag: A `Tag` object :param decode: Function for decoding bytes to unicode string :param outstream: Stream to write to """ print_tag(tag, decode, outstream) show_object(repo, repo[tag.object[1]], outstream) def show_object(repo, obj, decode, outstream): return { b"tree": show_tree, b"blob": show_blob, b"commit": show_commit, b"tag": show_tag, }[obj.type_name](repo, obj, decode, outstream) def print_name_status(changes): """Print a simple status summary, listing changed files. """ for change in changes: if not change: continue if isinstance(change, list): change = change[0] if change.type == CHANGE_ADD: path1 = change.new.path path2 = '' kind = 'A' elif change.type == CHANGE_DELETE: path1 = change.old.path path2 = '' kind = 'D' elif change.type == CHANGE_MODIFY: path1 = change.new.path path2 = '' kind = 'M' elif change.type in RENAME_CHANGE_TYPES: path1 = change.old.path path2 = change.new.path if change.type == CHANGE_RENAME: kind = 'R' elif change.type == CHANGE_COPY: kind = 'C' yield '%-8s%-20s%-20s' % (kind, path1, path2) def log(repo=".", paths=None, outstream=sys.stdout, max_entries=None, reverse=False, name_status=False): """Write commit logs. :param repo: Path to repository :param paths: Optional set of specific paths to print entries for :param outstream: Stream to write log output to :param reverse: Reverse order in which entries are printed :param name_status: Print name status :param max_entries: Optional maximum number of entries to display """ with open_repo_closing(repo) as r: walker = r.get_walker( max_entries=max_entries, paths=paths, reverse=reverse) for entry in walker: def decode(x): return commit_decode(entry.commit, x) print_commit(entry.commit, decode, outstream) if name_status: outstream.writelines( [l+'\n' for l in print_name_status(entry.changes())]) # TODO(jelmer): better default for encoding? def show(repo=".", objects=None, outstream=sys.stdout, default_encoding=DEFAULT_ENCODING): """Print the changes in a commit. :param repo: Path to repository :param objects: Objects to show (defaults to [HEAD]) :param outstream: Stream to write to :param default_encoding: Default encoding to use if none is set in the commit """ if objects is None: objects = ["HEAD"] if not isinstance(objects, list): objects = [objects] with open_repo_closing(repo) as r: for objectish in objects: o = parse_object(r, objectish) if isinstance(o, Commit): def decode(x): return commit_decode(o, x, default_encoding) else: def decode(x): return x.decode(default_encoding) show_object(r, o, decode, outstream) def diff_tree(repo, old_tree, new_tree, outstream=sys.stdout): """Compares the content and mode of blobs found via two tree objects. :param repo: Path to repository :param old_tree: Id of old tree :param new_tree: Id of new tree :param outstream: Stream to write to """ with open_repo_closing(repo) as r: write_tree_diff(outstream, r.object_store, old_tree, new_tree) def rev_list(repo, commits, outstream=sys.stdout): """Lists commit objects in reverse chronological order. :param repo: Path to repository :param commits: Commits over which to iterate :param outstream: Stream to write to """ with open_repo_closing(repo) as r: for entry in r.get_walker(include=[r[c].id for c in commits]): outstream.write(entry.commit.id + b"\n") def tag(*args, **kwargs): import warnings warnings.warn("tag has been deprecated in favour of tag_create.", DeprecationWarning) return tag_create(*args, **kwargs) def tag_create( repo, tag, author=None, message=None, annotated=False, objectish="HEAD", tag_time=None, tag_timezone=None): """Creates a tag in git via dulwich calls: :param repo: Path to repository :param tag: tag string :param author: tag author (optional, if annotated is set) :param message: tag message (optional) :param annotated: whether to create an annotated tag :param objectish: object the tag should point at, defaults to HEAD :param tag_time: Optional time for annotated tag :param tag_timezone: Optional timezone for annotated tag """ with open_repo_closing(repo) as r: object = parse_object(r, objectish) if annotated: # Create the tag object tag_obj = Tag() if author is None: # TODO(jelmer): Don't use repo private method. author = r._get_user_identity() tag_obj.tagger = author tag_obj.message = message tag_obj.name = tag tag_obj.object = (type(object), object.id) if tag_time is None: tag_time = int(time.time()) tag_obj.tag_time = tag_time if tag_timezone is None: # TODO(jelmer) Use current user timezone rather than UTC tag_timezone = 0 elif isinstance(tag_timezone, str): tag_timezone = parse_timezone(tag_timezone) tag_obj.tag_timezone = tag_timezone r.object_store.add_object(tag_obj) tag_id = tag_obj.id else: tag_id = object.id r.refs[_make_tag_ref(tag)] = tag_id def list_tags(*args, **kwargs): import warnings warnings.warn("list_tags has been deprecated in favour of tag_list.", DeprecationWarning) return tag_list(*args, **kwargs) def tag_list(repo, outstream=sys.stdout): """List all tags. :param repo: Path to repository :param outstream: Stream to write tags to """ with open_repo_closing(repo) as r: tags = sorted(r.refs.as_dict(b"refs/tags")) return tags def tag_delete(repo, name): """Remove a tag. :param repo: Path to repository :param name: Name of tag to remove """ with open_repo_closing(repo) as r: if isinstance(name, bytes): names = [name] elif isinstance(name, list): names = name else: raise TypeError("Unexpected tag name type %r" % name) for name in names: del r.refs[_make_tag_ref(name)] def reset(repo, mode, treeish="HEAD"): """Reset current HEAD to the specified state. :param repo: Path to repository :param mode: Mode ("hard", "soft", "mixed") :param treeish: Treeish to reset to """ if mode != "hard": raise ValueError("hard is the only mode currently supported") with open_repo_closing(repo) as r: tree = parse_tree(r, treeish) r.reset_index(tree.id) def push(repo, remote_location, refspecs, outstream=default_bytes_out_stream, errstream=default_bytes_err_stream, **kwargs): """Remote push with dulwich via dulwich.client :param repo: Path to repository :param remote_location: Location of the remote :param refspecs: Refs to push to remote :param outstream: A stream file to write output :param errstream: A stream file to write errors """ # Open the repo with open_repo_closing(repo) as r: # Get the client and path client, path = get_transport_and_path( remote_location, config=r.get_config_stack(), **kwargs) selected_refs = [] def update_refs(refs): selected_refs.extend(parse_reftuples(r.refs, refs, refspecs)) new_refs = {} # TODO: Handle selected_refs == {None: None} for (lh, rh, force) in selected_refs: if lh is None: new_refs[rh] = ZERO_SHA else: new_refs[rh] = r.refs[lh] return new_refs err_encoding = getattr(errstream, 'encoding', None) or DEFAULT_ENCODING remote_location_bytes = client.get_url(path).encode(err_encoding) try: client.send_pack( path, update_refs, generate_pack_data=r.object_store.generate_pack_data, progress=errstream.write) errstream.write( b"Push to " + remote_location_bytes + b" successful.\n") except (UpdateRefsError, SendPackError) as e: errstream.write(b"Push to " + remote_location_bytes + b" failed -> " + e.message.encode(err_encoding) + b"\n") def pull(repo, remote_location=None, refspecs=None, outstream=default_bytes_out_stream, errstream=default_bytes_err_stream, **kwargs): """Pull from remote via dulwich.client :param repo: Path to repository :param remote_location: Location of the remote :param refspec: refspecs to fetch :param outstream: A stream file to write to output :param errstream: A stream file to write to errors """ # Open the repo with open_repo_closing(repo) as r: if remote_location is None: # TODO(jelmer): Lookup 'remote' for current branch in config raise NotImplementedError( "looking up remote from branch config not supported yet") if refspecs is None: refspecs = [b"HEAD"] selected_refs = [] def determine_wants(remote_refs): selected_refs.extend( parse_reftuples(remote_refs, r.refs, refspecs)) return [remote_refs[lh] for (lh, rh, force) in selected_refs] client, path = get_transport_and_path( remote_location, config=r.get_config_stack(), **kwargs) fetch_result = client.fetch( path, r, progress=errstream.write, determine_wants=determine_wants) for (lh, rh, force) in selected_refs: r.refs[rh] = fetch_result.refs[lh] if selected_refs: r[b'HEAD'] = fetch_result.refs[selected_refs[0][1]] # Perform 'git checkout .' - syncs staged changes tree = r[b"HEAD"].tree r.reset_index(tree=tree) def status(repo=".", ignored=False): """Returns staged, unstaged, and untracked changes relative to the HEAD. :param repo: Path to repository or repository object :param ignored: Whether to include ignoed files in `untracked` :return: GitStatus tuple, staged - list of staged paths (diff index/HEAD) unstaged - list of unstaged paths (diff index/working-tree) untracked - list of untracked, un-ignored & non-.git paths """ with open_repo_closing(repo) as r: # 1. Get status of staged tracked_changes = get_tree_changes(r) # 2. Get status of unstaged index = r.open_index() unstaged_changes = list(get_unstaged_changes(index, r.path)) ignore_manager = IgnoreFilterManager.from_repo(r) untracked_paths = get_untracked_paths(r.path, r.path, index) if ignored: untracked_changes = list(untracked_paths) else: untracked_changes = [ p for p in untracked_paths if not ignore_manager.is_ignored(p)] return GitStatus(tracked_changes, unstaged_changes, untracked_changes) def get_untracked_paths(frompath, basepath, index): """Get untracked paths. ;param frompath: Path to walk :param basepath: Path to compare to :param index: Index to check against """ # If nothing is specified, add all non-ignored files. for dirpath, dirnames, filenames in os.walk(frompath): # Skip .git and below. if '.git' in dirnames: dirnames.remove('.git') if dirpath != basepath: continue if '.git' in filenames: filenames.remove('.git') if dirpath != basepath: continue for filename in filenames: ap = os.path.join(dirpath, filename) ip = path_to_tree_path(basepath, ap) if ip not in index: yield os.path.relpath(ap, frompath) def get_tree_changes(repo): """Return add/delete/modify changes to tree by comparing index to HEAD. :param repo: repo path or object :return: dict with lists for each type of change """ with open_repo_closing(repo) as r: index = r.open_index() # Compares the Index to the HEAD & determines changes # Iterate through the changes and report add/delete/modify # TODO: call out to dulwich.diff_tree somehow. tracked_changes = { 'add': [], 'delete': [], 'modify': [], } try: tree_id = r[b'HEAD'].tree except KeyError: tree_id = None for change in index.changes_from_tree(r.object_store, tree_id): if not change[0][0]: tracked_changes['add'].append(change[0][1]) elif not change[0][1]: tracked_changes['delete'].append(change[0][0]) elif change[0][0] == change[0][1]: tracked_changes['modify'].append(change[0][0]) else: raise AssertionError('git mv ops not yet supported') return tracked_changes def daemon(path=".", address=None, port=None): """Run a daemon serving Git requests over TCP/IP. :param path: Path to the directory to serve. :param address: Optional address to listen on (defaults to ::) :param port: Optional port to listen on (defaults to TCP_GIT_PORT) """ # TODO(jelmer): Support git-daemon-export-ok and --export-all. backend = FileSystemBackend(path) server = TCPGitServer(backend, address, port) server.serve_forever() def web_daemon(path=".", address=None, port=None): """Run a daemon serving Git requests over HTTP. :param path: Path to the directory to serve :param address: Optional address to listen on (defaults to ::) :param port: Optional port to listen on (defaults to 80) """ from dulwich.web import ( make_wsgi_chain, make_server, WSGIRequestHandlerLogger, WSGIServerLogger) backend = FileSystemBackend(path) app = make_wsgi_chain(backend) server = make_server(address, port, app, handler_class=WSGIRequestHandlerLogger, server_class=WSGIServerLogger) server.serve_forever() def upload_pack(path=".", inf=None, outf=None): """Upload a pack file after negotiating its contents using smart protocol. :param path: Path to the repository :param inf: Input stream to communicate with client :param outf: Output stream to communicate with client """ if outf is None: outf = getattr(sys.stdout, 'buffer', sys.stdout) if inf is None: inf = getattr(sys.stdin, 'buffer', sys.stdin) path = os.path.expanduser(path) backend = FileSystemBackend(path) def send_fn(data): outf.write(data) outf.flush() proto = Protocol(inf.read, send_fn) handler = UploadPackHandler(backend, [path], proto) # FIXME: Catch exceptions and write a single-line summary to outf. handler.handle() return 0 def receive_pack(path=".", inf=None, outf=None): """Receive a pack file after negotiating its contents using smart protocol. :param path: Path to the repository :param inf: Input stream to communicate with client :param outf: Output stream to communicate with client """ if outf is None: outf = getattr(sys.stdout, 'buffer', sys.stdout) if inf is None: inf = getattr(sys.stdin, 'buffer', sys.stdin) path = os.path.expanduser(path) backend = FileSystemBackend(path) def send_fn(data): outf.write(data) outf.flush() proto = Protocol(inf.read, send_fn) handler = ReceivePackHandler(backend, [path], proto) # FIXME: Catch exceptions and write a single-line summary to outf. handler.handle() return 0 def _make_branch_ref(name): if getattr(name, 'encode', None): name = name.encode(DEFAULT_ENCODING) return b"refs/heads/" + name def _make_tag_ref(name): if getattr(name, 'encode', None): name = name.encode(DEFAULT_ENCODING) return b"refs/tags/" + name def branch_delete(repo, name): """Delete a branch. :param repo: Path to the repository :param name: Name of the branch """ with open_repo_closing(repo) as r: if isinstance(name, list): names = name else: names = [name] for name in names: del r.refs[_make_branch_ref(name)] def branch_create(repo, name, objectish=None, force=False): """Create a branch. :param repo: Path to the repository :param name: Name of the new branch :param objectish: Target object to point new branch at (defaults to HEAD) :param force: Force creation of branch, even if it already exists """ with open_repo_closing(repo) as r: if objectish is None: objectish = "HEAD" object = parse_object(r, objectish) refname = _make_branch_ref(name) ref_message = b"branch: Created from " + objectish.encode('utf-8') if force: r.refs.set_if_equals(refname, None, object.id, message=ref_message) else: if not r.refs.add_if_new(refname, object.id, message=ref_message): raise KeyError("Branch with name %s already exists." % name) def branch_list(repo): """List all branches. :param repo: Path to the repository """ with open_repo_closing(repo) as r: return r.refs.keys(base=b"refs/heads/") -def fetch(repo, remote_location, outstream=sys.stdout, +def fetch(repo, remote_location, remote_name=b'origin', outstream=sys.stdout, errstream=default_bytes_err_stream, **kwargs): """Fetch objects from a remote server. :param repo: Path to the repository :param remote_location: String identifying a remote server + :param remote_name: Name for remote server :param outstream: Output stream (defaults to stdout) :param errstream: Error stream (defaults to stderr) :return: Dictionary with refs on the remote """ with open_repo_closing(repo) as r: client, path = get_transport_and_path( remote_location, config=r.get_config_stack(), **kwargs) - remote_refs = client.fetch(path, r, progress=errstream.write) - return remote_refs + fetch_result = client.fetch(path, r, progress=errstream.write) + ref_name = b'refs/remotes/' + remote_name + r.refs.import_refs(ref_name, strip_peeled_refs(fetch_result.refs)) + return fetch_result.refs def ls_remote(remote, config=None, **kwargs): """List the refs in a remote. :param remote: Remote repository location :param config: Configuration to use :return: Dictionary with remote refs """ if config is None: config = StackedConfig.default() client, host_path = get_transport_and_path(remote, config=config, **kwargs) return client.get_refs(host_path) def repack(repo): """Repack loose files in a repository. Currently this only packs loose objects. :param repo: Path to the repository """ with open_repo_closing(repo) as r: r.object_store.pack_loose_objects() def pack_objects(repo, object_ids, packf, idxf, delta_window_size=None): """Pack objects into a file. :param repo: Path to the repository :param object_ids: List of object ids to write :param packf: File-like object to write to :param idxf: File-like object to write to (can be None) """ with open_repo_closing(repo) as r: entries, data_sum = write_pack_objects( packf, r.object_store.iter_shas((oid, None) for oid in object_ids), delta_window_size=delta_window_size) if idxf is not None: entries = sorted([(k, v[0], v[1]) for (k, v) in entries.items()]) write_pack_index(idxf, entries, data_sum) def ls_tree(repo, treeish=b"HEAD", outstream=sys.stdout, recursive=False, name_only=False): """List contents of a tree. :param repo: Path to the repository :param tree_ish: Tree id to list :param outstream: Output stream (defaults to stdout) :param recursive: Whether to recursively list files :param name_only: Only print item name """ def list_tree(store, treeid, base): for (name, mode, sha) in store[treeid].iteritems(): if base: name = posixpath.join(base, name) if name_only: outstream.write(name + b"\n") else: outstream.write(pretty_format_tree_entry(name, mode, sha)) if stat.S_ISDIR(mode): list_tree(store, sha, name) with open_repo_closing(repo) as r: tree = parse_tree(r, treeish) list_tree(r.object_store, tree.id, "") def remote_add(repo, name, url): """Add a remote. :param repo: Path to the repository :param name: Remote name :param url: Remote URL """ if not isinstance(name, bytes): name = name.encode(DEFAULT_ENCODING) if not isinstance(url, bytes): url = url.encode(DEFAULT_ENCODING) with open_repo_closing(repo) as r: c = r.get_config() section = (b'remote', name) if c.has_section(section): raise RemoteExists(section) c.set(section, b"url", url) c.write_to_path() def check_ignore(repo, paths, no_index=False): """Debug gitignore files. :param repo: Path to the repository :param paths: List of paths to check for :param no_index: Don't check index :return: List of ignored files """ with open_repo_closing(repo) as r: index = r.open_index() ignore_manager = IgnoreFilterManager.from_repo(r) for path in paths: if os.path.isabs(path): path = os.path.relpath(path, r.path) if not no_index and path_to_tree_path(r.path, path) in index: continue if ignore_manager.is_ignored(path): yield path def update_head(repo, target, detached=False, new_branch=None): """Update HEAD to point at a new branch/commit. Note that this does not actually update the working tree. :param repo: Path to the repository :param detach: Create a detached head :param target: Branch or committish to switch to :param new_branch: New branch to create """ with open_repo_closing(repo) as r: if new_branch is not None: to_set = _make_branch_ref(new_branch) else: to_set = b"HEAD" if detached: # TODO(jelmer): Provide some way so that the actual ref gets # updated rather than what it points to, so the delete isn't # necessary. del r.refs[to_set] r.refs[to_set] = parse_commit(r, target).id else: r.refs.set_symbolic_ref(to_set, parse_ref(r, target)) if new_branch is not None: r.refs.set_symbolic_ref(b"HEAD", to_set) diff --git a/dulwich/refs.py b/dulwich/refs.py index ef5bdac0..4389031b 100644 --- a/dulwich/refs.py +++ b/dulwich/refs.py @@ -1,884 +1,890 @@ # 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 ~^:?*[') ANNOTATED_TAG_SUFFIX = b'^{}' def parse_symref_value(contents): """Parse a symref value. :param contents: Contents to parse :return: Destination """ if contents.startswith(SYMREF): return contents[len(SYMREF):].rstrip(b'\r\n') raise ValueError(contents) 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 __init__(self, logger=None): self._logger = logger def _log(self, ref, old_sha, new_sha, committer=None, timestamp=None, timezone=None, message=None): if self._logger is None: return if message is None: return self._logger(ref, old_sha, new_sha, committer, timestamp, timezone, message) def set_symbolic_ref(self, name, other, committer=None, timestamp=None, timezone=None, message=None): """Make a ref point at another ref. :param name: Name of the ref to set :param other: Name of the ref to point at :param message: Optional message """ 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, committer=None, timestamp=None, timezone=None, message=None): for name, value in other.items(): self.set_if_equals(b'/'.join((base, name)), None, value, message=message) 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, committer=None, timestamp=None, timezone=None, message=None): """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. :param message: Message for reflog :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. :param name: Ref name :param ref: Ref value :param message: Message for reflog """ 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, committer=None, timestamp=None, timezone=None, message=None): """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. :param message: Message for reflog :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) def get_symrefs(self): """Get a dict with all symrefs in this container. :return: Dictionary mapping source ref to target ref """ ret = {} for src in self.allkeys(): try: dst = parse_symref_value(self.read_ref(src)) except ValueError: pass else: ret[src] = dst return ret 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, logger=None): super(DictRefsContainer, self).__init__(logger=logger) 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, committer=None, timestamp=None, timezone=None, message=None): old = self.follow(name)[-1] self._refs[name] = SYMREF + other self._log(name, old, old, committer=committer, timestamp=timestamp, timezone=timezone, message=message) def set_if_equals(self, name, old_ref, new_ref, committer=None, timestamp=None, timezone=None, message=None): 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) old = self._refs.get(realname) self._refs[realname] = new_ref self._log(realname, old, new_ref, committer=committer, timestamp=timestamp, timezone=timezone, message=message) return True def add_if_new(self, name, ref, committer=None, timestamp=None, timezone=None, message=None): if name in self._refs: return False self._refs[name] = ref self._log(name, None, ref, committer=committer, timestamp=timestamp, timezone=timezone, message=message) return True def remove_if_equals(self, name, old_ref, committer=None, timestamp=None, timezone=None, message=None): if old_ref is not None and self._refs.get(name, ZERO_SHA) != old_ref: return False try: old = self._refs.pop(name) except KeyError: pass else: self._log(name, old, None, committer=committer, timestamp=timestamp, timezone=timezone, message=message) 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(ANNOTATED_TAG_SUFFIX): 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, worktree_path=None, logger=None): super(DiskRefsContainer, self).__init__(logger=logger) self.path = path self.worktree_path = worktree_path or 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) # TODO: as the 'HEAD' reference is working tree specific, it # should actually not be a part of RefsContainer if name == '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 in (errno.ENOENT, errno.EISDIR): 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, committer=None, timestamp=None, timezone=None, message=None): """Make a ref point at another ref. :param name: Name of the ref to set :param other: Name of the ref to point at :param message: Optional message to describe the change """ 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 else: sha = self.follow(name)[-1] self._log(name, sha, sha, committer=committer, timestamp=timestamp, timezone=timezone, message=message) finally: f.close() def set_if_equals(self, name, old_ref, new_ref, committer=None, timestamp=None, timezone=None, message=None): """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. :param message: Set message for reflog :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 self._log(realname, old_ref, new_ref, committer=committer, timestamp=timestamp, timezone=timezone, message=message) return True def add_if_new(self, name, ref, committer=None, timestamp=None, timezone=None, message=None): """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. :param message: Optional message for reflog :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 else: self._log(name, None, ref, committer=committer, timestamp=timestamp, timezone=timezone, message=message) return True def remove_if_equals(self, name, old_ref, committer=None, timestamp=None, timezone=None, message=None): """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. :param message: Optional message :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) self._log(name, old_ref, None, committer=committer, timestamp=timestamp, timezone=timezone, message=message) 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\r').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 line in f: if line[0] == b'#': continue line = line.rstrip(b'\r\n') if line.startswith(b'^'): if not last: raise PackedRefsException("unexpected peeled ref line") if not valid_hexsha(line[1:]): raise PackedRefsException("Invalid hex sha %r" % line[1:]) sha, name = _split_ref_line(last) last = None yield (sha, name, line[1:]) else: if last: sha, name = _split_ref_line(last) yield (sha, name, None) last = line 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 + ANNOTATED_TAG_SUFFIX + b'\n' def is_local_branch(x): return x.startswith(b'refs/heads/') + + +def strip_peeled_refs(refs): + """Remove all peeled refs""" + return {ref: sha for (ref, sha) in refs.items() + if not ref.endswith(ANNOTATED_TAG_SUFFIX)} diff --git a/dulwich/tests/test_porcelain.py b/dulwich/tests/test_porcelain.py index f6a243d1..0e77c250 100644 --- a/dulwich/tests/test_porcelain.py +++ b/dulwich/tests/test_porcelain.py @@ -1,1232 +1,1275 @@ # test_porcelain.py -- porcelain tests # Copyright (C) 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. # """Tests for dulwich.porcelain.""" 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 ( NoIndexPresent, Repo, ) from dulwich.tests import ( TestCase, ) from dulwich.tests.utils import ( build_commit_graph, make_commit, make_object, ) class PorcelainTestCase(TestCase): def setUp(self): super(PorcelainTestCase, self).setUp() self.test_dir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, self.test_dir) self.repo = Repo.init(os.path.join(self.test_dir, 'repo'), mkdir=True) self.addCleanup(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) def test_unicode(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="Some message", author="Joe ", committer="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 self.repo.refs[b"refs/tags/foo"] = 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) target_repo = Repo(target_path) self.assertEqual(0, len(target_repo.open_index())) self.assertEqual(c3.id, target_repo.refs[b'refs/tags/foo']) self.assertTrue(b'f1' not in os.listdir(target_path)) self.assertTrue(b'f2' not in os.listdir(target_path)) c = r.get_config() encoded_path = self.repo.path if not isinstance(encoded_path, bytes): encoded_path = encoded_path.encode('utf-8') self.assertEqual(encoded_path, c.get((b'remote', b'origin'), b'url')) self.assertEqual( b'+refs/heads/*:refs/remotes/origin/*', c.get((b'remote', b'origin'), b'fetch')) 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 porcelain.clone(self.repo.path, target_path, checkout=True, errstream=errstream) as r: self.assertEqual(r.path, target_path) with 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) with porcelain.clone( self.repo.path, target_path, bare=True, errstream=errstream) as r: self.assertEqual(r.path, target_path) with Repo(target_path) as r: r.head() self.assertRaises(NoIndexPresent, r.open_index) 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 self.repo.refs[b"HEAD"] = 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) def test_no_head_no_checkout(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() self.addCleanup(shutil.rmtree, target_path) errstream = BytesIO() r = porcelain.clone( self.repo.path, target_path, checkout=True, errstream=errstream) r.close() 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 fullpath = os.path.join(self.repo.path, 'blah') with open(fullpath, 'w') as f: f.write("\n") porcelain.add(repo=self.repo.path, paths=[fullpath]) 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") cwd = os.getcwd() try: os.chdir(self.repo.path) porcelain.add(self.repo.path) finally: os.chdir(cwd) # 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_default_paths_subdir(self): os.mkdir(os.path.join(self.repo.path, 'foo')) with open(os.path.join(self.repo.path, 'blah'), 'w') as f: f.write("\n") with open(os.path.join(self.repo.path, 'foo', 'blie'), 'w') as f: f.write("\n") cwd = os.getcwd() try: os.chdir(os.path.join(self.repo.path, 'foo')) porcelain.add(repo=self.repo.path) porcelain.commit(repo=self.repo.path, message=b'test', author=b'test ', committer=b'test ') finally: os.chdir(cwd) index = self.repo.open_index() self.assertEqual(sorted(index), [b'foo/blie']) def test_add_file(self): fullpath = os.path.join(self.repo.path, 'foo') with open(fullpath, 'w') as f: f.write("BAR") porcelain.add(self.repo.path, paths=[fullpath]) self.assertIn(b"foo", self.repo.open_index()) def test_add_ignored(self): with open(os.path.join(self.repo.path, '.gitignore'), 'w') as f: f.write("foo") with open(os.path.join(self.repo.path, 'foo'), 'w') as f: f.write("BAR") with open(os.path.join(self.repo.path, 'bar'), 'w') as f: f.write("BAR") (added, ignored) = porcelain.add(self.repo.path, paths=[ os.path.join(self.repo.path, "foo"), os.path.join(self.repo.path, "bar")]) self.assertIn(b"bar", self.repo.open_index()) self.assertEqual(set(['bar']), set(added)) self.assertEqual(set(['foo']), ignored) def test_add_file_absolute_path(self): # Absolute paths are (not yet) supported with open(os.path.join(self.repo.path, 'foo'), 'w') as f: f.write("BAR") porcelain.add(self.repo, paths=[os.path.join(self.repo.path, "foo")]) self.assertIn(b"foo", self.repo.open_index()) def test_add_not_in_repo(self): with open(os.path.join(self.test_dir, 'foo'), 'w') as f: f.write("BAR") self.assertRaises( ValueError, porcelain.add, self.repo, paths=[os.path.join(self.test_dir, "foo")]) self.assertRaises( ValueError, porcelain.add, self.repo, paths=["../foo"]) self.assertEqual([], list(self.repo.open_index())) class RemoveTests(PorcelainTestCase): def test_remove_file(self): fullpath = os.path.join(self.repo.path, 'foo') with open(fullpath, 'w') as f: f.write("BAR") porcelain.add(self.repo.path, paths=[fullpath]) porcelain.commit(repo=self.repo, message=b'test', author=b'test ', committer=b'test ') self.assertTrue(os.path.exists(os.path.join(self.repo.path, 'foo'))) cwd = os.getcwd() try: os.chdir(self.repo.path) porcelain.remove(self.repo.path, paths=["foo"]) finally: os.chdir(cwd) self.assertFalse(os.path.exists(os.path.join(self.repo.path, 'foo'))) def test_remove_file_staged(self): fullpath = os.path.join(self.repo.path, 'foo') with open(fullpath, 'w') as f: f.write("BAR") cwd = os.getcwd() try: os.chdir(self.repo.path) porcelain.add(self.repo.path, paths=[fullpath]) self.assertRaises(Exception, porcelain.rm, self.repo.path, paths=["foo"]) finally: os.chdir(cwd) 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") def test_commit_no_parent(self): a = Blob.from_string(b"The Foo\n") ta = Tree() ta.add(b"somename", 0o100644, a.id) ca = make_commit(tree=ta.id) self.repo.object_store.add_objects([(a, None), (ta, None), (ca, None)]) outstream = StringIO() porcelain.show(self.repo.path, objects=[ca.id], outstream=outstream) self.assertMultiLineEqual(outstream.getvalue(), """\ -------------------------------------------------- commit: 344da06c1bb85901270b3e8875c988a027ec087d Author: Test Author Committer: Test Committer Date: Fri Jan 01 2010 00:00:00 +0000 Test message. diff --git /dev/null b/somename new mode 100644 index 0000000..ea5c7bf 100644 --- /dev/null +++ b/somename @@ -0,0 +1 @@ +The Foo """) def test_commit_with_change(self): a = Blob.from_string(b"The Foo\n") ta = Tree() ta.add(b"somename", 0o100644, a.id) ca = make_commit(tree=ta.id) b = Blob.from_string(b"The Bar\n") tb = Tree() tb.add(b"somename", 0o100644, b.id) cb = make_commit(tree=tb.id, parents=[ca.id]) self.repo.object_store.add_objects( [(a, None), (b, None), (ta, None), (tb, None), (ca, None), (cb, None)]) outstream = StringIO() porcelain.show(self.repo.path, objects=[cb.id], outstream=outstream) self.assertMultiLineEqual(outstream.getvalue(), """\ -------------------------------------------------- commit: 2c6b6c9cb72c130956657e1fdae58e5b103744fa Author: Test Author Committer: Test Committer Date: Fri Jan 01 2010 00:00:00 +0000 Test message. diff --git a/somename b/somename index ea5c7bf..fd38bcb 100644 --- a/somename +++ b/somename @@ -1 +1 @@ -The Foo +The Bar """) 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()]) def test_unannotated_unicode(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, "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): fullpath = os.path.join(self.repo.path, 'foo') with open(fullpath, 'w') as f: f.write("BAR") porcelain.add(self.repo.path, paths=[fullpath]) 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) def test_hard_commit(self): fullpath = os.path.join(self.repo.path, 'foo') with open(fullpath, 'w') as f: f.write("BAR") porcelain.add(self.repo.path, paths=[fullpath]) sha = porcelain.commit(self.repo.path, message=b"Some message", committer=b"Jane ", author=b"John ") with open(fullpath, 'wb') as f: f.write(b"BAZ") porcelain.add(self.repo.path, paths=[fullpath]) porcelain.commit(self.repo.path, message=b"Some other message", committer=b"Jane ", author=b"John ") porcelain.reset(self.repo, "hard", sha) index = self.repo.open_index() changes = list(tree_changes(self.repo, index.commit(self.repo.object_store), self.repo[sha].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'author ', committer=b'committer ') # 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=[fullpath]) porcelain.commit(repo=clone_path, message=b'push', author=b'author ', committer=b'committer ') # 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 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_path].id) # 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'author ', committer=b'committer ') # 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) self.assertEqual({ b'HEAD': new_id, b'refs/heads/master': new_id, }, self.repo.get_refs()) class PullTests(PorcelainTestCase): def setUp(self): super(PullTests, self).setUp() # create a file for initial commit handle, fullpath = tempfile.mkstemp(dir=self.repo.path) os.close(handle) porcelain.add(repo=self.repo.path, paths=fullpath) porcelain.commit(repo=self.repo.path, message=b'test', author=b'test ', committer=b'test ') # Setup target repo self.target_path = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, self.target_path) target_repo = porcelain.clone(self.repo.path, target=self.target_path, errstream=BytesIO()) target_repo.close() # create a second file to be pushed handle, fullpath = tempfile.mkstemp(dir=self.repo.path) os.close(handle) porcelain.add(repo=self.repo.path, paths=fullpath) 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) def test_simple(self): outstream = BytesIO() errstream = BytesIO() # Pull changes into the cloned repo porcelain.pull(self.target_path, self.repo.path, b'refs/heads/master', outstream=outstream, errstream=errstream) # Check the target repo for pushed changes with Repo(self.target_path) as r: self.assertEqual(r[b'HEAD'].id, self.repo[b'HEAD'].id) def test_no_refspec(self): outstream = BytesIO() errstream = BytesIO() # Pull changes into the cloned repo porcelain.pull(self.target_path, self.repo.path, outstream=outstream, errstream=errstream) # Check the target repo for pushed changes with Repo(self.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=[fullpath]) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'author ', committer=b'committer ') # 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=fullpath) 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' fullpath = os.path.join(self.repo.path, filename) with open(fullpath, 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=fullpath) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'author ', committer=b'committer ') 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=fullpath) 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=fullpath) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'author ', committer=b'committer ') with open(fullpath, 'w') as f: f.write('otherstuff') porcelain.add(repo=self.repo.path, paths=fullpath) 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' fullpath = os.path.join(self.repo.path, filename) with open(fullpath, 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=fullpath) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'author ', committer=b'committer ') cwd = os.getcwd() try: os.chdir(self.repo.path) porcelain.remove(repo=self.repo.path, paths=[filename]) finally: os.chdir(cwd) 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) def test_get_untracked_paths(self): with open(os.path.join(self.repo.path, '.gitignore'), 'w') as f: f.write('ignored\n') with open(os.path.join(self.repo.path, 'ignored'), 'w') as f: f.write('blah\n') with open(os.path.join(self.repo.path, 'notignored'), 'w') as f: f.write('blah\n') self.assertEqual( set(['ignored', 'notignored', '.gitignore']), set(porcelain.get_untracked_paths(self.repo.path, self.repo.path, self.repo.open_index()))) self.assertEqual(set(['.gitignore', 'notignored']), set(porcelain.status(self.repo).untracked)) self.assertEqual(set(['.gitignore', 'notignored', 'ignored']), set(porcelain.status(self.repo, ignored=True) .untracked)) def test_get_untracked_paths_nested(self): with open(os.path.join(self.repo.path, 'notignored'), 'w') as f: f.write('blah\n') subrepo = Repo.init(os.path.join(self.repo.path, 'nested'), mkdir=True) with open(os.path.join(subrepo.path, 'another'), 'w') as f: f.write('foo\n') self.assertEqual( set(['notignored']), set(porcelain.get_untracked_paths(self.repo.path, self.repo.path, self.repo.open_index()))) self.assertEqual( set(['another']), set(porcelain.get_untracked_paths(subrepo.path, subrepo.path, subrepo.open_index()))) # 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' fullpath = os.path.join(self.repo.path, filename) with open(fullpath, 'w') as f: f.write('stuff') porcelain.add(repo=self.repo.path, paths=fullpath) self.repo.do_commit(message=b'test status', author=b'author ', committer=b'committer ', 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'0091319b56ce3aee2d489f759736a79cc552c9bb86d9 HEAD\x00 report-status ' # noqa: E501 b'delete-refs quiet ofs-delta side-band-64k ' b'no-done symref=HEAD:refs/heads/master', b'003f319b56ce3aee2d489f759736a79cc552c9bb86d9 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)) def test_simple_unicode(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo[b"HEAD"] = c1.id porcelain.branch_create(self.repo, 'foo') self.assertTrue(b"foo" in porcelain.branch_list(self.repo)) porcelain.branch_delete(self.repo, '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) porcelain.add(repo=self.repo.path, paths=fullpath) 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) porcelain.add(repo=self.repo.path, paths=fullpath) 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) + porcelain.fetch(target_path, self.repo.path, + outstream=outstream, errstream=errstream) # Check the target repo for pushed changes with Repo(target_path) as r: self.assertTrue(self.repo[b'HEAD'].id in r) + def test_with_remote_name(self): + remote_name = b'origin' + outstream = BytesIO() + errstream = BytesIO() + + # create a file for initial commit + handle, fullpath = tempfile.mkstemp(dir=self.repo.path) + os.close(handle) + porcelain.add(repo=self.repo.path, paths=fullpath) + 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) + + # Capture current refs + target_refs = target_repo.get_refs() + + # create a second file to be pushed + handle, fullpath = tempfile.mkstemp(dir=self.repo.path) + os.close(handle) + porcelain.add(repo=self.repo.path, paths=fullpath) + 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, remote_name=remote_name, + outstream=outstream, errstream=errstream) + + # Check the target repo for pushed changes, as well as updates + # for the refs + with Repo(target_path) as r: + self.assertTrue(self.repo[b'HEAD'].id in r) + self.assertNotEqual(self.repo.get_refs(), target_refs) + 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) porcelain.add(repo=self.repo.path, paths=fullpath) porcelain.repack(self.repo) class LsTreeTests(PorcelainTestCase): def test_empty(self): porcelain.commit(repo=self.repo.path, message=b'test status', author=b'author ', committer=b'committer ') f = StringIO() porcelain.ls_tree(self.repo, b"HEAD", outstream=f) self.assertEqual(f.getvalue(), "") def test_simple(self): # 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=[fullpath]) porcelain.commit(repo=self.repo.path, message=b'test status', author=b'author ', committer=b'committer ') f = StringIO() porcelain.ls_tree(self.repo, b"HEAD", outstream=f) self.assertEqual( f.getvalue(), '100644 blob 8b82634d7eae019850bb883f06abf428c58bc9aa\tfoo\n') class LsRemoteTests(PorcelainTestCase): def test_empty(self): self.assertEqual({}, porcelain.ls_remote(self.repo.path)) def test_some(self): cid = porcelain.commit(repo=self.repo.path, message=b'test status', author=b'author ', committer=b'committer ') self.assertEqual({ b'refs/heads/master': cid, b'HEAD': cid}, porcelain.ls_remote(self.repo.path)) class RemoteAddTests(PorcelainTestCase): def test_new(self): porcelain.remote_add( self.repo, 'jelmer', 'git://jelmer.uk/code/dulwich') c = self.repo.get_config() self.assertEqual( c.get((b'remote', b'jelmer'), b'url'), b'git://jelmer.uk/code/dulwich') def test_exists(self): porcelain.remote_add( self.repo, 'jelmer', 'git://jelmer.uk/code/dulwich') self.assertRaises(porcelain.RemoteExists, porcelain.remote_add, self.repo, 'jelmer', 'git://jelmer.uk/code/dulwich') class CheckIgnoreTests(PorcelainTestCase): def test_check_ignored(self): with open(os.path.join(self.repo.path, '.gitignore'), 'w') as f: f.write("foo") with open(os.path.join(self.repo.path, 'foo'), 'w') as f: f.write("BAR") with open(os.path.join(self.repo.path, 'bar'), 'w') as f: f.write("BAR") self.assertEqual( ['foo'], list(porcelain.check_ignore(self.repo, ['foo']))) self.assertEqual([], list(porcelain.check_ignore(self.repo, ['bar']))) def test_check_added(self): with open(os.path.join(self.repo.path, 'foo'), 'w') as f: f.write("BAR") self.repo.stage(['foo']) with open(os.path.join(self.repo.path, '.gitignore'), 'w') as f: f.write("foo\n") self.assertEqual( [], list(porcelain.check_ignore(self.repo, ['foo']))) self.assertEqual( ['foo'], list(porcelain.check_ignore(self.repo, ['foo'], no_index=True))) class UpdateHeadTests(PorcelainTestCase): def test_set_to_branch(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo.refs[b"refs/heads/blah"] = c1.id porcelain.update_head(self.repo, "blah") self.assertEqual(c1.id, self.repo.head()) self.assertEqual(b'ref: refs/heads/blah', self.repo.refs.read_ref('HEAD')) def test_set_to_branch_detached(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo.refs[b"refs/heads/blah"] = c1.id porcelain.update_head(self.repo, "blah", detached=True) self.assertEqual(c1.id, self.repo.head()) self.assertEqual(c1.id, self.repo.refs.read_ref(b'HEAD')) def test_set_to_commit_detached(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo.refs[b"refs/heads/blah"] = c1.id porcelain.update_head(self.repo, c1.id, detached=True) self.assertEqual(c1.id, self.repo.head()) self.assertEqual(c1.id, self.repo.refs.read_ref(b'HEAD')) def test_set_new_branch(self): [c1] = build_commit_graph(self.repo.object_store, [[1]]) self.repo.refs[b"refs/heads/blah"] = c1.id porcelain.update_head(self.repo, "blah", new_branch="bar") self.assertEqual(c1.id, self.repo.head()) self.assertEqual(b'ref: refs/heads/bar', self.repo.refs.read_ref(b'HEAD')) diff --git a/dulwich/tests/test_refs.py b/dulwich/tests/test_refs.py index 517102be..2ec94444 100644 --- a/dulwich/tests/test_refs.py +++ b/dulwich/tests/test_refs.py @@ -1,554 +1,579 @@ # test_refs.py -- tests for refs.py # encoding: utf-8 # Copyright (C) 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. # """Tests for dulwich.refs.""" from io import BytesIO import os import sys import tempfile from dulwich import errors from dulwich.file import ( GitFile, ) from dulwich.objects import ZERO_SHA from dulwich.refs import ( DictRefsContainer, InfoRefsContainer, check_ref_format, _split_ref_line, parse_symref_value, read_packed_refs_with_peeled, read_packed_refs, + strip_peeled_refs, write_packed_refs, ) from dulwich.repo import Repo from dulwich.tests import ( SkipTest, TestCase, ) from dulwich.tests.utils import ( open_repo, tear_down_repo, ) class CheckRefFormatTests(TestCase): """Tests for the check_ref_format function. These are the same tests as in the git test suite. """ def test_valid(self): self.assertTrue(check_ref_format(b'heads/foo')) self.assertTrue(check_ref_format(b'foo/bar/baz')) self.assertTrue(check_ref_format(b'refs///heads/foo')) self.assertTrue(check_ref_format(b'foo./bar')) self.assertTrue(check_ref_format(b'heads/foo@bar')) self.assertTrue(check_ref_format(b'heads/fix.lock.error')) def test_invalid(self): self.assertFalse(check_ref_format(b'foo')) self.assertFalse(check_ref_format(b'heads/foo/')) self.assertFalse(check_ref_format(b'./foo')) self.assertFalse(check_ref_format(b'.refs/foo')) self.assertFalse(check_ref_format(b'heads/foo..bar')) self.assertFalse(check_ref_format(b'heads/foo?bar')) self.assertFalse(check_ref_format(b'heads/foo.lock')) self.assertFalse(check_ref_format(b'heads/v@{ation')) self.assertFalse(check_ref_format(b'heads/foo\bar')) ONES = b'1' * 40 TWOS = b'2' * 40 THREES = b'3' * 40 FOURS = b'4' * 40 class PackedRefsFileTests(TestCase): def test_split_ref_line_errors(self): self.assertRaises(errors.PackedRefsException, _split_ref_line, b'singlefield') self.assertRaises(errors.PackedRefsException, _split_ref_line, b'badsha name') self.assertRaises(errors.PackedRefsException, _split_ref_line, ONES + b' bad/../refname') def test_read_without_peeled(self): f = BytesIO(b'\n'.join([ b'# comment', ONES + b' ref/1', TWOS + b' ref/2'])) self.assertEqual([(ONES, b'ref/1'), (TWOS, b'ref/2')], list(read_packed_refs(f))) def test_read_without_peeled_errors(self): f = BytesIO(b'\n'.join([ ONES + b' ref/1', b'^' + TWOS])) self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f)) def test_read_with_peeled(self): f = BytesIO(b'\n'.join([ ONES + b' ref/1', TWOS + b' ref/2', b'^' + THREES, FOURS + b' ref/4'])) self.assertEqual([ (ONES, b'ref/1', None), (TWOS, b'ref/2', THREES), (FOURS, b'ref/4', None), ], list(read_packed_refs_with_peeled(f))) def test_read_with_peeled_errors(self): f = BytesIO(b'\n'.join([ b'^' + TWOS, ONES + b' ref/1'])) self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f)) f = BytesIO(b'\n'.join([ ONES + b' ref/1', b'^' + TWOS, b'^' + THREES])) self.assertRaises(errors.PackedRefsException, list, read_packed_refs(f)) def test_write_with_peeled(self): f = BytesIO() write_packed_refs(f, {b'ref/1': ONES, b'ref/2': TWOS}, {b'ref/1': THREES}) self.assertEqual( b'\n'.join([b'# pack-refs with: peeled', ONES + b' ref/1', b'^' + THREES, TWOS + b' ref/2']) + b'\n', f.getvalue()) def test_write_without_peeled(self): f = BytesIO() write_packed_refs(f, {b'ref/1': ONES, b'ref/2': TWOS}) self.assertEqual(b'\n'.join([ONES + b' ref/1', TWOS + b' ref/2']) + b'\n', f.getvalue()) # Dict of refs that we expect all RefsContainerTests subclasses to define. _TEST_REFS = { b'HEAD': b'42d06bd4b77fed026b154d16493e5deab78f02ec', b'refs/heads/40-char-ref-aaaaaaaaaaaaaaaaaa': b'42d06bd4b77fed026b154d16493e5deab78f02ec', b'refs/heads/master': b'42d06bd4b77fed026b154d16493e5deab78f02ec', b'refs/heads/packed': b'42d06bd4b77fed026b154d16493e5deab78f02ec', b'refs/tags/refs-0.1': b'df6800012397fb85c56e7418dd4eb9405dee075c', b'refs/tags/refs-0.2': b'3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8', b'refs/heads/loop': b'ref: refs/heads/loop', } class RefsContainerTests(object): def test_keys(self): actual_keys = set(self._refs.keys()) self.assertEqual(set(self._refs.allkeys()), actual_keys) self.assertEqual(set(_TEST_REFS.keys()), actual_keys) actual_keys = self._refs.keys(b'refs/heads') actual_keys.discard(b'loop') self.assertEqual( [b'40-char-ref-aaaaaaaaaaaaaaaaaa', b'master', b'packed'], sorted(actual_keys)) self.assertEqual([b'refs-0.1', b'refs-0.2'], sorted(self._refs.keys(b'refs/tags'))) def test_as_dict(self): # refs/heads/loop does not show up even if it exists expected_refs = dict(_TEST_REFS) del expected_refs[b'refs/heads/loop'] self.assertEqual(expected_refs, self._refs.as_dict()) def test_get_symrefs(self): self._refs.set_symbolic_ref(b'refs/heads/src', b'refs/heads/dst') symrefs = self._refs.get_symrefs() if b'HEAD' in symrefs: symrefs.pop(b'HEAD') self.assertEqual({b'refs/heads/src': b'refs/heads/dst', b'refs/heads/loop': b'refs/heads/loop'}, symrefs) def test_setitem(self): self._refs[b'refs/some/ref'] = ( b'42d06bd4b77fed026b154d16493e5deab78f02ec') self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs[b'refs/some/ref']) self.assertRaises( errors.RefFormatError, self._refs.__setitem__, b'notrefs/foo', b'42d06bd4b77fed026b154d16493e5deab78f02ec') def test_set_if_equals(self): nines = b'9' * 40 self.assertFalse(self._refs.set_if_equals(b'HEAD', b'c0ffee', nines)) self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs[b'HEAD']) self.assertTrue(self._refs.set_if_equals( b'HEAD', b'42d06bd4b77fed026b154d16493e5deab78f02ec', nines)) self.assertEqual(nines, self._refs[b'HEAD']) self.assertTrue(self._refs.set_if_equals(b'refs/heads/master', None, nines)) self.assertEqual(nines, self._refs[b'refs/heads/master']) self.assertTrue(self._refs.set_if_equals( b'refs/heads/nonexistant', ZERO_SHA, nines)) self.assertEqual(nines, self._refs[b'refs/heads/nonexistant']) def test_add_if_new(self): nines = b'9' * 40 self.assertFalse(self._refs.add_if_new(b'refs/heads/master', nines)) self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs[b'refs/heads/master']) self.assertTrue(self._refs.add_if_new(b'refs/some/ref', nines)) self.assertEqual(nines, self._refs[b'refs/some/ref']) def test_set_symbolic_ref(self): self._refs.set_symbolic_ref(b'refs/heads/symbolic', b'refs/heads/master') self.assertEqual(b'ref: refs/heads/master', self._refs.read_loose_ref(b'refs/heads/symbolic')) self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs[b'refs/heads/symbolic']) def test_set_symbolic_ref_overwrite(self): nines = b'9' * 40 self.assertFalse(b'refs/heads/symbolic' in self._refs) self._refs[b'refs/heads/symbolic'] = nines self.assertEqual(nines, self._refs.read_loose_ref(b'refs/heads/symbolic')) self._refs.set_symbolic_ref(b'refs/heads/symbolic', b'refs/heads/master') self.assertEqual(b'ref: refs/heads/master', self._refs.read_loose_ref(b'refs/heads/symbolic')) self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs[b'refs/heads/symbolic']) def test_check_refname(self): self._refs._check_refname(b'HEAD') self._refs._check_refname(b'refs/stash') self._refs._check_refname(b'refs/heads/foo') self.assertRaises(errors.RefFormatError, self._refs._check_refname, b'refs') self.assertRaises(errors.RefFormatError, self._refs._check_refname, b'notrefs/foo') def test_contains(self): self.assertTrue(b'refs/heads/master' in self._refs) self.assertFalse(b'refs/heads/bar' in self._refs) def test_delitem(self): self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs[b'refs/heads/master']) del self._refs[b'refs/heads/master'] self.assertRaises(KeyError, lambda: self._refs[b'refs/heads/master']) def test_remove_if_equals(self): self.assertFalse(self._refs.remove_if_equals(b'HEAD', b'c0ffee')) self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs[b'HEAD']) self.assertTrue(self._refs.remove_if_equals( b'refs/tags/refs-0.2', b'3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8')) self.assertTrue(self._refs.remove_if_equals( b'refs/tags/refs-0.2', ZERO_SHA)) self.assertFalse(b'refs/tags/refs-0.2' in self._refs) class DictRefsContainerTests(RefsContainerTests, TestCase): def setUp(self): TestCase.setUp(self) self._refs = DictRefsContainer(dict(_TEST_REFS)) def test_invalid_refname(self): # FIXME: Move this test into RefsContainerTests, but requires # some way of injecting invalid refs. self._refs._refs[b'refs/stash'] = b'00' * 20 expected_refs = dict(_TEST_REFS) del expected_refs[b'refs/heads/loop'] expected_refs[b'refs/stash'] = b'00' * 20 self.assertEqual(expected_refs, self._refs.as_dict()) class DiskRefsContainerTests(RefsContainerTests, TestCase): def setUp(self): TestCase.setUp(self) self._repo = open_repo('refs.git') self.addCleanup(tear_down_repo, self._repo) self._refs = self._repo.refs def test_get_packed_refs(self): self.assertEqual({ b'refs/heads/packed': b'42d06bd4b77fed026b154d16493e5deab78f02ec', b'refs/tags/refs-0.1': b'df6800012397fb85c56e7418dd4eb9405dee075c', }, self._refs.get_packed_refs()) def test_get_peeled_not_packed(self): # not packed self.assertEqual(None, self._refs.get_peeled(b'refs/tags/refs-0.2')) self.assertEqual(b'3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8', self._refs[b'refs/tags/refs-0.2']) # packed, known not peelable self.assertEqual(self._refs[b'refs/heads/packed'], self._refs.get_peeled(b'refs/heads/packed')) # packed, peeled self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs.get_peeled(b'refs/tags/refs-0.1')) def test_setitem(self): RefsContainerTests.test_setitem(self) f = open(os.path.join(self._refs.path, 'refs', 'some', 'ref'), 'rb') self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', f.read()[:40]) f.close() def test_setitem_symbolic(self): ones = b'1' * 40 self._refs[b'HEAD'] = ones self.assertEqual(ones, self._refs[b'HEAD']) # ensure HEAD was not modified f = open(os.path.join(self._refs.path, 'HEAD'), 'rb') v = next(iter(f)).rstrip(b'\n\r') f.close() self.assertEqual(b'ref: refs/heads/master', v) # ensure the symbolic link was written through f = open(os.path.join(self._refs.path, 'refs', 'heads', 'master'), 'rb') self.assertEqual(ones, f.read()[:40]) f.close() def test_set_if_equals(self): RefsContainerTests.test_set_if_equals(self) # ensure symref was followed self.assertEqual(b'9' * 40, self._refs[b'refs/heads/master']) # ensure lockfile was deleted self.assertFalse(os.path.exists( os.path.join(self._refs.path, 'refs', 'heads', 'master.lock'))) self.assertFalse(os.path.exists( os.path.join(self._refs.path, 'HEAD.lock'))) def test_add_if_new_packed(self): # don't overwrite packed ref self.assertFalse(self._refs.add_if_new(b'refs/tags/refs-0.1', b'9' * 40)) self.assertEqual(b'df6800012397fb85c56e7418dd4eb9405dee075c', self._refs[b'refs/tags/refs-0.1']) def test_add_if_new_symbolic(self): # Use an empty repo instead of the default. repo_dir = os.path.join(tempfile.mkdtemp(), 'test') os.makedirs(repo_dir) repo = Repo.init(repo_dir) self.addCleanup(tear_down_repo, repo) refs = repo.refs nines = b'9' * 40 self.assertEqual(b'ref: refs/heads/master', refs.read_ref(b'HEAD')) self.assertFalse(b'refs/heads/master' in refs) self.assertTrue(refs.add_if_new(b'HEAD', nines)) self.assertEqual(b'ref: refs/heads/master', refs.read_ref(b'HEAD')) self.assertEqual(nines, refs[b'HEAD']) self.assertEqual(nines, refs[b'refs/heads/master']) self.assertFalse(refs.add_if_new(b'HEAD', b'1' * 40)) self.assertEqual(nines, refs[b'HEAD']) self.assertEqual(nines, refs[b'refs/heads/master']) def test_follow(self): self.assertEqual(([b'HEAD', b'refs/heads/master'], b'42d06bd4b77fed026b154d16493e5deab78f02ec'), self._refs.follow(b'HEAD')) self.assertEqual(([b'refs/heads/master'], b'42d06bd4b77fed026b154d16493e5deab78f02ec'), self._refs.follow(b'refs/heads/master')) self.assertRaises(KeyError, self._refs.follow, b'refs/heads/loop') def test_delitem(self): RefsContainerTests.test_delitem(self) ref_file = os.path.join(self._refs.path, 'refs', 'heads', 'master') self.assertFalse(os.path.exists(ref_file)) self.assertFalse(b'refs/heads/master' in self._refs.get_packed_refs()) def test_delitem_symbolic(self): self.assertEqual(b'ref: refs/heads/master', self._refs.read_loose_ref(b'HEAD')) del self._refs[b'HEAD'] self.assertRaises(KeyError, lambda: self._refs[b'HEAD']) self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs[b'refs/heads/master']) self.assertFalse(os.path.exists(os.path.join(self._refs.path, 'HEAD'))) def test_remove_if_equals_symref(self): # HEAD is a symref, so shouldn't equal its dereferenced value self.assertFalse(self._refs.remove_if_equals( b'HEAD', b'42d06bd4b77fed026b154d16493e5deab78f02ec')) self.assertTrue(self._refs.remove_if_equals( b'refs/heads/master', b'42d06bd4b77fed026b154d16493e5deab78f02ec')) self.assertRaises(KeyError, lambda: self._refs[b'refs/heads/master']) # HEAD is now a broken symref self.assertRaises(KeyError, lambda: self._refs[b'HEAD']) self.assertEqual(b'ref: refs/heads/master', self._refs.read_loose_ref(b'HEAD')) self.assertFalse(os.path.exists( os.path.join(self._refs.path, 'refs', 'heads', 'master.lock'))) self.assertFalse(os.path.exists( os.path.join(self._refs.path, 'HEAD.lock'))) def test_remove_packed_without_peeled(self): refs_file = os.path.join(self._repo.path, 'packed-refs') f = GitFile(refs_file) refs_data = f.read() f.close() f = GitFile(refs_file, 'wb') f.write(b'\n'.join(l for l in refs_data.split(b'\n') if not l or l[0] not in b'#^')) f.close() self._repo = Repo(self._repo.path) refs = self._repo.refs self.assertTrue(refs.remove_if_equals( b'refs/heads/packed', b'42d06bd4b77fed026b154d16493e5deab78f02ec')) def test_remove_if_equals_packed(self): # test removing ref that is only packed self.assertEqual(b'df6800012397fb85c56e7418dd4eb9405dee075c', self._refs[b'refs/tags/refs-0.1']) self.assertTrue( self._refs.remove_if_equals( b'refs/tags/refs-0.1', b'df6800012397fb85c56e7418dd4eb9405dee075c')) self.assertRaises(KeyError, lambda: self._refs[b'refs/tags/refs-0.1']) def test_read_ref(self): self.assertEqual(b'ref: refs/heads/master', self._refs.read_ref(b'HEAD')) self.assertEqual(b'42d06bd4b77fed026b154d16493e5deab78f02ec', self._refs.read_ref(b'refs/heads/packed')) self.assertEqual(None, self._refs.read_ref(b'nonexistant')) def test_non_ascii(self): try: encoded_ref = u'refs/tags/schön'.encode( sys.getfilesystemencoding()) except UnicodeEncodeError: raise SkipTest( "filesystem encoding doesn't support special character") p = os.path.join(self._repo.path, 'refs', 'tags', u'schön') with open(p, 'w') as f: f.write('00' * 20) expected_refs = dict(_TEST_REFS) expected_refs[encoded_ref] = b'00' * 20 del expected_refs[b'refs/heads/loop'] self.assertEqual(expected_refs, self._repo.get_refs()) _TEST_REFS_SERIALIZED = ( b'42d06bd4b77fed026b154d16493e5deab78f02ec\t' b'refs/heads/40-char-ref-aaaaaaaaaaaaaaaaaa\n' b'42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/master\n' b'42d06bd4b77fed026b154d16493e5deab78f02ec\trefs/heads/packed\n' b'df6800012397fb85c56e7418dd4eb9405dee075c\trefs/tags/refs-0.1\n' b'3ec9c43c84ff242e3ef4a9fc5bc111fd780a76a8\trefs/tags/refs-0.2\n') class InfoRefsContainerTests(TestCase): def test_invalid_refname(self): text = _TEST_REFS_SERIALIZED + b'00' * 20 + b'\trefs/stash\n' refs = InfoRefsContainer(BytesIO(text)) expected_refs = dict(_TEST_REFS) del expected_refs[b'HEAD'] expected_refs[b'refs/stash'] = b'00' * 20 del expected_refs[b'refs/heads/loop'] self.assertEqual(expected_refs, refs.as_dict()) def test_keys(self): refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED)) actual_keys = set(refs.keys()) self.assertEqual(set(refs.allkeys()), actual_keys) expected_refs = dict(_TEST_REFS) del expected_refs[b'HEAD'] del expected_refs[b'refs/heads/loop'] self.assertEqual(set(expected_refs.keys()), actual_keys) actual_keys = refs.keys(b'refs/heads') actual_keys.discard(b'loop') self.assertEqual( [b'40-char-ref-aaaaaaaaaaaaaaaaaa', b'master', b'packed'], sorted(actual_keys)) self.assertEqual([b'refs-0.1', b'refs-0.2'], sorted(refs.keys(b'refs/tags'))) def test_as_dict(self): refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED)) # refs/heads/loop does not show up even if it exists expected_refs = dict(_TEST_REFS) del expected_refs[b'HEAD'] del expected_refs[b'refs/heads/loop'] self.assertEqual(expected_refs, refs.as_dict()) def test_contains(self): refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED)) self.assertTrue(b'refs/heads/master' in refs) self.assertFalse(b'refs/heads/bar' in refs) def test_get_peeled(self): refs = InfoRefsContainer(BytesIO(_TEST_REFS_SERIALIZED)) # refs/heads/loop does not show up even if it exists self.assertEqual( _TEST_REFS[b'refs/heads/master'], refs.get_peeled(b'refs/heads/master')) class ParseSymrefValueTests(TestCase): def test_valid(self): self.assertEqual( b'refs/heads/foo', parse_symref_value(b'ref: refs/heads/foo')) def test_invalid(self): self.assertRaises(ValueError, parse_symref_value, b'foobar') + + +class StripPeeledRefsTests(TestCase): + + all_refs = { + b'refs/heads/master': b'8843d7f92416211de9ebb963ff4ce28125932878', + b'refs/heads/testing': b'186a005b134d8639a58b6731c7c1ea821a6eedba', + b'refs/tags/1.0.0': b'a93db4b0360cc635a2b93675010bac8d101f73f0', + b'refs/tags/1.0.0^{}': b'a93db4b0360cc635a2b93675010bac8d101f73f0', + b'refs/tags/2.0.0': b'0749936d0956c661ac8f8d3483774509c165f89e', + b'refs/tags/2.0.0^{}': b'0749936d0956c661ac8f8d3483774509c165f89e', + } + non_peeled_refs = { + b'refs/heads/master': b'8843d7f92416211de9ebb963ff4ce28125932878', + b'refs/heads/testing': b'186a005b134d8639a58b6731c7c1ea821a6eedba', + b'refs/tags/1.0.0': b'a93db4b0360cc635a2b93675010bac8d101f73f0', + b'refs/tags/2.0.0': b'0749936d0956c661ac8f8d3483774509c165f89e', + } + + def test_strip_peeled_refs(self): + # Simple check of two dicts + self.assertEqual( + strip_peeled_refs(self.all_refs), + self.non_peeled_refs)