/brz/remove-bazaar

# Copyright (C) 2008-2018 Jelmer Vernooij <jelmer@jelmer.uk>

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""An adapter between a Git index and a Bazaar Working Tree"""

import itertools

from collections import defaultdict

import errno

from dulwich.ignore import (

    IgnoreFilterManager,

    )

from dulwich.config import ConfigFile as GitConfigFile

from dulwich.file import GitFile, FileLocked

from dulwich.index import (

    Index,

    SHA1Writer,

    build_index_from_tree,

    index_entry_from_path,

    index_entry_from_stat,

    FLAG_STAGEMASK,

    read_submodule_head,

    validate_path,

    write_index_dict,

    )

from dulwich.object_store import (

    tree_lookup_path,

    )

from dulwich.objects import (

    S_ISGITLINK,

    )

import os

import posixpath

import stat

import sys

from .. import (

    branch as _mod_branch,

    conflicts as _mod_conflicts,

    errors,

    controldir as _mod_controldir,

    globbing,

    ignores,

    lock,

    merge,

    osutils,

    revision as _mod_revision,

    trace,

    transport as _mod_transport,

    tree,

    urlutils,

    workingtree,

    )

from ..decorators import (

    only_raises,

    )

from ..mutabletree import (

    BadReferenceTarget,

    MutableTree,

    )

from .dir import (

    LocalGitDir,

    )

from .tree import (

    MutableGitIndexTree,

    )

from .mapping import (

    mode_kind,

    )

class GitWorkingTree(MutableGitIndexTree, workingtree.WorkingTree):

    """A Git working tree."""

    def __init__(self, controldir, repo, branch):

        MutableGitIndexTree.__init__(self)

        basedir = controldir.root_transport.local_abspath('.')

        self.basedir = osutils.realpath(basedir)

        self.controldir = controldir

        self.repository = repo

        self.store = self.repository._git.object_store

        self.mapping = self.repository.get_mapping()

        self._branch = branch

        self._transport = self.repository._git._controltransport

        self._format = GitWorkingTreeFormat()

        self.index = None

        self._index_file = None

        self.views = self._make_views()

        self._rules_searcher = None

        self._detect_case_handling()

        self._reset_data()

    def supports_tree_reference(self):

        return True

    def supports_rename_tracking(self):

        return False

    def _read_index(self):

        self.index = Index(self.control_transport.local_abspath('index'))

        self._index_dirty = False

    def _get_submodule_index(self, relpath):

        if not isinstance(relpath, bytes):

            raise TypeError(relpath)

        try:

            info = self._submodule_info()[relpath]

        except KeyError:

            index_path = os.path.join(self.basedir, relpath.decode('utf-8'), '.git', 'index')

        else:

            index_path = self.control_transport.local_abspath(

                posixpath.join('modules', info[1].decode('utf-8'), 'index'))

        return Index(index_path)

    def lock_read(self):

        """Lock the repository for read operations.

        :return: A breezy.lock.LogicalLockResult.

        """

        if not self._lock_mode:

            self._lock_mode = 'r'

            self._lock_count = 1

            self._read_index()

        else:

            self._lock_count += 1

        self.branch.lock_read()

        return lock.LogicalLockResult(self.unlock)

    def _lock_write_tree(self):

        if not self._lock_mode:

            self._lock_mode = 'w'

            self._lock_count = 1

            try:

                self._index_file = GitFile(

                    self.control_transport.local_abspath('index'), 'wb')

            except FileLocked:

                raise errors.LockContention('index')

            self._read_index()

        elif self._lock_mode == 'r':

            raise errors.ReadOnlyError(self)

        else:

            self._lock_count += 1

    def lock_tree_write(self):

        self.branch.lock_read()

        try:

            self._lock_write_tree()

            return lock.LogicalLockResult(self.unlock)

        except BaseException:

            self.branch.unlock()

            raise

    def lock_write(self, token=None):

        self.branch.lock_write()

        try:

            self._lock_write_tree()

            return lock.LogicalLockResult(self.unlock)

        except BaseException:

            self.branch.unlock()

            raise

    def is_locked(self):

        return self._lock_count >= 1

    def get_physical_lock_status(self):

        return False

    def break_lock(self):

        try:

            self.control_transport.delete('index.lock')

        except errors.NoSuchFile:

            pass

        self.branch.break_lock()

    @only_raises(errors.LockNotHeld, errors.LockBroken)

    def unlock(self):

        if not self._lock_count:

            return lock.cant_unlock_not_held(self)

        try:

            self._cleanup()

            self._lock_count -= 1

            if self._lock_count > 0:

                return

            if self._index_file is not None:

                if self._index_dirty:

                    self._flush(self._index_file)

                    self._index_file.close()

                else:

                    # Something else already triggered a write of the index

                    # file by calling .flush()

                    self._index_file.abort()

                self._index_file = None

            self._lock_mode = None

            self.index = None

        finally:

            self.branch.unlock()

    def _cleanup(self):

        pass

    def _detect_case_handling(self):

        try:

            self._transport.stat(".git/cOnFiG")

        except errors.NoSuchFile:

            self.case_sensitive = True

        else:

            self.case_sensitive = False

    def get_transform(self, pb=None):

        from ..transform import TreeTransform

        return TreeTransform(self, pb=pb)

    def merge_modified(self):

        return {}

    def set_merge_modified(self, modified_hashes):

        raise errors.UnsupportedOperation(self.set_merge_modified, self)

    def set_parent_trees(self, parents_list, allow_leftmost_as_ghost=False):

        self.set_parent_ids([p for p, t in parents_list])

    def _set_merges_from_parent_ids(self, rhs_parent_ids):

        try:

            merges = [self.branch.lookup_bzr_revision_id(

                revid)[0] for revid in rhs_parent_ids]

        except errors.NoSuchRevision as e:

            raise errors.GhostRevisionUnusableHere(e.revision)

        if merges:

            self.control_transport.put_bytes(

                'MERGE_HEAD', b'\n'.join(merges),

                mode=self.controldir._get_file_mode())

        else:

            try:

                self.control_transport.delete('MERGE_HEAD')

            except errors.NoSuchFile:

                pass

    def set_parent_ids(self, revision_ids, allow_leftmost_as_ghost=False):

        """Set the parent ids to revision_ids.

        See also set_parent_trees. This api will try to retrieve the tree data

        for each element of revision_ids from the trees repository. If you have

        tree data already available, it is more efficient to use

        set_parent_trees rather than set_parent_ids. set_parent_ids is however

        an easier API to use.

        :param revision_ids: The revision_ids to set as the parent ids of this

            working tree. Any of these may be ghosts.

        """

        with self.lock_tree_write():

            self._check_parents_for_ghosts(

                revision_ids, allow_leftmost_as_ghost=allow_leftmost_as_ghost)

            for revision_id in revision_ids:

                _mod_revision.check_not_reserved_id(revision_id)

            revision_ids = self._filter_parent_ids_by_ancestry(revision_ids)

            if len(revision_ids) > 0:

                self.set_last_revision(revision_ids[0])

            else:

                self.set_last_revision(_mod_revision.NULL_REVISION)

            self._set_merges_from_parent_ids(revision_ids[1:])

    def get_parent_ids(self):

        """See Tree.get_parent_ids.

        This implementation reads the pending merges list and last_revision

        value and uses that to decide what the parents list should be.

        """

        last_rev = _mod_revision.ensure_null(self._last_revision())

        if _mod_revision.NULL_REVISION == last_rev:

            parents = []

        else:

            parents = [last_rev]

        try:

            merges_bytes = self.control_transport.get_bytes('MERGE_HEAD')

        except errors.NoSuchFile:

            pass

        else:

            for l in osutils.split_lines(merges_bytes):

                revision_id = l.rstrip(b'\n')

                parents.append(

                    self.branch.lookup_foreign_revision_id(revision_id))

        return parents

    def check_state(self):

        """Check that the working state is/isn't valid."""

        pass

    def remove(self, files, verbose=False, to_file=None, keep_files=True,

               force=False):

        """Remove nominated files from the working tree metadata.

        :param files: File paths relative to the basedir.

        :param keep_files: If true, the files will also be kept.

        :param force: Delete files and directories, even if they are changed

            and even if the directories are not empty.

        """

        if not isinstance(files, list):

            files = [files]

        if to_file is None:

            to_file = sys.stdout

        def backup(file_to_backup):

            abs_path = self.abspath(file_to_backup)

            backup_name = self.controldir._available_backup_name(

                file_to_backup)

            osutils.rename(abs_path, self.abspath(backup_name))

            return "removed %s (but kept a copy: %s)" % (

                file_to_backup, backup_name)

        # Sort needed to first handle directory content before the directory

        files_to_backup = []

        all_files = set()

        def recurse_directory_to_add_files(directory):

            # Recurse directory and add all files

            # so we can check if they have changed.

            for parent_info, file_infos in self.walkdirs(directory):

                for relpath, basename, kind, lstat, fileid, kind in file_infos:

                    # Is it versioned or ignored?

                    if self.is_versioned(relpath):

                        # Add nested content for deletion.

                        all_files.add(relpath)

                    else:

                        # Files which are not versioned

                        # should be treated as unknown.

                        files_to_backup.append(relpath)

        with self.lock_tree_write():

            for filepath in files:

                # Get file name into canonical form.

                abspath = self.abspath(filepath)

                filepath = self.relpath(abspath)

                if filepath:

                    all_files.add(filepath)

                    recurse_directory_to_add_files(filepath)

            files = list(all_files)

            if len(files) == 0:

                return  # nothing to do

            # Sort needed to first handle directory content before the

            # directory

            files.sort(reverse=True)

            # Bail out if we are going to delete files we shouldn't

            if not keep_files and not force:

                for change in self.iter_changes(

                        self.basis_tree(), include_unchanged=True,

                        require_versioned=False, want_unversioned=True,

                        specific_files=files):

                    if change.versioned[0] is False:

                        # The record is unknown or newly added

                        files_to_backup.append(change.path[1])

                        files_to_backup.extend(

                            osutils.parent_directories(change.path[1]))

                    elif (change.changed_content and (change.kind[1] is not None)

                            and osutils.is_inside_any(files, change.path[1])):

                        # Versioned and changed, but not deleted, and still

                        # in one of the dirs to be deleted.

                        files_to_backup.append(change.path[1])

                        files_to_backup.extend(

                            osutils.parent_directories(change.path[1]))

            for f in files:

                if f == '':

                    continue

                try:

                    kind = self.kind(f)

                except errors.NoSuchFile:

                    kind = None

                abs_path = self.abspath(f)

                if verbose:

                    # having removed it, it must be either ignored or unknown

                    if self.is_ignored(f):

                        new_status = 'I'

                    else:

                        new_status = '?'

                    kind_ch = osutils.kind_marker(kind)

                    to_file.write(new_status + '       ' + f + kind_ch + '\n')

                if kind is None:

                    message = "%s does not exist" % (f, )

                else:

                    if not keep_files:

                        if f in files_to_backup and not force:

                            message = backup(f)

                        else:

                            if kind == 'directory':

                                osutils.rmtree(abs_path)

                            else:

                                osutils.delete_any(abs_path)

                            message = "deleted %s" % (f,)

                    else:

                        message = "removed %s" % (f,)

                self._unversion_path(f)

                # print only one message (if any) per file.

                if message is not None:

                    trace.note(message)

            self._versioned_dirs = None

    def smart_add(self, file_list, recurse=True, action=None, save=True):

        if not file_list:

            file_list = [u'.']

        # expand any symlinks in the directory part, while leaving the

        # filename alone

        # only expanding if symlinks are supported avoids windows path bugs

        if self.supports_symlinks():

            file_list = list(map(osutils.normalizepath, file_list))

        conflicts_related = set()

        for c in self.conflicts():

            conflicts_related.update(c.associated_filenames())

        added = []

        ignored = {}

        user_dirs = []

        def call_action(filepath, kind):

            if filepath == '':

                return

            if action is not None:

                parent_path = posixpath.dirname(filepath)

                parent_id = self.path2id(parent_path)

                parent_ie = self._get_dir_ie(parent_path, parent_id)

                file_id = action(self, parent_ie, filepath, kind)

                if file_id is not None:

                    raise workingtree.SettingFileIdUnsupported()

        with self.lock_tree_write():

            for filepath in osutils.canonical_relpaths(

                    self.basedir, file_list):

                filepath, can_access = osutils.normalized_filename(filepath)

                if not can_access:

                    raise errors.InvalidNormalization(filepath)

                abspath = self.abspath(filepath)

                kind = osutils.file_kind(abspath)

                if kind in ("file", "symlink"):

                    (index, subpath) = self._lookup_index(

                        filepath.encode('utf-8'))

                    if subpath in index:

                        # Already present

                        continue

                    call_action(filepath, kind)

                    if save:

                        self._index_add_entry(filepath, kind)

                    added.append(filepath)

                elif kind == "directory":

                    (index, subpath) = self._lookup_index(

                        filepath.encode('utf-8'))

                    if subpath not in index:

                        call_action(filepath, kind)

                    if recurse:

                        user_dirs.append(filepath)

                else:

                    raise errors.BadFileKindError(filename=abspath, kind=kind)

            for user_dir in user_dirs:

                abs_user_dir = self.abspath(user_dir)

                if user_dir != '':

                    try:

                        transport = _mod_transport.get_transport_from_path(

                            abs_user_dir)

                        _mod_controldir.ControlDirFormat.find_format(transport)

                        subtree = True

                    except errors.NotBranchError:

                        subtree = False

                    except errors.UnsupportedFormatError:

                        subtree = False

                else:

                    subtree = False

                if subtree:

                    trace.warning('skipping nested tree %r', abs_user_dir)

                    continue

                for name in os.listdir(abs_user_dir):

                    subp = os.path.join(user_dir, name)

                    if (self.is_control_filename(subp) or

                            self.mapping.is_special_file(subp)):

                        continue

                    ignore_glob = self.is_ignored(subp)

                    if ignore_glob is not None:

                        ignored.setdefault(ignore_glob, []).append(subp)

                        continue

                    abspath = self.abspath(subp)

                    kind = osutils.file_kind(abspath)

                    if kind == "directory":

                        user_dirs.append(subp)

                    else:

                        (index, subpath) = self._lookup_index(

                            subp.encode('utf-8'))

                        if subpath in index:

                            # Already present

                            continue

                        if subp in conflicts_related:

                            continue

                        call_action(subp, kind)

                        if save:

                            self._index_add_entry(subp, kind)

                        added.append(subp)

            return added, ignored

    def has_filename(self, filename):

        return osutils.lexists(self.abspath(filename))

    def _iter_files_recursive(self, from_dir=None, include_dirs=False,

                              recurse_nested=False):

        if from_dir is None:

            from_dir = u""

        if not isinstance(from_dir, str):

            raise TypeError(from_dir)

        encoded_from_dir = self.abspath(from_dir).encode(osutils._fs_enc)

        for (dirpath, dirnames, filenames) in os.walk(encoded_from_dir):

            dir_relpath = dirpath[len(self.basedir):].strip(b"/")

            if self.controldir.is_control_filename(

                    dir_relpath.decode(osutils._fs_enc)):

                continue

            for name in list(dirnames):

                if self.controldir.is_control_filename(

                        name.decode(osutils._fs_enc)):

                    dirnames.remove(name)

                    continue

                relpath = os.path.join(dir_relpath, name)

                if not recurse_nested and self._directory_is_tree_reference(relpath.decode(osutils._fs_enc)):

                    dirnames.remove(name)

                if include_dirs:

                    try:

                        yield relpath.decode(osutils._fs_enc)

                    except UnicodeDecodeError:

                        raise errors.BadFilenameEncoding(

                            relpath, osutils._fs_enc)

                    if not self.is_versioned(relpath.decode(osutils._fs_enc)):

                        dirnames.remove(name)

            for name in filenames:

                if self.mapping.is_special_file(name):

                    continue

                if self.controldir.is_control_filename(

                        name.decode(osutils._fs_enc, 'replace')):

                    continue

                yp = os.path.join(dir_relpath, name)

                try:

                    yield yp.decode(osutils._fs_enc)

                except UnicodeDecodeError:

                    raise errors.BadFilenameEncoding(

                        yp, osutils._fs_enc)

    def extras(self):

        """Yield all unversioned files in this WorkingTree.

        """

        with self.lock_read():

            index_paths = set(

                [p.decode('utf-8') for p, i in self._recurse_index_entries()])

            all_paths = set(self._iter_files_recursive(include_dirs=False))

            return iter(all_paths - index_paths)

    def _gather_kinds(self, files, kinds):

        """See MutableTree._gather_kinds."""

        with self.lock_tree_write():

            for pos, f in enumerate(files):

                if kinds[pos] is None:

                    fullpath = osutils.normpath(self.abspath(f))

                    try:

                        kind = osutils.file_kind(fullpath)

                    except OSError as e:

                        if e.errno == errno.ENOENT:

                            raise errors.NoSuchFile(fullpath)

                    if f != '' and self._directory_is_tree_reference(f):

                        kind = 'tree-reference'

                    kinds[pos] = kind

    def flush(self):

        if self._lock_mode != 'w':

            raise errors.NotWriteLocked(self)

        # TODO(jelmer): This shouldn't be writing in-place, but index.lock is

        # already in use and GitFile doesn't allow overriding the lock file

        # name :(

        f = open(self.control_transport.local_abspath('index'), 'wb')

        # Note that _flush will close the file

        self._flush(f)

    def _flush(self, f):

        try:

            shaf = SHA1Writer(f)

            write_index_dict(shaf, self.index)

            shaf.close()

        except BaseException:

            f.abort()

            raise

        self._index_dirty = False

    def get_file_mtime(self, path):

        """See Tree.get_file_mtime."""

        try:

            return self._lstat(path).st_mtime

        except OSError as e: