### XXX: can we reduce number of calls to dpkg-parsechangelog
### XXX: is any of this data in lp already?

import collections
from contextlib import contextmanager
from copy import copy
import datetime
import enum
from functools import lru_cache
import itertools
import logging
import os
import posixpath
import re
import shutil
import stat
from subprocess import CalledProcessError
import sys
import tempfile
from gitubuntu.__main__ import top_level_defaults
import gitubuntu.build
from gitubuntu.dsc import component_tarball_matches
from gitubuntu.patch_state import PatchState
from gitubuntu.run import (
   decode_binary,
   run,
   runq,
   run_gbp,
   run_quilt,
)
import gitubuntu.spec
from gitubuntu.test_util import get_test_changelog
import gitubuntu.versioning
import debian.changelog
import debian.debian_support
import pygit2
from pygit2.enums import ObjectType
import pytest


def _follow_symlinks_to_blob(repo, top_tree_object, search_path,
    _rel_tree=None, _rel_path=''
):
    '''Recursively follow a path down a tree, following symlinks, to find blob

    repo: pygit2.Repository object
    top_tree: pygit2.Tree object of the top of the tree structure
    search_path: '/'-separated path string of blob to find
    _rel_tree: (internal) which tree to look further into
    _rel_path: (internal) the path we are in so far
    '''

    NORMAL_BLOB_MODES = set([
        pygit2.GIT_FILEMODE_BLOB,
        pygit2.GIT_FILEMODE_BLOB_EXECUTABLE,
    ])

    _rel_tree = _rel_tree or top_tree_object
    head, tail = posixpath.split(search_path)

    # A traditional functional split would put a single entry in head with tail
    # empty, but posixpath.split doesn't necessarily do this. Jiggle it round
    # to make it appear to have traditional semantics.
    if not head:
        head = tail
        tail = None

    entry = _rel_tree[head]
    if entry.type in [ObjectType.TREE, 'tree']:
        return _follow_symlinks_to_blob(
            repo=repo,
            top_tree_object=top_tree_object,
            search_path=tail,
            _rel_tree=repo.get(entry.id),
            _rel_path=posixpath.join(_rel_path, head),
        )
    elif entry.type in [ObjectType.BLOB, 'blob'] and entry.filemode == pygit2.GIT_FILEMODE_LINK:
        # Found a symlink. Start again from the top with adjustment for symlink
        # following
        target_tail = [decode_binary(repo.get(entry.id).data)]
        if tail is not None:
            target_tail.append(tail)
        search_path = posixpath.normpath(
            posixpath.join(_rel_path, *target_tail)
        )
        return _follow_symlinks_to_blob(
            repo=repo,
            top_tree_object=top_tree_object,
            search_path=search_path,
        )
    elif entry.type in [ObjectType.BLOB, 'blob'] and entry.filemode in NORMAL_BLOB_MODES:
        return repo.get(entry.id)
    else:
        # Found some special entry such as a "gitlink" (submodule entry)
        raise ValueError(
            "Found %r filemode %r looking for %r" %
                (entry, entry.filemode, posixpath.join(_rel_path, search_path))
        )


def follow_symlinks_to_blob(repo, treeish_object, path):
    return _follow_symlinks_to_blob(
        repo=repo,
        top_tree_object=treeish_object.peel(pygit2.Tree),
        search_path=posixpath.normpath(path),
    )


def _derive_git_cli_env(
    pygit2_repo,
    initial_env=None,
    update_env=None,
    work_tree_path=None,
    index_path=None,
):
    """Calculate the environment to be used in a call to the git CLI

    :param pygit2.Repository pygit2_repo: the repository for which to calculate
        the environment
    :param dict initial_env: the environment to start with
    :param dict update_env: additional environment setings with which to
        override the result
    :param str work_tree_path: in the case of an alternate work tree being
        used, specify this here and GIT_WORK_TREE will be set to it instead of
        the default being taken from the work tree used by pygit2_repo
    :param str index_path: if an alternate index is being used, specify it here
        and GIT_INDEX_FILE will be set accordingly.
    :rtype: dict
    :returns: a dictionary representing the environment with which to call the
        git CLI

    This function encapsulates the setting of the GIT_DIR, GIT_WORK_TREE and
    GIT_INDEX_FILE environment variables as necessary. The provided
    pygit2.Repository instance is used to determine these values. initial_env,
    if provided, specifies the initial environment to use instead of defaulting
    to the process' current environment. update_env allows extra environment
    variables to be added as well as the override of any variables set by this
    function, including GIT_DIR, GIT_WORK_TREE and GIT_INDEX_FILE.
    """
    if initial_env is None:
        env = os.environ.copy()
    else:
        env = initial_env.copy()

    env['GIT_DIR'] = pygit2_repo.path

    if work_tree_path is None:
        env['GIT_WORK_TREE'] = pygit2_repo.workdir
    else:
        env['GIT_WORK_TREE'] = work_tree_path

    if index_path is not None:
        env['GIT_INDEX_FILE'] = index_path

    if update_env:
        env.update(update_env)

    return env


def _derive_target_branch_string(remote_branch_objects):
    '''Given a list of branch objects, return the name of the one to use as the target branch

    Returns either one of the branch objects' names, or the empty string
    to indicate no suitable candidate.
    '''
    if len(remote_branch_objects) == 0:
        logging.error("Unable to automatically determine importer "
            "branch: No candidate branches found."
        )
        return ''
    remote_branch_strings = [
        b.branch_name for b in remote_branch_objects
    ]
    if len(remote_branch_objects) > 1:
        # do the trees of each branch's tip match?
        if len(
            set(b.peel(pygit2.Tree).id for b in remote_branch_objects)
        ) != 1:
            logging.error("Unable to automatically determine importer "
                "branch: Multiple candidate branches found and "
                "their trees do not match: %s. This might be a "
                "bug in `git ubuntu lint`, please report it at "
                "https://bugs.launchpad.net/git-ubuntu. "
                "Please pass  --target-branch.",
                ", ".join(remote_branch_strings)
            )
            return ''
        # is ubuntu/devel one of the candidates?
        try:
            return [
                b for b in remote_branch_strings if 'ubuntu/devel' in b
            ].pop()
        except IndexError:
            pass
        # are all candidate branches for the same series?
        pkg_remote_branch_serieses = set(
            # remove the prefix, trim the distribution and
            # extract the series
            b[len('pkg/'):].split('/')[1].split('-')[0] for
            b in remote_branch_strings
        )
        if len(pkg_remote_branch_serieses) != 1:
            logging.error("Unable to automatically determine importer "
                "branch: Multiple candidate branches found and "
                "they do not target the same series: %s. Please pass "
                "--target-branch.", ", ".join(remote_branch_strings)
            )
            return ''
        # is a -devel branch present?
        if not any('-devel' in b for b in remote_branch_strings):
            logging.error("Unable to automatically determine importer "
                "branch: Multiple candidate branches found and "
                "none appear to be a -devel branch: %s. Please "
                "pass  --target-branch.", ", ".join(remote_branch_strings)
            )
            return ''
        # if so, favor -devel
        remote_branch_strings = [
            b for b in remote_branch_strings if '-devel' in b
        ]
    return remote_branch_strings.pop()

def derive_target_branch(repo, commitish_string, namespace='pkg'):
    return _derive_target_branch_string(
        repo.nearest_remote_branches(commitish_string, namespace)
    )


def git_run(
    pygit2_repo,
    args,
    initial_env=None,
    update_env=None,
    work_tree_path=None,
    index_path=None,
    **kwargs
):
    """Run the git CLI with the provided arguments

    :param pygit2.Repository: the repository on which to act
    :param list(str) args: arguments to the git CLI
    :param dict initial_env: the environment to use
    :param dict update_env: additional environment variables and overrides
    :param dict **kwargs: further arguments to pass through to
        gitubuntu.run.run()
    :raises subprocess.CalledProcessError: if git exits non-zero
    :rtype: (str, str)
    :returns: stdout and stderr strings containing the subprocess output

    If initial_env is not set, it defaults to the current process' environment.

    The GIT_DIR, GIT_WORK_TREE and GIT_INDEX_FILE environment variables are set
    automatically as necessary based on the repository's existing location and
    settings.

    If update_env is set, then the environment to be used is updated with env
    before the call to git is made. This can override GIT_DIR,
    GIT_WORK_TREE, GIT_INDEX_FILE and anything else.
    """
    env = _derive_git_cli_env(
        pygit2_repo=pygit2_repo,
        initial_env=initial_env,
        update_env=update_env,
        work_tree_path=work_tree_path,
        index_path=index_path,
    )
    return run(['git'] + list(args), env=env, **kwargs)


class RenameableDir:
    """An on-disk directory that can be renamed and traversed recursively.

    This is a thin wrapper around a filesystem path string (and must be
    instantiated with one). Methods and attributes are modeled around a
    py.path, but we do not use py.path as we don't really need its
    functionality and it would add another dependency. This interface allows
    for filesystem operations to be easily faked with a FakeRenameableDir for
    testing consumers of this class.

    One wart around renaming and py.path is that once renamed a py.path object
    becomes useless as it no longer validly refers to an on-disk path. Rather
    than supporting a rename method, this wrapper provides a basename
    setter to handle the rename and replacement wrapped string object
    transparently. This moves complexity away from the class consumer, allowing
    the consumer to be tested more easily.

    Since the underlying purpose of this class is to handle manipulations of a
    directory tree for adjustments needed during import/export, symlink
    handling is effectively "turned off" in the specification of this class.
    Symlinks to directories are not recursed into; they are handled no
    differently to a regular file, in the same manner as lstat(2).
    """
    def __init__(self, path):
        """Create a new RenameableDir instance.

        :param str path: the on-disk directory to wrap, which must exist. For
            symlinks, it is the symlink itself that must exist; the existence
            of a symlink's target does not matter.
        :raises FileNotFoundError: if the path supplied does not exist.
        """
        # Ignore the return value of os.lstat(); this call is used to raise
        # FileNotFoundError if the path does not exist (as required in the spec
        # specified by the docstring), or succeed otherwise. The
        # call for os.path.lexists() would use the same underlying system call
        # anyway, so this is equivalent and this way we end up with a full
        # FileNotFoundError exception created for us with all the correct
        # parameters.
        os.lstat(path)

        self._path = path

    @property
    def basename(self):
        """The name of the directory itself."""
        return os.path.basename(self._path)

    @basename.setter
    def basename(self, new_basename):
        """Rename this directory."""
        renamed_path = os.path.join(os.path.dirname(self._path), new_basename)
        os.rename(self._path, renamed_path)
        self._path = renamed_path

    def listdir(self, fil=lambda x: True):
        """Return subdirectory objects.

        :param fil: a function that, given a basename, returns a boolean
            indicating whether or not the corresponding object should be
            returned in the results.
        """
        return [
            RenameableDir(os.path.join(self._path, p))
            for p in os.listdir(self._path)
            if fil(p)
        ]

    @property
    def recursive(self):
        """Indicate if this object can contain subdirectory objects.

        An object representing a file will return False. An object representing
        a directory will return True, even if it is empty.

        Symlinks return False even if they point to a directory. Broken
        symlinks also always return False.

        :rtype: bool
        """
        st = os.stat(self._path, follow_symlinks=False)
        return stat.S_ISDIR(st.st_mode)

    def __str__(self):
        return str(self._path)

    def __repr__(self):
        return 'RenameableDir(%r)' % str(self)

    def __hash__(self):
        # https://stackoverflow.com/q/2909106/478206
        return hash((
            type(self),
            self._path
        ))

    def __eq__(self, other):
        return hash(self) == hash(other)


class FakeRenameableDir:
    """A fake RenameDir that retains its structure in memory.

    This is useful for testing consumers of a RenameableDir.

    In addition, renames are recorded and those records passed up to parent
    FakeRenameableDir objects so that the order of renames that occur can be
    checked later.
    """
    def __init__(self, basename, subdirs):
        """Create a new RenameableDir instance.

        :param str basename: the basename of this instance.
        :param subdirs: FakeRenameableDir objects contained within this one.
            For non-recursive objects (such as those intended to represent
            files), use None.
        :type subdirs: list(FakeRenameableDir)
        """
        self._basename = basename
        self._subdirs = subdirs

        if self._subdirs:
            for subdir in self._subdirs:
                subdir._parent = self

        self._parent = None
        self._rename_record = []

    @property
    def basename(self):
        return self._basename

    @basename.setter
    def basename(self, new_basename):
        self._record_rename(self)
        self._basename = new_basename

    def _record_rename(self, obj):
        self._rename_record.append(obj)
        if self._parent:
            self._parent._record_rename(obj)

    def listdir(self, fil=lambda x: True):
        return (subdir for subdir in self._subdirs if fil(subdir.basename))

    @property
    def recursive(self):
        return self._subdirs is not None

    def __hash__(self):
        # https://stackoverflow.com/q/2909106/478206
        return hash((
            type(self),
            self.basename,
            None if self._subdirs is None else tuple(self._subdirs),
        ))

    def __eq__(self, other):
        return hash(self) == hash(other)

    def __repr__(self):
        return 'FakeRenameableDir(%r, %r)' % (self.basename, self._subdirs)


_dot_git_match = re.compile(r'^\.+git$').search
_EscapeDirection = enum.Enum('EscapeDirection', ['ESCAPE', 'UNESCAPE'])


def _escape_unescape_dot_git(path, direction):
    """Escape or unescape .git entries in a directory recursively.

    :param RenameableDir path: top of directory tree to escape or unescape.
    :param _EscapeDirection direction: whether to escape or unescape.

    Escaping rules:
        .git -> ..git
        ..git -> ...git
        ...git -> ....git
        etc.

        All these escaping rules apply all of the time, regardless of whether
        or not .git exists. Only names matching '.git' with zero or more '.'
        prepended are touched.

    This allows any directory tree to be losslessly stored in git, since git
    does not permit entries named '.git'.

    Unescaping is the inverse of escaping. Before unescaping, an entry called
    '.git' must not exist. If it does, RuntimeError is raised, and the
    directory is left in an undefined (probably partially unescaped) state.
    """
    # When escaping, we have to rename ..git to ...git before renaming .git to
    # ..git in order to make room, and the reverse for unescaping. If we do the
    # renames ordered by length of name, we can meet this requirement.
    # Escaping: order by longest first; unescaping: order by shortest first.
    sorted_subpaths_to_rename = sorted(
        path.listdir(fil=_dot_git_match),
        key=lambda p: len(p.basename),
        reverse=direction is _EscapeDirection.ESCAPE,
    )
    for entry in sorted_subpaths_to_rename:
        if direction is _EscapeDirection.ESCAPE:
            # Add a leading '.'
            entry.basename = '.' + entry.basename
        else:
            assert direction is _EscapeDirection.UNESCAPE
            if entry.basename == '.git':
                raise RuntimeError(
                    "%s exists but is invalid when unescaping" % entry,
                )
            # Drop the leading '.'
            assert entry.basename[0] == '.'
            entry.basename = entry.basename[1:]

    # Traverse the entire directory for recursive escapes;
    # sorted_subpaths_to_rename is already filtered so is not complete by
    # itself
    for entry in path.listdir():
        if entry.recursive:
            _escape_unescape_dot_git(entry, direction=direction)


def escape_dot_git(path):
    """Apply .git escaping to a filesystem path.

    :param str path: path to filesystem to change
    """
    return _escape_unescape_dot_git(
        path=RenameableDir(path),
        direction=_EscapeDirection.ESCAPE,
    )


def unescape_dot_git(path):
    """Unapply .git escaping to a filesystem path.

    :param str path: path to filesystem to change

    Any entry (including recursively) called '.git' in path is an error and
    will raise a RuntimeError. If an exception is raised, path may be left in a
    partially unescaped state.
    """
    return _escape_unescape_dot_git(
        path=RenameableDir(path),
        direction=_EscapeDirection.UNESCAPE,
    )


class ChangelogError(Exception):
    pass


def _apply_re_substitutions(original_string, subs):
    """Apply a sequence of regular expression substitutions to a string

    :param str original_string: the string to which to apply substitutions
    :param list(str, str) subs: a list of substitutions to apply. The first
        element of each tuple is the regexp to match, and the second is what to
        replace it with.
    :rtype: str
    :returns: the original string but altered by all the substitutions
    """
    changed_string = original_string
    for regex, replacement in subs:
        changed_string = re.sub(regex, replacement, changed_string)
    return changed_string


class Changelog:
    '''Representation of a debian/changelog file found inside a git tree-ish

    Uses dpkg-parsechangelog for parsing, but when this fails we fall
    back to grep/sed-based pattern matching automatically.
    '''
    def __init__(self, content_bytes):
        '''
        contents: bytes string of file contents
        '''
        self._contents = content_bytes
        try:
            self._changelog = debian.changelog.Changelog(
                self._contents,
                strict=True
            )
            if not len(self._changelog.versions):
                # assume bad read, so fall back to shell later
                self._changelog = None
        except (
            UnicodeDecodeError,
            ValueError,
            debian.changelog.ChangelogParseError
        ):
            self._changelog = None

    @classmethod
    def from_treeish(cls, repo, treeish_object):
        '''
        repo: pygit2.Repository instance
        treeish_object: pygit2.Object subclass instance (must peel to pygit2.Tree)
        '''
        blob = follow_symlinks_to_blob(
            repo=repo,
            treeish_object=treeish_object,
            path='debian/changelog'
        )
        return cls(blob.data)

    @classmethod
    def from_path(cls, path):
        with open(path, 'rb') as f:
            return cls(f.read())

    @lru_cache()
    def _dpkg_parsechangelog(self, parse_params):
        stdout, _ = run(
            'dpkg-parsechangelog -l- %s' % parse_params,
            input=self._contents,
            shell=True,
            verbose_on_failure=False,
        )
        return stdout.strip()

    @lru_cache()
    def _shell(self, cmd):
        stdout, _ = run(
            cmd,
            input=self._contents,
            shell=True,
            verbose_on_failure=False,
        )
        return stdout.strip()

    @property
    def _shell_version(self):
        parse_params = '-n1 -SVersion'
        shell_cmd = "grep -m1 '^\\S' | sed 's/.*(\\(.*\\)).*/\\1/'"
        try:
            raw_out = self._dpkg_parsechangelog(parse_params)
        except CalledProcessError:
            raw_out = self._shell(shell_cmd)
        return None if raw_out == '' else raw_out

    @property
    def upstream_version(self):
        if self._changelog:
           return self._changelog.upstream_version
        version = self._shell_version
        m = debian.debian_support.Version.re_valid_version.match(version)
        if m is None:
            raise ValueError("Invalid version string: %s", version)
        return m.group('upstream_version')

    @property
    def version(self):
        if self._changelog:
            try:
                ret = str(self._changelog.versions[0]).strip()
                shell_version = self._shell_version
                if shell_version != 'unknown' and ret != shell_version:
                    raise ChangelogError(
                        'Old (%s) and new (%s) changelog values do not agree' %
                        (self._shell_version, ret)
                    )
                return ret
            except IndexError:
                return None
        return self._shell_version

    @property
    def _shell_previous_version(self):
        parse_params = '-n1 -o1 -SVersion'
        shell_cmd = "grep -m1 '^\\S' | tail -1 | sed 's/.*(\\(.*\\)).*/\\1/'"
        try:
            raw_out = self._dpkg_parsechangelog(parse_params)
        except CalledProcessError:
            raw_out = self._shell(shell_cmd)
        return None if raw_out == '' else raw_out

    @property
    def previous_version(self):
        if self._changelog:
            try:
                ret = str(self._changelog.versions[1]).strip()
                if ret != self._shell_previous_version:
                    raise ChangelogError(
                        'Old (%s) and new (%s) changelog values do not agree' %
                        (self._shell_previous_version, ret)
                    )
                return ret
            except IndexError:
                return None
        return self._shell_previous_version

    @property
    def _shell_maintainer(self):
        parse_params = '-SMaintainer'
        shell_cmd = "grep -m1 '^ --' | sed 's/ -- \\(.*\\)  \\(.*\\)/\\1/'"
        try:
            return self._dpkg_parsechangelog(parse_params)
        except CalledProcessError:
            return self._shell(shell_cmd)

    @property
    def maintainer(self):
        if self._changelog:
            ret = self._changelog.author.strip()
            if ret != self._shell_maintainer:
                raise ChangelogError(
                    'Old (%s) and new (%s) changelog values do not agree' %
                    (self._shell_maintainer, ret)
                )
        else:
            ret = self._shell_maintainer
        if not ret:
            raise ValueError("Unable to parse maintainer from changelog")
        return ret

    @property
    def _shell_date(self):
        parse_params = '-SDate'
        shell_cmd = "grep -m1 '^ --' | sed 's/ -- \\(.*\\)  \\(.*\\)/\\2/'"
        try:
            return self._dpkg_parsechangelog(parse_params)
        except CalledProcessError:
            return self._shell(shell_cmd)

    @property
    def date(self):
        if self._changelog:
            ret = self._changelog.date.strip()
            if ret != self._shell_date:
                raise ChangelogError(
                    'Old (%s) and new (%s) changelog values do not agree' %
                    (self._shell_date, ret)
                )
            return ret
        return self._shell_date

    @property
    def _shell_all_versions(self):
        parse_params = '--format rfc822 -SVersion --all'
        shell_cmd = "grep '^\\S' | sed 's/.*(\\(.*\\)).*/\\1/'"
        try:
            version_lines = self._dpkg_parsechangelog(parse_params)
        except CalledProcessError:
            version_lines = self._shell(shell_cmd)
        return [
            v_stripped
            for v_stripped in (
                v.strip() for v in version_lines.splitlines()
            )
            if v_stripped
        ]

    @property
    def all_versions(self):
        if self._changelog:
            ret = [str(v).strip() for v in self._changelog.versions]
            shell_all_versions = self._shell_all_versions
            is_equivalent = (
                len(ret) == len(shell_all_versions) and
                all(
                    shell_version == 'unknown' or shell_version == api_version
                    for shell_version, api_version
                    in zip(shell_all_versions, ret)
                )
            )
            if not is_equivalent:
                raise ChangelogError(
                    "Old and new changelog values do not agree"
                )
            return ret
        else:
            return self._shell_all_versions

    @property
    def _shell_distribution(self):
        parse_params = '-SDistribution'
        shell_cmd = "grep -m1 '^\\S' | sed 's/.*\\ .*\\ \\(.*\\);.*/\\1/'"
        try:
            return self._dpkg_parsechangelog(parse_params)
        except CalledProcessError:
            return self._shell(shell_cmd)

    @property
    def distribution(self):
        if self._changelog:
            ret = self._changelog.distributions
            if ret != self._shell_distribution:
                raise ChangelogError(
                    'Old (%s) and new (%s) changelog values do not agree' %
                    (self._shell_distribution, ret)
                )
            return ret
        return self._shell_distribution

    @property
    def _shell_srcpkg(self):
        parse_params = '-SSource'
        shell_cmd = "grep -m1 '^\\S' | sed 's/\\(.*\\)\\ .*\\ .*;.*/\\1/'"
        try:
            return self._dpkg_parsechangelog(parse_params)
        except CalledProcessError:
            return self._shell(shell_cmd)

    @property
    def srcpkg(self):
        if self._changelog:
            ret = self._changelog.package.strip()
            if ret != self._shell_srcpkg:
                raise ChangelogError(
                    'Old (%s) and new (%s) changelog values do not agree' %
                    (self._shell_srcpkg, ret)
                )
            return ret
        return self._shell_srcpkg

    @staticmethod
    def _parse_changelog_date(changelog_timestamp_string):
        """Convert changelog timestamp into datetime object

        This function currently requires the locale to have been set to C.UTF-8
        by the caller. This would typically be done at the main entry point to
        the importer.

        :param str changelog_timestamp_string: the timestamp part of the the
            signoff line from a changelog entry
        :rtype: datetime.datetime
        :returns: the timestamp as a datetime object
        :raises ValueError: if the string could not be parsed
        """
        # We avoid using something like dateutil.parser here because the
        # parsing behaviour of malformed or unusually formatted dates must be
        # precisely as specified and not ever change behaviour. If it did, then
        # imports would no longer be reproducible.
        #
        # However, adding new a form of parsing an unambigious date is
        # acceptable if the spec is first updated accordingly since that would
        # only introduce new imports that would have previously failed.
        #
        # time.strptime ignores time zones, so we must use datetime.strptime()

        # strptime doesn't support anything other than standard locale names
        # for days of the week, so handle "Thur", "Thurs", "Tues" and "Sept"
        # abbreviations as special cases as defined in the spec since they are
        # unambiguous.
        adjusted_changelog_timestamp_string = _apply_re_substitutions(
            original_string=changelog_timestamp_string,
            subs=[
                (r'^Thur(s)?,', 'Thu,'),
                (r'^Tues,', 'Tue,'),
                (r'\bSept\b', 'Sep'),
            ],
        )
        acceptable_date_formats = [
            '%a, %d %b %Y %H:%M:%S %z',  # standard
            '%A, %d %b %Y %H:%M:%S %z',  # full day of week
            '%d %b %Y %H:%M:%S %z',      # missing day of week
            '%a, %d %B %Y %H:%M:%S %z',  # full month name
            '%A, %d %B %Y %H:%M:%S %z',  # full day of week and month name
            '%d %B %Y %H:%M:%S %z',      # missing day of week with full month
                                         # name
        ]
        for date_format in acceptable_date_formats:
            try:
                return datetime.datetime.strptime(
                    adjusted_changelog_timestamp_string,
                    date_format,
                )
            except ValueError:
                pass
        else:
            raise ValueError(
                "Could not parse date %r" % changelog_timestamp_string,
            )

    def git_authorship(self, author_date=None):
        """Extract last changelog entry's maintainer and timestamp

        Parse the first changelog entry's sign-off line into git's commit
        authorship metadata model according to the import specification.

        :param datetime.datetime author_date: overrides the author date
            normally parsed from the changelog entry (i.e. for handling date
            parsing edge cases). Any sub-second part of the timestamp is
            truncated.
        :rtype: tuple(str, str, int, int)
        :returns: tuple of name, email, time (in seconds since epoch) and
            offset from UTC (in minutes)
        :raises ValueError: if the changelog sign-off line cannot be parsed
        """
        m = re.match(r'(?P<name>.*)<+(?P<email>.*?)>+', self.maintainer)
        if m is None:
            raise ValueError('Cannot get authorship')

        author_epoch_seconds, author_tz_offset = datetime_to_signature_spec(
            self._parse_changelog_date(self.date)
            if author_date is None
            else author_date
        )

        return (
            # If the author name is empty, then it must be
            # EMPTY_GIT_AUTHOR_NAME because git will not accept an empty author
            # name. See the specification for details.
            (
                m.group('name').strip()
                or gitubuntu.spec.EMPTY_GIT_AUTHOR_NAME
            ),
            m.group('email'),
            author_epoch_seconds,
            author_tz_offset,
        )


class GitUbuntuChangelogError(Exception):
    pass

class PristineTarError(Exception):
    pass

class PristineTarNotFoundError(PristineTarError):
    pass

class MultiplePristineTarFoundError(PristineTarError):
    pass


def git_dep14_tag(version):
    """Munge a version string according to http://dep.debian.net/deps/dep14/"""
    version = str(version)
    version = version.replace('~', '_')
    version = version.replace(':', '%')
    version = version.replace('..', '.#.')
    if version.endswith('.'):
        version = version + '#'
    if version.endswith('.lock'):
        pre, _, _ = version.partition('.lock')
        version = pre + '.#lock'
    return version

def import_tag(version, namespace, patch_state=PatchState.UNAPPLIED):
    return '%s/%s/%s' % (
        namespace,
        {
            PatchState.UNAPPLIED: 'import',
            PatchState.APPLIED: 'applied',
        }[patch_state],
        git_dep14_tag(version),
    )

def reimport_tag_prefix(version, namespace, patch_state=PatchState.UNAPPLIED):
    return '%s/reimport/%s/%s' % (
        namespace,
        {
            PatchState.UNAPPLIED: 'import',
            PatchState.APPLIED: 'applied',
        }[patch_state],
        git_dep14_tag(version),
    )

def reimport_tag(
    version,
    namespace,
    reimport,
    patch_state=PatchState.UNAPPLIED,
):
    return '%s/%s' % (
        reimport_tag_prefix(version, namespace, patch_state=patch_state),
        reimport,
    )

def upload_tag(version, namespace):
    return '%s/upload/%s' % (namespace, git_dep14_tag(version))

def upstream_tag(version, namespace):
    return '%s/upstream/%s' % (namespace, git_dep14_tag(version))

def orphan_tag(version, namespace):
    return '%s/orphan/%s' % (namespace, git_dep14_tag(version))

def is_dir_3_0_quilt(_dir=None):
    _dir = _dir if _dir else '.'
    try:
        fmt, _ = run(['dpkg-source', '--print-format', _dir])
        if '3.0 (quilt)' in fmt:
            return True
    except CalledProcessError as e:
        try:
            with open(os.path.join(_dir, 'debian/source/format'), 'r') as f:
                for line in f:
                     if re.match(r'3.0 (.*)', line):
                         return True
        # `man dpkg-source` indicates no d/s/format implies 1.0
        except OSError:
            pass

    return False

def is_3_0_quilt(repo, commitish='HEAD'):
    with repo.temporary_worktree(commitish):
       return is_dir_3_0_quilt()

class GitUbuntuRepositoryFetchError(Exception):
    pass


def determine_quilt_series_path(pygit2_repo, treeish_obj):
    """Find the active quilt series file path in use.

    Look in the given tree for the Debian patch series file that is active
    according to the search algorithm described in dpkg-source(1). If none are
    found, return the default series path (again from dpkg-source(1)).

    :param pygit2.Repo pygit2_repo: repository to look in.
    :param pygit2.Object treeish_obj: object that peels to a pygit2.Tree.
    :returns: relative path to series file.
    :rtype: str
    """
    for series_name in ['debian.series', 'series']:
        try:
            series_path = posixpath.join('debian/patches', series_name)
            blob = follow_symlinks_to_blob(
                repo=pygit2_repo,
                treeish_object=treeish_obj,
                path=series_path,
            )
        except KeyError:
            continue  # try the next path using our search list
        return series_path  # series file blob found at this path

    logging.debug("Unable to find a series file in %r", treeish_obj)
    return 'debian/patches/series'  # default when no series file found


def quilt_env(pygit2_repo, treeish):
    """Find the appropriate quilt environment to use.

    Return the canonical environment that should be used when calling quilt.
    Since the series file doesn't necessarily always have the same name, a
    source tree is examined to determine the name and set QUILT_SERIES
    appropriately.

    This does not integrate any other environment variables. Only environment
    variables that influence quilt are returned.

    :param pygit2.Repo pygit2_repo: repository to look in.
    :param pygit2.Object treeish: object that peels to a pygit2.Tree.
    :returns: quilt-specific environment settings
    :rtype: dict
    """
    return {
        'QUILT_PATCHES': 'debian/patches',
        'QUILT_SERIES': determine_quilt_series_path(pygit2_repo, treeish),
        'QUILT_NO_DIFF_INDEX': '1',
        'QUILT_NO_DIFF_TIMESTAMPS': '1',
        'EDITOR': 'true',
    }


def datetime_to_signature_spec(datetime):
    """Convert a datetime to the time and offset required by a pygit2.Signature

    :param datetime datetime: the timezone-aware datetime to convert
    :rtype: tuple(int, int)
    :returns: the time since epoch and timezone offset in minutes as suitable
        for passing to the pygit2.Signature constructor parameters time and
        offset.
    """
    # Divide by 60 for seconds -> minutes
    offset_td = datetime.utcoffset()
    offset_mins = (
        int(offset_td.total_seconds()) // 60
        if offset_td
        else 0
    )

    return int(datetime.timestamp()), offset_mins


class HeadInfoItem(collections.namedtuple(
    'HeadInfoItem',
    [
        'version',
        'commit_time',
        'commit_id',
    ],
)):
    """Information associated with a single branch head

    :ivar str version: the package version found in debian/changelog at the
        branch head.
    :ivar int commit_time: the timestamp of the commit at the branch head,
        expressed as seconds since the Unix epoch.
    :ivar pygit2.Oid commit_id: the hash of the commit at the branch head.
    """
    pass


class GitUbuntuRepository:
    """A class for interacting with an importer git repository

    This class attempts to put all objects it manipulates in an
    'importer/' namespace. It also uses tags in one of three namespaces:
    'import/' for successfully imported published versions (these are
    created by the importer); 'upload/' for uploaded version by Ubuntu
    developers (these are understood by the importer and are aliased by
    import/ tags when succesfully imported); and 'orphan/' for published
    versions for which no parents can be found (these are also created
    by the importer).

    To access the underlying pygit2.Repository object, use the raw_repo
    property.
    """

    def __init__(
        self,
        local_dir,
        lp_user=None,
        fetch_proto=None,
        delete_on_close=True,
    ):
        """
        If fetch_proto is None, the default value from
        gitubuntu.__main__ will be used (top_level_defaults.proto).
        """
        if local_dir is None:
            self._local_dir = tempfile.mkdtemp()
        else:
            local_dir = os.path.abspath(local_dir)
            try:
                os.mkdir(local_dir)
            except FileExistsError:
                local_dir_list = os.listdir(local_dir)
                if local_dir_list and os.getenv(
                    'GIT_DIR',
                    '.git',
                ) not in local_dir_list:
                    logging.error('Specified directory %s must either '
                                  'be empty or have been previously '
                                  'imported to.', local_dir)
                    sys.exit(1)
            self._local_dir = local_dir

        self.raw_repo = pygit2.init_repository(self._local_dir)
        # We rely on raw_repo.workdir to be identical to self._local_dir to
        # avoid changing previous behaviour in the setting of GIT_WORK_TREE, so
        # assert that it is so. This may not be the case if the git repository
        # has a different workdir stored in its configuration or if the git
        # repository is a bare repository. We didn't handle these cases before
        # anyway, so with this assertion we can fail noisily and early.
        assert (
            os.path.normpath(self.raw_repo.workdir) ==
                os.path.normpath(self._local_dir)
        )

        # Since previous behaviour of this class depended on the state of the
        # environment at the time it was constructed, save this for later use
        # (for example in deriving the environment to use for calls to the git
        # CLI). This permits the behaviour to remain identical for now.
        # Eventually we can break previous behaviour and eliminate the need for
        # this. See also: gitubuntu.test_fixtures.repo; the handling of EMAIL
        # there could be made cleaner when this is cleaned up.
        self._initial_env = os.environ.copy()

        self.set_git_attributes()

        if lp_user:
            self._lp_user = lp_user
        else:
            try:
                self._lp_user, _ = self.git_run(
                    ['config', 'gitubuntu.lpuser'],
                    verbose_on_failure=False,
                )
                self._lp_user = self._lp_user.strip()
            except CalledProcessError:
                self._lp_user = None

        if fetch_proto is None:
            fetch_proto = top_level_defaults.proto

        self._fetch_proto = fetch_proto
        self._delete_on_close = delete_on_close

    def close(self):
        """Free resources associated with this instance

        If delete_on_close was True on instance construction, local_dir (as
        specified on instance construction) will be deleted.

        After this method is called, the instance is invalid and can no longer
        be used.
        """
        if self.raw_repo and self._delete_on_close:
            shutil.rmtree(self.local_dir)
        self.raw_repo = None

    def create_orphan_branch(self, branch_name, msg):
        if self.get_head_by_name(branch_name) is None:
            self.git_run(['checkout', '--orphan', branch_name])
            self.git_run(['commit', '--allow-empty', '-m', msg])
            self.git_run(['checkout', '--orphan', 'master'])

    @contextmanager
    def pristine_tar_branches(self, dist, namespace='pkg', create=True):
        """Context manager wrapping pristine-tar branch manipulation

        In this context, the repository pristine-tar branch will point to
        the pristine-tar branch for @dist distribution in @namespace.

        Because of our model, the distribution-pristine-tar branch may
        be a local branch (import-time) or a remote-tracking branch
        (build-time) and we need different behavior in both cases.
        Specifically, we want to affect the local branch's contents, but
        we cannot do that to a remote-tracking branch.

        Upon entry to the context, detect the former case (by doing a
        local only lookup first) and doing a branch rename there.
        Otherwise, create a new local branch.

        Upon exit, if a local branch had been found, rename pristine-tar
        back to the original name. Otherwise, simply delete the created
        pristine-tar branch.

        If a local branch named pristine-tar existed outside this
        context, it will be restored upon leaving the context.

        :param dist str One of 'ubuntu' or 'debian'
        :param namespace str Namespace under which Git refs are found
        :param create bool If an appropriate local pristine-tar Git
             branch does not exist, create one using the above algorithm.
        """
        pt_branch = '%s/importer/%s/pristine-tar' % (namespace, dist)
        old_pt_branch = self.raw_repo.lookup_branch('pristine-tar')
        old_pt_branch_commit = None
        if old_pt_branch:
            old_pt_branch_commit = old_pt_branch.peel(pygit2.Commit)
            old_pt_branch.delete()
        local_pt_branch = self.raw_repo.lookup_branch(pt_branch)
        remote_pt_branch = self.raw_repo.lookup_branch(
            pt_branch,
            pygit2.GIT_BRANCH_REMOTE,
        )
        if local_pt_branch:
            local_pt_branch.rename('pristine-tar')
        elif remote_pt_branch:
            self.raw_repo.create_branch(
                'pristine-tar',
                remote_pt_branch.peel(pygit2.Commit),
            )
        elif create:
            # This should only be possible when importing and the first
            # pristine-tar usage, create an orphan branch at the local
            # pt branch location and flag it for cleanup
            local_pt_branch = True
            self.create_orphan_branch(
                'pristine-tar',
                'Initial %s pristine-tar branch.' % dist,
            )
            if not self.raw_repo.lookup_branch('do-not-push'):
                self.create_orphan_branch(
                    'do-not-push',
                    'Initial upstream branch.',
                )
        try:
            yield
        except:
            raise
        finally:
            if local_pt_branch: # or create above
                self.raw_repo.lookup_branch('pristine-tar').rename(pt_branch)
            elif remote_pt_branch:
                self.raw_repo.lookup_branch('pristine-tar').delete()
            if old_pt_branch_commit:
                self.raw_repo.create_branch(
                    'pristine-tar',
                    old_pt_branch_commit,
                )

    def pristine_tar_list(self, dist, namespace='pkg'):
        """List tarballs stored in pristine-tar branch for @dist distribution in @namespace.

        If there is no pristine-tar branch, `pristine-tar list` returns
        nothing.

        :param dist str One of 'ubuntu' or 'debian'
        :param namespace str Namespace under which Git refs are found
        :rtype list(str)
        :returns List of orig tarball names stored in the pristine-tar
             branches
        """
        with self.pristine_tar_branches(dist, namespace, create=False):
            stdout, _ = run(['pristine-tar', 'list'])
            return stdout.strip().splitlines()

    def pristine_tar_extract(self, pkgname, version, dist=None, namespace='pkg'):
        '''Extract orig tarballs for a given package and upstream version

        This function will fail if the expected tarballs are already
        present by name in the parent directory. If, at some point, this
        is not desired, we would need to pass --git-force-create to
        gbp-buildpackage.

        The files, once created, are the responsibility of the caller to
        remove, if necessary.

        raises:
        - PristineTarNotFoundError if no suitable tarballs are found
        - MultiplePristineTarFoundError if multiple distinct suitable tarballs
          are found
        - CalledProcessError if gbp-buildpackage fails

        :param pkgname str Source package name
        :param version str Source package upstream version
        :param dist str One of 'ubuntu' or 'debian'
        :param namespace str Namespace under which Git refs are found
        :rtype list(str)
        :returns List of tarball paths that are now present on the
            filesystem. They will be in the parent directory.
        '''
        dists = [dist] if dist else ['debian', 'ubuntu']
        for dist in dists:
            main_tarball = '%s_%s.orig.tar' % (pkgname, version)

            all_tarballs = self.pristine_tar_list(dist, namespace)

            potential_main_tarballs = [tarball for tarball
                in all_tarballs if tarball.startswith(main_tarball)]
            if len(potential_main_tarballs) == 0:
                continue
            if len(potential_main_tarballs) > 1:
                # This will need some extension/flag for the case of there
                # being multiple imports with varying compression
                raise MultiplePristineTarFoundError(
                    'More than one pristine-tar tarball found for %s: %s' %
                    (version, potential_main_tarballs)
                )
            ext = os.path.splitext(potential_main_tarballs[0])[1]
            tarballs = []
            tarballs.append(
                os.path.join(os.path.pardir, potential_main_tarballs[0])
            )
            args = ['buildpackage', '--git-builder=/bin/true',
                   '--git-pristine-tar', '--git-ignore-branch',
                   '--git-upstream-tag=%s/upstream/%s/%%(version)s%s' %
                   (namespace, dist, ext)]
            # This will probably break if the component tarballs get
            # compressed differently, as each component tarball will show up
            # multiple times
            # Breaks may be too strong -- we will 'over cache' tarballs, and
            # then it's up to dpkg-buildpackage to use the 'correct' one
            potential_component_tarballs = {
                component_tarball_matches(tarball, pkgname, version).group('component') : tarball
                for tarball in all_tarballs
                if component_tarball_matches(tarball, pkgname, version)
            }
            tarballs.extend(map(lambda x : os.path.join(os.path.pardir, x),
                list(potential_component_tarballs.values()))
            )
            args.extend(map(lambda x : '--git-component=%s' % x,
                list(potential_component_tarballs.keys()))
            )
            with self.pristine_tar_branches(dist, namespace):
                run_gbp(args, env=self.env)
            return tarballs

        raise PristineTarNotFoundError(
            'No pristine-tar tarball found for %s' % version
        )

    def pristine_tar_exists(self, pkgname, version, namespace='pkg'):
        '''Report distributions that contain pristine-tar data for @version

        raises:
        - MultiplePristineTarFoundError if multiple distinct suitable tarballs
          are found

        :param pkgname str Source package name
        :param version str Source package upstream version
        :param namespace str Namespace under which Git refs are found
        :rtype list(str)
        :returns List of distribution names which contain a pristine-tar
             import for @pkgname and @version
        '''
        results = []
        for dist in ['debian', 'ubuntu']:
            main_tarball = '%s_%s.orig.tar' % (pkgname, version)

            all_tarballs = self.pristine_tar_list(dist, namespace)

            potential_main_tarballs = [tarball for tarball
                in all_tarballs if tarball.startswith(main_tarball)]
            if len(potential_main_tarballs) == 0:
                continue
            if len(potential_main_tarballs) > 1:
                # This will need some extension/flag for the case of there
                # being multiple imports with varying compression
                raise MultiplePristineTarFoundError(
                    'More than one pristine-tar tarball found for %s: %s' %
                    (version, potential_main_tarballs)
                )
            results.append(dist)

        return results

    def verify_pristine_tar(self, tarball_paths, dist, namespace='pkg'):
        '''Verify the pristine-tar data matches for a set of paths

        raises:
             PristineTarError - if a tarball has been imported before,
             but the contents of the new tarball do not match

        :param tarball_paths list(str) List of filesystem paths of orig
             tarballs to verify
        :param dist str One of 'ubuntu' or 'debian'
        :param namespace str Namespace under which Git refs are found
        :rtype bool
        :returns True if all paths in @tarball_paths exist in @dist's
             pristine-tar branch under @namespace and match the
             corresponding pristine-tar contents exactly
        '''
        all_tarballs = self.pristine_tar_list(dist, namespace)
        for path in tarball_paths:
            if os.path.basename(path) not in all_tarballs:
                break
            try:
                with self.pristine_tar_branches(dist, namespace):
                    # need to handle this not existing
                    run(['pristine-tar', 'verify', path])
            except CalledProcessError as e:
                raise PristineTarError(
                    'Tarball has already been imported to %s with '
                    'different contents' % dist
                )
        else:
            return True

        return False

    def set_git_attributes(self):
        git_attr_path = os.path.join(self.raw_repo.path,
                                     'info',
                                     'attributes'
                                    )
        try:
            # common-case: create an attributes file
            with open(git_attr_path, 'x') as f:
                f.write('* -ident\n')
                f.write('* -text\n')
                f.write('* -eol\n')
        except FileExistsError:
            # next-most common-case: attributes file already exists and
            # contains our desired value
            try:
                runq(['grep', '-q', '* -ident', git_attr_path])
            except CalledProcessError:
                # least-common case: attributes file exists, but does
                # not contain our desired value
                try:
                    with open(git_attr_path, 'a') as f:
                        f.write('* -ident\n')
                except:
                    # failed all three cases to set our desired value in
                    # attributes file
                    logging.exception('Unable to set \'* -ident\' in %s' %
                                      git_attr_path
                                     )
                    sys.exit(1)
            try:
                runq(['grep', '-q', '* -text', git_attr_path])
            except CalledProcessError:
                # least-common case: attributes file exists, but does
                # not contain our desired value
                try:
                    with open(git_attr_path, 'a') as f:
                        f.write('* -text\n')
                except:
                    # failed all three cases to set our desired value in
                    # attributes file
                    logging.exception('Unable to set \'* -text\' in %s' %
                                      git_attr_path
                                     )
                    sys.exit(1)
            try:
                runq(['grep', '-q', '* -eol', git_attr_path])
            except CalledProcessError:
                # least-common case: attributes file exists, but does
                # not contain our desired value
                try:
                    with open(git_attr_path, 'a') as f:
                        f.write('* -eol\n')
                except:
                    # failed all three cases to set our desired value in
                    # attributes file
                    logging.exception('Unable to set \'* -eol\' in %s' %
                                      git_attr_path
                                     )
                    sys.exit(1)

    def remote_exists(self, remote_name):
        # https://github.com/libgit2/pygit2/issues/671
        return any(remote.name == remote_name for remote in self.raw_repo.remotes)

    def _add_remote_by_fetch_url(
        self,
        remote_name,
        fetch_url,
        push_url=None,
        changelog_notes=False,
    ):
        """Add a remote by URL

        If a remote with the given name doesn't exist, then create it.
        Otherwise, do nothing.

        :param str remote_name: the name of the remote to create
        :param str fetch_url: the fetch URL for the remote
        :param str push_url: the push URL for the remote. If None, then a
            specific push URL will not be set.
        :param bool changelog_notes: if True, then a fetch refspec will be
            added to fetch changelog notes. This only makes sense for an
            official importer remote such as 'pkg'.
        :returns: None
        """
        if not self._fetch_proto:
            raise Exception('Cannot fetch using an object without a protocol')

        logging.debug('Adding %s as remote %s', fetch_url, remote_name)

        if not self.remote_exists(remote_name):
            self.raw_repo.remotes.create(
                remote_name,
                fetch_url,
                '+refs/heads/*:refs/remotes/%s/*' % remote_name,
            )
            # grab unreachable tags (orphans)
            self.raw_repo.remotes.add_fetch(
                remote_name,
                '+refs/tags/*:refs/tags/%s/*' % remote_name,
            )
            if changelog_notes:
                # The changelog notes are kept at refs/notes/commits on
                # Launchpad due to LP: #1871838 even though our standard place
                # for them is refs/notes/changelog.
                self.raw_repo.remotes.add_fetch(
                    remote_name,
                    '+refs/notes/commits:refs/notes/changelog',
                )
            if push_url:
                self.raw_repo.remotes.set_push_url(
                    remote_name,
                    push_url,
                )
            self.git_run(
                [
                     'config',
                     'remote.%s.tagOpt' % remote_name,
                     '--no-tags',
                ]
            )

    def _add_remote(self, remote_name, remote_url, changelog_notes=False):
        """Add a remote by URL location

        URL location means the part of the URL after the proto:// prefix. The
        protocol to be used will be determined by what was specified by the
        fetch_proto at class instance construction time. Separate fetch and
        push URL protocols will be automatically determined.

        If a remote with the given name doesn't exist, then create it.
        Otherwise, do nothing.

        :param str remote_name: the name of the remote to create
        :param str remote_url: the URL for the remote but with the proto://
            prefix missing.
        :param bool changelog_notes: if True, then a fetch refspec will be
            added to fetch changelog notes. This only makes sense for an
            official importer remote such as 'pkg'.
        :returns: None
        """
        if not self._fetch_proto:
            raise Exception('Cannot fetch using an object without a protocol')
        if not self._lp_user:
            raise RuntimeError("Cannot add remote without knowing lp_user")
        fetch_url = '%s://%s' % (self._fetch_proto, remote_url)
        push_url = 'ssh://%s@%s' % (self.lp_user, remote_url)

        self._add_remote_by_fetch_url(
            remote_name=remote_name,
            fetch_url=fetch_url,
            push_url=push_url,
            changelog_notes=changelog_notes,
        )

    def add_remote(
        self,
        pkgname,
        repo_owner,
        remote_name,
        changelog_notes=False,
    ):
        """Add a remote to the repository configuration
        :param str pkgname: the name of the source package reflected by this
            repository.
        :param str repo_owner: the name of the Launchpad user or team whose
            repository for the package will be pointed to by this new remote.
            If None, the default repository for the source package will be
            used.
        :param str remote_name: the name of the remote to add.
        :param bool changelog_notes: if True, then a fetch refspec will be
            added to fetch changelog notes. This only makes sense for an
            official importer remote such as 'pkg'.
        :returns: None
        """
        if not self._fetch_proto:
            raise Exception('Cannot fetch using an object without a protocol')
        if repo_owner:
            remote_url = ('git.launchpad.net/~%s/ubuntu/+source/%s' %
                          (repo_owner, pkgname))
        else:
            remote_url = ('git.launchpad.net/ubuntu/+source/%s' % pkgname)

        self._add_remote(
            remote_name=remote_name,
            remote_url=remote_url,
            changelog_notes=changelog_notes,
        )

    def add_remote_by_url(self, remote_name, fetch_url):
        if not self._fetch_proto:
            raise Exception('Cannot fetch using an object without a protocol')

        self._add_remote_by_fetch_url(remote_name, fetch_url)

    def add_base_remotes(self, pkgname, repo_owner=None):
        """Add the 'pkg' base remote to the repository configuration

        :param str pkgname: the name of the source package reflected by this
            repository.
        :param str repo_owner: the name of the Launchpad user or team whose
            repository for the package will be pointed to by this new remote.
            If None, the default repository for the source package will be
            used.
        :returns: None
        """
        self.add_remote(pkgname, repo_owner, 'pkg', changelog_notes=True)

    def add_lpuser_remote(self, pkgname):
        if not self._fetch_proto:
            raise Exception('Cannot add a fetch using an object without a protocol')
        if not self._lp_user:
            raise RuntimeError("Cannot add remote without knowing lp_user")
        remote_url = ('git.launchpad.net/~%s/ubuntu/+source/%s' %
                      (self.lp_user, pkgname))

        self._add_remote(remote_name=self.lp_user, remote_url=remote_url)
        # XXX: want a remote alias of 'lpme' -> self.lp_user
        # self.git_run(['config', 'url.%s.insteadof' % self.lp_user, 'lpme'])

    def fetch_remote(self, remote_name, verbose=False):
        # Does not seem to be working with https
        # https://github.com/libgit2/pygit2/issues/573
        # https://github.com/libgit2/libgit2/issues/3786
        # self.raw_repo.remotes[remote_name].fetch()
        kwargs = {}
        kwargs['verbose_on_failure'] = True
        if verbose:
            # If we are redirecting stdout/stderr to the console, we
            # do not need to have run() also emit it
            kwargs['verbose_on_failure'] = False
            kwargs['stdout'] = None
            kwargs['stderr'] = None
        try:
            logging.debug("Fetching remote %s", remote_name)
            self.git_run(
                args=['fetch', remote_name],
                env={'GIT_TERMINAL_PROMPT': '0',},
                **kwargs
            )
        except CalledProcessError:
            raise GitUbuntuRepositoryFetchError(
                "Unable to fetch remote %s" % remote_name
            )

    def fetch_base_remotes(self, verbose=False):
        self.fetch_remote(remote_name='pkg', verbose=verbose)

    def fetch_remote_refspecs(self, remote_name, refspecs, verbose=False):
        # Does not seem to be working with https
        # https://github.com/libgit2/pygit2/issues/573
        # https://github.com/libgit2/libgit2/issues/3786
        # self.raw_repo.remotes[remote_name].fetch()
        for refspec in refspecs:
            kwargs = {}
            kwargs['verbose_on_failure'] = True
            if verbose:
                # If we are redirecting stdout/stderr to the console, we
                # do not need to have run() also emit it
                kwargs['verbose_on_failure'] = False
                kwargs['stdout'] = None
                kwargs['stderr'] = None
            try:
                logging.debug(
                    "Fetching refspec %s from remote %s",
                    refspec,
                    remote_name,
                )
                self.git_run(
                    args=['fetch', remote_name, refspec],
                    env={'GIT_TERMINAL_PROMPT': '0',},
                    **kwargs,
                )
            except CalledProcessError:
                raise GitUbuntuRepositoryFetchError(
                    "Unable to fetch %s from remote %s" % (
                        refspecs,
                        remote_name,
                    )
                )

    def fetch_lpuser_remote(self, verbose=False):
        if not self._fetch_proto:
            raise Exception('Cannot fetch using an object without a protocol')
        if not self._lp_user:
            raise RuntimeError("Cannot fetch without knowing lp_user")
        self.fetch_remote(remote_name=self.lp_user, verbose=verbose)

    def copy_base_references(self, namespace):
        for ref in self.references:
            for (target_refs, source_refs) in [
                ('refs/heads/%s/' % namespace, 'refs/remotes/pkg/'),]:
                if ref.name.startswith(source_refs):
                    self.raw_repo.create_reference(
                        '%s/%s' % (target_refs, ref.name[len(source_refs):]),
                        ref.peel().id)

    def delete_branches_in_namespace(self, namespace):
        _local_branches = copy(self.local_branches)
        for head in self.local_branches:
            if head.branch_name.startswith(namespace):
                head.delete()

    def delete_tags_in_namespace(self, namespace):
        _tags = copy(self.tags)
        for ref in self.tags:
            if ref.name.startswith('refs/tags/%s' % namespace):
                ref.delete()

    @property
    def env(self):
        # Return a copy of the cached _derive_env method result so that the
        # caller cannot inadvertently modify our cached answer. Unfortunately
        # this leaks the lru_cache-ness of the _derive_env method to this
        # property getter, but this seems better than nothing.
        return dict(self._derive_env())

    @lru_cache()
    def _derive_env(self):
        """Determine what the git CLI environment should be

        This depends on the initial environment saved from the constructor and
        the paths associated with self.raw_repo, neither of which should change
        in the lifetime of this class instance.
        """
        return _derive_git_cli_env(
            self.raw_repo,
            initial_env=self._initial_env
        )

    @property
    def local_dir(self):
        """Base directory of this git repository (contains .git/)"""
        return self._local_dir

    @property
    def git_dir(self):
        """Same as cached object in the environment"""
        return self.raw_repo.path

    def _references(self, prefix=''):
        return [self.raw_repo.lookup_reference(r) for r in
                self.raw_repo.listall_references() if
                r.startswith(prefix)]

    def references_with_prefix(self, prefix):
        return self._references(prefix)

    @property
    def references(self):
        return self._references()

    @property
    def tags(self):
        return self._references('refs/tags')

    def _branches(self,
                  branch_type=pygit2.GIT_BRANCH_LOCAL | pygit2.GIT_BRANCH_REMOTE):
        branches = []
        if branch_type & pygit2.GIT_BRANCH_LOCAL:
            branches.extend([self.raw_repo.lookup_branch(b) for b in
                self.raw_repo.listall_branches(pygit2.GIT_BRANCH_LOCAL)])
        if branch_type & pygit2.GIT_BRANCH_REMOTE:
            branches.extend([self.raw_repo.lookup_branch(b, pygit2.GIT_BRANCH_REMOTE) for b in
                self.raw_repo.listall_branches(pygit2.GIT_BRANCH_REMOTE)])
        return branches

    @property
    def branches(self):
        return self._branches()

    @property
    def branch_names(self):
        return [b.branch_name for b in self.branches]

    @property
    def local_branches(self):
        return self._branches(pygit2.GIT_BRANCH_LOCAL)

    @property
    def local_branch_names(self):
        return [b.branch_name for b in self.local_branches]

    @property
    def remote_branches(self):
        return self._branches(pygit2.GIT_BRANCH_REMOTE)

    @property
    def remote_branch_names(self):
        return [b.branch_name for b in self.remote_branches]

    @property
    def lp_user(self):
        if not self._lp_user:
            raise RuntimeError("lp_user is not set")
        return self._lp_user

    def get_commitish(self, commitish):
        return self.raw_repo.revparse_single(commitish)

    def head_to_commit(self, head_name):
        return str(self.get_head_by_name(head_name).peel().id)

    def get_short_hash(self, hash):
        """Return an unambiguous but abbreviated form of a commit hash

        Note that the hash may still become ambiguous in the future.
        """
        stdout, _ = self.git_run(['rev-parse', '--short', hash])
        return stdout.strip()

    def git_run(self, args, env=None, **kwargs):
        """Run the git CLI with the provided arguments

        :param list(str) args: arguments to the git CLI
        :param dict env: additional environment variables to use
        :param dict **kwargs: further arguments to pass through to
            gitubuntu.run.run()
        :raises subprocess.CalledProcessError: if git exits non-zero
        :rtype: (str, str)
        :returns: stdout and stderr strings containing the subprocess output

        The environment used is based on the Python process' environment at the
        time this class instance was constructed.

        The GIT_DIR and GIT_WORK_TREE environment variables are set
        automatically based on the repository's existing location and settings.

        If env is set, then the environment to be used is updated with env
        before the call to git is made. This can override GIT_DIR,
        GIT_WORK_TREE, and anything else.
        """
        return git_run(
            pygit2_repo=self.raw_repo,
            args=args,
            initial_env=self._initial_env,
            update_env=env,
            **kwargs,
        )

    def garbage_collect(self):
        self.git_run(['gc'])

    def extract_file_from_treeish(self, treeish_string, filename):
        """extract a file from @treeish to a local file

        Arguments:
        treeish - SHA1 of treeish
        filename - file to extract from @treeish

        Returns a NamedTemporaryFile that is flushed but not rewound.
        """
        blob = follow_symlinks_to_blob(
            self.raw_repo,
            treeish_object=self.raw_repo.revparse_single(treeish_string),
            path=filename,
        )
        outfile = tempfile.NamedTemporaryFile()
        outfile.write(blob.data)
        outfile.flush()
        return outfile

    @lru_cache()
    def get_changelog_from_treeish(self, treeish_string):
        return Changelog.from_treeish(
            self.raw_repo,
            self.raw_repo.revparse_single(treeish_string),
        )

    def get_changelog_versions_from_treeish(self, treeish_string):
        """Extract current and prior versions from debian/changelog in a
        given @treeish_string

        Returns (None, None) if the treeish supplied is None or if
        'debian/changelog' does not exist in the treeish.

        Returns (current, previous) on success.
        """
        try:
            changelog = self.get_changelog_from_treeish(treeish_string)
        except KeyError:
            # If 'debian/changelog' does
            # not exist, then (None, None) is returned. KeyError propagates up
            # from Changelog's __init__.
            return None, None
        try:
            return changelog.version, changelog.previous_version
        except CalledProcessError:
            raise GitUbuntuChangelogError(
                'Cannot get changelog versions'
            )

    def get_changelog_distribution_from_treeish(self, treeish_string):
        """Extract targetted distribution from debian/changelog in a
        given treeish
        """

        if treeish_string is None:
            return None

        try:
            return self.get_changelog_from_treeish(treeish_string).distribution
        except (KeyError, CalledProcessError):
            raise GitUbuntuChangelogError(
                'Cannot get changelog distribution'
            )

    def get_changelog_srcpkg_from_treeish(self, treeish_string):
        """Extract srcpkg from debian/changelog in a given treeish
        """

        if treeish_string is None:
            return None

        try:
            return self.get_changelog_from_treeish(treeish_string).srcpkg
        except (KeyError, CalledProcessError):
            raise GitUbuntuChangelogError(
                'Cannot get changelog source package name'
            )

    def get_head_info(self, head_prefix, namespace):
        """Extract package versions at branch heads

        Extract the version from debian/changelog of all
        f'{namespace}/{head_prefix>/*' branches, excluding any branch that
        contains 'ubuntu/devel'.

        :param str namespace: the namespace under which git refs are found
        :param str head_prefix: the prefix to look for
        :rtype: dict(str, HeadInfoItem)
        :returns: a dictionary keyed by the namespaced branch name (ie. without
            a 'refs/heads/' prefix but with the namespace prefix, eg.
            'importer/ubuntu/focal-devel').
        """
        head_info = dict()
        for head in self.local_branches:
            prefix = '%s/%s' % (namespace, head_prefix)
            if not head.branch_name.startswith(prefix):
                continue
            if 'ubuntu/devel' in head.branch_name:
                continue
            version, _ = (
                self.get_changelog_versions_from_treeish(str(head.peel().id))
            )
            head_info[head.branch_name] = HeadInfoItem(
                version=version,
                commit_time=head.peel().commit_time,
                commit_id=head.peel().id,
            )

        return head_info

    def treeishs_identical(self, treeish_string1, treeish_string2):
        if treeish_string1 is None or treeish_string2 is None:
            return False
        _tree_obj1 = self.raw_repo.revparse_single(treeish_string1)
        _tree_id1 = _tree_obj1.peel(pygit2.Tree).id
        _tree_obj2 = self.raw_repo.revparse_single(treeish_string2)
        _tree_id2 = _tree_obj2.peel(pygit2.Tree).id
        return _tree_id1 == _tree_id2

    def get_head_by_name(self, name):
        try:
            return self.raw_repo.lookup_branch(name)
        except TypeError:
            return None

    def get_tag_reference(self, tag):
        """Return the tag object if it exists in the repository"""
        try:
            return self.raw_repo.lookup_reference('refs/tags/%s' % tag)
        except (KeyError, ValueError):
            return None

    def get_import_tag(
        self,
        version,
        namespace,
        patch_state=PatchState.UNAPPLIED,
    ):
        """
        Return the import tag matching the given specification.

        :param str version: the package version string to match
        :param str namespace: the namespace under which git refs are found
        :param PatchState patch_state: whether to look for unapplied or applied
            tags
        :returns: the matching import tag, or None if there is no match
        :rtype: pygit2.Reference or None
        """
        return self.get_tag_reference(
            import_tag(version, namespace, patch_state)
        )

    def get_reimport_tag(
        self,
        version,
        namespace,
        reimport,
        patch_state=PatchState.UNAPPLIED,
    ):
        """
        Return the reimport tag matching the given specification.

        :param str version: the package version string to match
        :param str namespace: the namespace under which git refs are found
        :param int reimport: the sequence number of the reimport tag
        :param PatchState patch_state: whether to look for unapplied or applied
            tags
        :returns: the matching reimport tag, or None if there is no match
        :rtype: pygit2.Reference or None
        """
        return self.get_tag_reference(
            reimport_tag(version, namespace, reimport, patch_state)
        )

    def get_all_reimport_tags(
        self,
        version,
        namespace,
        patch_state=PatchState.UNAPPLIED,
    ):
        """
        Return all reimport tags matching the given specification.

        :param str version: the package version string to match
        :param str namespace: the namespace under which git refs are found
        :param PatchState patch_state: whether to look for unapplied or applied
            tags
        :returns: matching reimport tags
        :rtype: sequence(pygit2.Reference)
        """
        return self.references_with_prefix(
            'refs/tags/%s/' % reimport_tag_prefix(
                version,
                namespace,
                patch_state,
            )
        )

    def get_upload_tag(self, version, namespace):
        """
        Return the upload tag matching the given specification.

        :param str version: the package version string to match
        :param str namespace: the namespace under which git refs are found
        :returns: the matching upload tag, or None if there is no match
        :rtype: pygit2.Reference or None
        """
        return self.get_tag_reference(upload_tag(version, namespace))

    def get_upstream_tag(self, version, namespace):
        """
        Return the upstream tag matching the given specification.

        :param str version: the package version string to match
        :param str namespace: the namespace under which git refs are found
        :returns: the matching upstream tag, or None if there is no match
        :rtype: pygit2.Reference or None
        """
        return self.get_tag_reference(upstream_tag(version, namespace))

    def get_orphan_tag(self, version, namespace):
        """
        Return the orphan tag matching the given specification.

        :param str version: the package version string to match
        :param str namespace: the namespace under which git refs are found
        :returns: the matching orphan tag, or None if there is no match
        :rtype: pygit2.Reference or None
        """
        return self.get_tag_reference(orphan_tag(version, namespace))

    def create_tag(self,
        commit_hash,
        tag_name,
        tag_msg,
        tagger=None,
    ):
        """Create a tag in the repository

        :param str commit_hash: the commit hash the tag will point to.
        :param str tag_name: the name of the tag to be created.
        :param str tag_msg: the text of the tag annotation.
        :param pygit2.Signature tagger: if supplied, use this signature in the
            created tag's "tagger" metadata. If not supplied, an arbitrary name
            and email address is used with the current time.
        :returns: None
        """
        if not tagger:
            tagger_time, tagger_offset = datetime_to_signature_spec(
                datetime.datetime.now(),
            )
            tagger = pygit2.Signature(
               gitubuntu.spec.SYNTHESIZED_COMMITTER_NAME,
               gitubuntu.spec.SYNTHESIZED_COMMITTER_EMAIL,
               tagger_time,
               tagger_offset,
            )

        logging.debug("Creating tag %s pointing to %s", tag_name, commit_hash)
        self.raw_repo.create_tag(
            tag_name,
            pygit2.Oid(hex=commit_hash),
            ObjectType.COMMIT,
            tagger,
            tag_msg,
        )

    def nearest_remote_branches(self, commit_hash, prefix=None,
        max_commits=100
    ):
        '''Return the set of remote branches nearest to @commit_hash

        This is a set of remote branch objects that are currently
        pointing at a commit, where that commit is the nearest ancestor
        to @commit_hash among the possible commits.

        If no such commit is found, an empty set is returned.

        Only consider remote branches that start with @prefix.

        Stop searching beyond the @max_commits'-th ancestor. Usually this method
        is only used as a heuristic that generally will never need to go too far
        back in history, and this avoids searching all the way back to the root
        commit, which may be a long way.
        '''

        # 1) cache all prefixed branch names by commit
        remote_heads_by_commit = collections.defaultdict(set)
        for b in self.remote_branches:
            if prefix is None or b.branch_name.startswith(prefix):
                remote_heads_by_commit[b.peel().id].add(b)

        # 2) walk from commit_hash backwards until a cached commit is found
        commits = self.raw_repo.walk(
            self.get_commitish(commit_hash).id,
            pygit2.GIT_SORT_TOPOLOGICAL,
        )
        for commit in itertools.islice(commits, max_commits):
            if commit.id not in remote_heads_by_commit:
                continue  # avoid creating a bunch of empty sets

            if remote_heads_by_commit[commit.id]:
                return remote_heads_by_commit[commit.id]

            # in the currently impossible (but permitted in this state) case
            # that the dictionary returned an empty set, we loop around again
            # which is what we want.

        return set()


    def nearest_tag(
        self,
        commitish_string,
        prefix,
        max_commits=100,
    ):
        # 1) cache all patterned tag names by commit
        pattern_tags_by_commit = collections.defaultdict(set)
        for t in self.tags:
            if t.name.startswith('refs/tags/' + prefix):
                pattern_tags_by_commit[t.peel(pygit2.Commit).id].add(t)

        commits = self.raw_repo.walk(
            self.get_commitish(commitish_string).id,
            pygit2.GIT_SORT_TOPOLOGICAL,
        )
        for commit in itertools.islice(commits, max_commits):
            if commit.id not in pattern_tags_by_commit:
                continue

            return pattern_tags_by_commit[commit.id].pop()

        return None

    @staticmethod
    def tag_to_pretty_name(tag):
        _, _, pretty_name = tag.name.partition('refs/tags/')
        return pretty_name

    def create_tracking_branch(self, branch_name, upstream_name, force=False):
        return self.raw_repo.create_branch(
            branch_name,
            self.raw_repo.lookup_branch(
                upstream_name,
                pygit2.GIT_BRANCH_REMOTE
            ).peel(pygit2.Commit),
            force
        )

    def checkout_commitish(self, commitish):
        # pygit2 checkout does not accept hashes
        # https://github.com/libgit2/pygit2/issues/412
        # self.raw_repo.checkout_tree(self.get_commitish(commitish))
        self.git_run(['checkout', commitish])

    def reset_commitish(self, commitish):
        # pygit2 checkout does not accept hashes
        # https://github.com/libgit2/pygit2/issues/412
        # self.checkout_tree(self.get_commitish(commitish))
        self.git_run(['reset', '--hard', commitish])

    def update_head_to_commit(self, head_name, commit_hash):
        try:
            self.raw_repo.lookup_branch(head_name).set_target(commit_hash)
        except AttributeError:
            self.raw_repo.create_branch(head_name,
                self.raw_repo.get(commit_hash)
            )

    def clean_repository_state(self):
        """Cleanup working tree"""
        runq(['git', 'checkout', '--orphan', 'master'],
             check=False, env=self.env)
        runq(['git', 'reset', '--hard'], env=self.env)
        runq(['git', 'clean', '-f', '-d'], env=self.env)

    def get_all_changelog_versions_from_treeish(self, treeish):
        changelog = self.get_changelog_from_treeish(treeish)
        return changelog.all_versions

    def annotated_tag(self, tag_name, commitish, force, msg=None):
        try:
            args = ['tag', '-a', tag_name, commitish]
            if force:
                args += ['-f']
            if msg is not None:
                args += ['-m', msg]
            self.git_run(args, stdin=None, stdout=None, stderr=None)
            version, _ = self.get_changelog_versions_from_treeish(commitish)
            logging.info('Created annotated tag %s for version %s' % (tag_name, version))
        except:
            logging.error('Unable to tag %s. Does it already exist (pass -f)?' %
                          tag_name
                         )
            raise

    def tag(self, tag_name, commitish, force):
        try:
            args = ['tag', tag_name, commitish]
            if force:
                args += ['-f']
            self.git_run(args)
            version, _ = self.get_changelog_versions_from_treeish(commitish)
            logging.info('Created tag %s for version %s' % (tag_name, version))
        except:
            logging.error('Unable to tag %s. Does it already exist (pass -f)?' %
                          tag_name
                         )
            raise

    def commit_source_tree(
        self,
        tree,
        parents,
        log_message,
        commit_date=None,
        author_date=None,
    ):
        """Commit a git tree with appropriate parents and message

        Given a git tree that contains a source package, create a matching
        commit using metadata derived from the tree as required according to
        the import specification.

        Commit metadata elements that are not specified as derived from the
        tree itself are required as parameters.

        :param pygit2.Oid tree: reference to the git tree in this repository
            that contains a debian/changelog file
        :param list(pygit2.Oid) parents: parent commits of the commit to be
            created
        :param bytes log_message: commit message
        :param datetime.datetime commit_date: the commit date to use (any
            sub-second part of the timestamp is truncated). If None, use the
            current date.
        :param datetime.datetime author_date: overrides the author date
            normally parsed from the changelog entry (i.e. for handling date
            parsing edge cases). Any sub-second part of the timestamp is
            truncated.
        :returns: reference to the created commit
        :rtype: pygit2.Oid
        """
        if commit_date is None:
            commit_date = datetime.datetime.now()

        commit_time, commit_offset = datetime_to_signature_spec(commit_date)
        changelog = self.get_changelog_from_treeish(str(tree))

        return self.raw_repo.create_commit(
            None,  # ref: do not update any ref
            pygit2.Signature(*changelog.git_authorship(author_date)),  # author
            pygit2.Signature(  # committer
                name=gitubuntu.spec.SYNTHESIZED_COMMITTER_NAME,
                email=gitubuntu.spec.SYNTHESIZED_COMMITTER_EMAIL,
                time=commit_time,
                offset=commit_offset,
            ),
            log_message,  # message
            tree,  # tree
            parents,  # parents
        )


    @classmethod
    def _create_replacement_tree_builder(cls, repo, treeish, sub_path):
        '''Create a replacement TreeBuilder

        Create a TreeBuilder based on an existing repository, top-level
        tree-ish and path inside that tree.

        A sub_path of '' is taken to mean a request for a replacement
        TreeBuilder for the top level tree.

        Returns a TreeBuilder object pre-populated with the previous contents.
        If the path did not previously exist in the tree-ish, then return an
        empty TreeBuilder instead.
        '''

        tree = treeish.peel(ObjectType.TREE)

        # Short path: sub_path == '' means want root
        if not sub_path:
            return repo.TreeBuilder(tree)

        try:
            tree_entry = tree[sub_path]
        except KeyError:
            # sub_path does not exist in tree, so return an empty TreeBuilder
            tree_builder = repo.TreeBuilder()
        else:
            # The tree entry must itself be a tree
            assert tree_entry.filemode == pygit2.GIT_FILEMODE_TREE
            sub_tree = repo.get(tree_entry.id).peel(ObjectType.TREE)
            tree_builder = repo.TreeBuilder(sub_tree)

        return tree_builder

    @classmethod
    def _add_missing_tree_dirs(cls, repo, top_path, top_tree_object, _sub_path=''):
        """
        Recursively add empty directories to a tree object

        Find empty directories under top_path and make sure that empty tree
        objects exist for them. If this means that the tree object must change,
        then a replacement tree object is created accordingly.

        repo: pygit2.Repository object
        top_path: path to the extracted contents of the tree
        top_tree_object: tree object
        _sub_path (internal): relative path for where we are for recursive call

        Returns None if oid unchanged, or oid if it changed.
        """

        # full path to our _sub_path, including top_path
        full_path = os.path.join(top_path, _sub_path)

        dir_list = os.listdir(full_path)
        if not dir_list:
            # directory is empty, so this is always the empty tree object
            return repo.TreeBuilder().write()

        # tree_builder is None if we don't need one yet, or is the replacement
        # for the tree object for this recursive call
        tree_builder = None
        for entry in dir_list:
            entry_path = os.path.join(full_path, entry)
            # We cannot use os.path.isdir() here as we don't want to recurse
            # down symlinks to directories.
            if stat.S_ISDIR(os.lstat(entry_path).st_mode):
                # this is a directory, so recurse down
                entry_oid = cls._add_missing_tree_dirs(
                    repo=repo,
                    top_path=top_path,
                    top_tree_object=top_tree_object,
                    _sub_path=os.path.join(_sub_path, entry),
                )
                if entry_oid:
                    # The recursive call reported a change to the tree, so we
                    # must adopt it in what we return to propogate the change
                    # upwards.
                    if tree_builder is None:
                        # There is no replacement in progress for this
                        # recursive call's tree object, so start one.
                        tree_builder = cls._create_replacement_tree_builder(
                            repo=repo,
                            treeish=top_tree_object,
                            sub_path=_sub_path,
                        )
                    # If the entry previous existed, remove it.
                    if tree_builder.get(entry):
                        tree_builder.remove(entry)
                    # Add the replacement tree entry
                    tree_builder.insert(           # (takes no kwargs)
                        entry,                     # name
                        entry_oid,                 # oid
                        pygit2.GIT_FILEMODE_TREE,  # attr
                    )

        if tree_builder is None:
            return None  # no changes
        else:
            return tree_builder.write()  # create replacement tree object

    @classmethod
    def dir_to_tree(cls, pygit2_repo, path, escape=False):
        """Create a git tree object from the given filesystem path

        :param pygit2.Repository pygit2_repo: the repository on which to
            operate. If you have a GitUbuntuRepository instance, you can use
            its raw_repo property.
        :param path: path to filesystem directory to be the root of the tree
        :param escape: if True, escape using escape_dot_git() first. This
                       mutates the provided filesystem tree.

        escape should be used when the directory being moved into git is
        directly from a source package, since the source package may contain
        files or directories named '.git' and these cannot otherwise be
        represented in a git tree object.

        escape should not be used if the directory has already been escaped
        previously. For example: if escape was previously used to move into a
        git tree object, and that git tree object has been extracted to a
        working directory for manipulation without unescaping, then escape
        should not be used again to move that result back into a git tree
        object.
        """
        if escape:
            escape_dot_git(path)
        # git expects the index file to not exist (in order to create a fresh
        # one), so create a temporary directory to put it in so we have a name
        # we can use safely.
        with tempfile.TemporaryDirectory() as index_dir:
            index_path = os.path.join(index_dir, 'index')
            def indexed_git_run(*args):
                return git_run(
                    pygit2_repo=pygit2_repo,
                    args=args,
                    work_tree_path=path,
                    index_path=index_path,
                )
            indexed_git_run('add', '-f', '-A')
            indexed_git_run('reset', 'HEAD', '--', '.git')
            indexed_git_run('reset', 'HEAD', '--', '.pc')
            tree_hash_str, _ = indexed_git_run('write-tree')
            tree_hash_str = tree_hash_str.strip()
            tree = pygit2_repo.get(tree_hash_str)

            # Add any empty directories that git did not import. Workaround for LP:
            # #1687057.
            replacement_oid = cls._add_missing_tree_dirs(
                repo=pygit2_repo,
                top_path=path,
                top_tree_object=tree,
            )
            if replacement_oid:
                # Empty directories had to be added
                return str(replacement_oid)  # return the replacement instead
            else:
                # No empty directories were added
                return tree_hash_str         # no replacement was needed

    @contextmanager
    def temporary_worktree(self, commitish, prefix=None):
        with tempfile.TemporaryDirectory(prefix=prefix) as tempdir:
            self.git_run(
                [
                    'worktree',
                    'add',
                    '--detach',
                    '--force',
                    tempdir,
                    commitish,
                ]
            )

            oldcwd = os.getcwd()
            os.chdir(tempdir)

            try:
                yield
            except:
                raise
            finally:
                os.chdir(oldcwd)

        self.git_run(['worktree', 'prune'])

    def tree_hash_after_command(self, commitish, cmd):
        with self.temporary_worktree(commitish):
            try:
                run(cmd)
            except CalledProcessError as e:
                logging.error("Unable to execute `%s`", ' '.join(cmd))
                raise

            run(["git", "add", "-f", ".",])
            tree_hash, _ = run(["git", "write-tree"])
            return tree_hash.strip()

    def tree_hash_subpath(self, treeish_string, path):
        """Get the tree hash for path at a given treeish

        Arguments:
        @treeish_string: a string Git treeish
        @path: a string path present in @treeish_string

        Returns:
        String hash of Git tree corresponding to @path in @treeish_string
        """
        tree_obj = self.raw_repo.revparse_single(treeish_string).peel(
            pygit2.Tree
        )
        return str(tree_obj[path].id)

    def paths_are_identical(self, treeish1_string, treeish2_string, path):
        """Determine if a given path is the same in two treeishs

        Arguments:
        @treeish1_string: a string Git treeish
        @treeish2_string: a string Git treeish
        @path: a string path present in @treeish1_string and @treeish2_string

        Returns:
        True, if @path is the same in @treeish1_string and @treeish2_string
        False, otherwise
        """
        try:
            subpath_tree_hash1 = self.tree_hash_subpath(
                treeish1_string,
                path,
            )
        except KeyError:
            # if the path does not exist in treeish
            subpath_tree_hash1 = None
        try:
            subpath_tree_hash2 = self.tree_hash_subpath(
                treeish2_string,
                path,
            )
        except KeyError:
            subpath_tree_hash2 = None

        return subpath_tree_hash1 == subpath_tree_hash2

    @lru_cache()
    def quilt_env(self, treeish):
        """Return a suitable environment for running quilt.

        This varies depending on the supplied commit since both
        debian/patches/series and debian/patches/debian.series may be valid.
        See dpkg-source(1) for details.

        The returned environment includes all necessary variables by
        combining self.env with the needed quilt-specific environment.

        :param pygit.Object treeish: object that peels to the pygit2.Tree on
            which quilt will operate.
        :rtype: dict
        :returns: an environment suitable for running quilt.
        """
        env = self.env.copy()
        env.update(quilt_env(self.raw_repo, treeish))
        return env

    def quilt_env_from_treeish_str(self, treeish_str):
        """Return a suitable environment for running quilt.

        This is a thin wrapper around quilt_env() that works with a treeish hex
        string instead of directly with a treeish object.

        :param str treeish_str: the hash of the tree on which quilt will
            operate, in hex.
        :rtype: dict
        :returns: an environment suitable for running quilt.
        """
        return self.quilt_env(self.raw_repo.get(treeish_str))

    def is_patches_applied(self, commit_hash, regenerated_pc_path):
        # first see if quilt push -a would do anything to
        # differentiate between applied and unapplied
        with self.temporary_worktree(commit_hash):
            try:
                run_quilt(
                    ['push', '-a'],
                    env=self.quilt_env_from_treeish_str(commit_hash),
                )
                # False if in an unapplied state, which is signified by
                # successful push (rc=0)
                return False
            except CalledProcessError as e:
                # non-zero return might be an error or it might mean no
                # patches exist
                if e.returncode == 1:
                    # an error may occur if we need to recreate the .pc
                    # first
                    try:
                        # the first quilt push may have created a .pc/
                        shutil.rmtree('.pc')
                        shutil.copytree(
                            regenerated_pc_path,
                            '.pc',
                        )
                    except FileNotFoundError:
                        # if there was no .pc directory, then the first
                        # quilt push failure was a real error
                        raise e

                    try:
                        run_quilt(
                            ['push', '-a'],
                            env=self.quilt_env_from_treeish_str(commit_hash),
                        )
                        # False if in an unapplied state
                        return False
                    except CalledProcessError as e:
                        # True if in a patches-applied state or
                        # there are no patches to apply
                        if e.returncode == 2:
                            return True
                        else:
                            raise
                # True if in a patches-applied state or there are
                # no patches to apply
                elif e.returncode == 2:
                    return True
                else:
                    raise

    def _maybe_quiltify_tree_hash(self, commit_hash):
        """Determine if quiltify is needed and yield the quiltify'd tree hash

        The imported patches-applied trees do not contain .pc
        directories. To determine if an additional quilt patch is
        necessary, we have to first regenerate the .pc directory, then
        see if dpkg-source --commit generates a new quilt patch.

        In order for dpkg-source --commit to function, we need to know
        if the commit we are building is patches-unapplied or
        patches-applied. In the latter case, we can build the commit
        directly after copying the regenerated .pc directory. In the
        former case, we do not want to copy the regenerated .pc
        directory, as dpkg-source will do this for us, as it applies the
        current patches. We determine if patches are applied or
        unapplied by relying on `quilt push -a`'s exit status at
        @commit_hash.

        This is a common method used by multiple callers.

        Arguments:
        @commit_hash: a string Git commit hash

        Returns:
        String tree hash of quiltify'ing @commit_hash.
        If no quiltify is needed, the return value is @commit_hash's
        tree hash
        """
        commit_tree_hash = str(
            self.raw_repo.get(commit_hash).peel(pygit2.Tree).id
        )
        if not is_3_0_quilt(self, commit_hash):
            return commit_tree_hash
        # the tarballs need to be in the parent directory from where
        # we need the orig tarballs for quilt and dpkg-source
        # but suppress any logging
        logger = logging.getLogger()
        oldLevel = logger.getEffectiveLevel()
        logger.setLevel(logging.WARNING)
        tarballs = gitubuntu.build.fetch_orig(
            orig_search_list=gitubuntu.build.derive_orig_search_list_from_args(
                self,
                commitish=commit_hash,
                for_merge=False,
                no_pristine_tar=False,
            ),
            changelog=Changelog.from_treeish(
                self.raw_repo,
                self.raw_repo.get(commit_hash)
            ),
        )
        logger.setLevel(oldLevel)
        # run dpkg-source
        with tempfile.TemporaryDirectory() as tempdir:
            # copy the generated tarballs
            new_tarballs = []
            for tarball in tarballs:
                new_tarballs.append(shutil.copy(tarball, tempdir))
            tarballs = new_tarballs

            # create a nested temporary directory where we will recreate
            # the .pc directory
            with tempfile.TemporaryDirectory(prefix=tempdir+'/') as ttempdir:
                oldcwd = os.getcwd()
                os.chdir(ttempdir)

                for tarball in tarballs:
                    run(['tar', '-x', '--strip-components=1', '-f', tarball,])

                # need the debia/patches
                shutil.copytree(
                    os.path.join(self.local_dir, 'debian',),
                    'debian',
                )

                # generate the equivalent .pc directory
                run_quilt(
                    ['push', '-a'],
                    env=self.quilt_env_from_treeish_str(commit_hash),
                    rcs=[2],
                )

                regenerated_pc_path = os.path.join(tempdir, '.pc')

                if os.path.exists(".pc"):
                    shutil.copytree(
                        '.pc',
                        regenerated_pc_path,
                    )

                os.chdir(oldcwd)

            patches_applied = self.is_patches_applied(
                commit_hash,
                regenerated_pc_path,
            )

            with self.temporary_worktree(commit_hash, prefix=tempdir+'/'):
                # we only need to copy the generated .pc directory
                # if we are building a patches-applied tree, which
                # we determine by comparing our current tree hash to
                # the generated tree hash.
                if patches_applied:
                    try:
                        shutil.copytree(
                            regenerated_pc_path,
                            '.pc',
                        )
                    except FileNotFoundError:
                        # it is possible no quilt patches exist yet
                        pass

                fixup_patch_path = os.path.join(
                    'debian',
                    'patches',
                    'git-ubuntu-fixup.patch'
                )

                if os.path.exists(fixup_patch_path):
                    raise ValueError(
                        "A quilt patch with the name git-ubuntu-fixup.patch "
                        "already exists in %s" % commit_hash
                    )

                run(
                    [
                        'dpkg-source',
                        '--commit',
                        '.',
                        'git-ubuntu-fixup.patch',
                    ],
                    env=self.quilt_env_from_treeish_str(commit_hash),
                )

                # do not want the .pc directory in the resulting
                # treeish
                if os.path.exists('.pc'):
                    shutil.rmtree('.pc')

                if os.path.exists(fixup_patch_path):
                    # dpkg-source uses debian/changelog to generate some
                    # fields. We do not know yet if the changelog has
                    # been updated, so elide that section of comments.
                    with open(fixup_patch_path, 'r+') as f:
                        for line in f:
                            if '---' in line:
                                break
                        text = """Description: git-ubuntu generated quilt fixup patch
TODO: Put a short summary on the line above and replace this paragraph
with a longer explanation of this change. Complete the meta-information
with other relevant fields (see below for details).
---\n"""
                        for line in f:
                            text += line
                        f.seek(0)
                        f.write(text)
                        f.truncate()

                    # If we are on a patches-unapplied tree, then we
                    # need to reset ourselves back to @commit_hash with
                    # our new patch.
                    # In order for this to be buildable, we have to
                    # reverse-apply our patch, to undo the git-commited
                    # upstream changes.
                    if not patches_applied:
                        run(['git', 'add', '-f', 'debian/patches',])
                        # if any patches add files that are untracked,
                        # remove them
                        run(['git', 'clean', '-f', '-d',])
                        # reset all the other files to their status in
                        # HEAD
                        run(['git', 'checkout', commit_hash, '--', '*',])
                        with open(fixup_patch_path, 'rb') as f:
                            run(['patch', '-Rp1',], input=f.read())

                    return self.dir_to_tree(self.raw_repo, '.')
                else:
                    return commit_tree_hash

    def maybe_quiltify_tree_hash(self, commitish_string):
        """Determine if quiltify is needed and return the quiltify'd tree hash

        See _maybe_quiltify_tree_hash for details.

        Arguments:
        @commitish_string: a string Git commitish

        Returns:
        String tree hash of quiltify'ing @commitish_string.
        If no quiltify is needed, the return value is the tree hash of
        @commitish_string.
        """
        commit_hash = str(
            self.get_commitish(commitish_string).peel(pygit2.Commit).id
        )
        return self._maybe_quiltify_tree_hash(commit_hash)

    def maybe_changelogify_tree_hash(self, commit_hash):
        """Determine if changelogify is needed and yield the changelogify'd tree hash

        Given a commit, we need to detect if the user has inserted a
        changelog entry relative to a published version for the purpose
        of test builds.

        Arguments:
        @commit_hash: a string Git commit hash

        Returns:
        String tree hash of changelogify'ing @commit_hash.
        If no changelogify is needed, the return value is the tree hash of
        @commit_hash.
        """
        commit_tree_hash = str(
            self.raw_repo.get(commit_hash).peel(pygit2.Tree).id
        )

        # one of these are the "base" pkg that @commit_hash's changes
        # are based on
        remote_tag = self.nearest_tag(
            commit_hash,
            prefix='pkg/',
        )
        remote_branch = derive_target_branch(
            self,
            commit_hash,
        )

        assert remote_tag or remote_branch

        if remote_tag:
            if remote_branch:
                try:
                    self.git_run(
                        [
                            'merge-base',
                            '--is-ancestor',
                            remote_tag.name,
                            remote_branch,
                        ],
                        verbose_on_failure=False,
                    )
                    parent_ref = remote_branch
                except CalledProcessError as e:
                    if e.returncode == 1:
                        parent_ref = remote_tag.name
                    else:
                        raise
            else:
                parent_ref = remote_tag.name
        else:
            parent_ref = remote_branch

        # If there are any changes relative to parent_ref but there are
        # not any changelog changes, insert a snapshot changelog entry,
        # starting from parent_ref, and return the resulting tree hash.
        if str(self.raw_repo.revparse_single(parent_ref).peel(
            pygit2.Tree
        ).id) != commit_tree_hash and self.paths_are_identical(
            parent_ref,
            commit_hash,
            'debian/changelog',
        ):
            with self.temporary_worktree(commit_hash):
                run_gbp(
                    [
                        'dch',
                        '--snapshot',
                        '--ignore-branch',
                        '--since=%s' % str(parent_ref),
                    ],
                    env=self.env,
                )
                return self.dir_to_tree(self.raw_repo, '.')

        # otherwise, return @commit_hash's tree hash
        return commit_tree_hash

    def quiltify_and_changelogify_tree_hash(self, commitish_string):
        """Given a commitish, possibly quiltify and changelogify its tree

        Definitions:
            quiltify: generate a quilt patch from untracked upstream
            changes
            changelogify: generate a snapshot changelog entry if any
            changes exist, and no new changelog entry yet exists

        Arguments:
        @commitish_string: string Git commitish

        Returns:
        string Git tree hash of quiltify-ing and changelogify-ing
        @commitish_string, if needed
        if neither quiltify or changelogify are needed, return
        @commitish_string's tree hash
        """
        commit_hash = str(
            self.get_commitish(commitish_string).peel(pygit2.Commit).id
        )
        quiltify_tree_hash = self._maybe_quiltify_tree_hash(commit_hash)
        changelogify_tree_hash = self.maybe_changelogify_tree_hash(commit_hash)

        quiltify_tree_obj = self.raw_repo.get(quiltify_tree_hash)
        changelogify_tree_obj = self.raw_repo.get(changelogify_tree_hash)

        # There are multiple ways to solve this problem, but the
        # simplest is to use a TreeBuilder to merge the quiltify tree
        # with the changelog from the changelogify tree
        # top-level TreeBuilder
        tb = self.raw_repo.TreeBuilder(quiltify_tree_obj)
        te = tb.get('debian')
        # TreeBuilder for debian/
        dtb = self.raw_repo.TreeBuilder(self.raw_repo.get(te.id))
        dtb.insert( # does not take kwargs
            'changelog',                                   # name
            changelogify_tree_obj['debian/changelog'].oid, # oid
            pygit2.GIT_FILEMODE_BLOB,                      # attr
        )
        # insert can replace
        tb.insert( # does not take kwargs
            'debian',                 # name
            dtb.write(),              # oid
            pygit2.GIT_FILEMODE_TREE, # attr
        )
        return str(tb.write())

    def find_ubuntu_merge_base(
        self,
        ubuntu_commitish,
    ):
        """Find the Ubuntu merge point for a given Ubuntu version

        :param ubuntu_commitish str A commitish describing the latest
            Ubuntu commit

        :rtype str
        :returns Commit hash of import of Debian version
            @ubuntu_commitish is based on. The imported Debian version
            must be an ancestor of @ubuntu_commitish. If no suitable
            commit is found, an empty string is returned.
        """
        merge_base_tag = None

        # obtain the nearest imported Debian version per the changelog
        for version in self.get_all_changelog_versions_from_treeish(
            ubuntu_commitish,
        ):
            # extract corresponding Debian version
            debian_parts, _ = gitubuntu.versioning.split_version_string(
                version
            )
            expected_debian_version = "".join(debian_parts)

            # We do not currently handle the case of a Debian version
            # being reimported. I think the proper way to support that
            # would be to add a parameter to `git ubuntu merge` for the
            # user to tell us which reimport tag is the one the Ubuntu
            # delta is based on.
            merge_base_tag = self.get_import_tag(
                expected_debian_version,
                'pkg',
            )

            if merge_base_tag:
                assert not self.get_all_reimport_tags(
                    expected_debian_version,
                    'pkg',
                )
                break

        if not merge_base_tag:
            logging.error(
                "Unable to find an import tag for any Debian version "
                "in %s:debian/changelog.",
                ubuntu_commitish,
            )
            return ''

        merge_base_commit_hash = str(merge_base_tag.peel(pygit2.Commit).id)

        try:
            self.git_run(
                [
                    'merge-base',
                    '--is-ancestor',
                    merge_base_commit_hash,
                    ubuntu_commitish,
                ],
                verbose_on_failure=False,
            )
        except CalledProcessError as e:
            if e.returncode != 1:
                raise
            logging.error(
                "Found an import tag for %s (commit: %s), but it is "
                "not an ancestor of %s.",
                expected_debian_version,
                merge_base_commit_hash,
                ubuntu_commitish,
            )
            return ''

        return merge_base_commit_hash