/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"""

from __future__ import absolute_import

import itertools

from collections import defaultdict

import errno

from dulwich.ignore import (

    IgnoreFilterManager,

    )

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 (

    conflicts as _mod_conflicts,

    errors,

    controldir as _mod_controldir,

    globbing,

    ignores,

    lock,

    merge,

    osutils,

    revision as _mod_revision,

    trace,

    transport as _mod_transport,

    tree,

    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 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):

        if from_dir is None:

            from_dir = u""

        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 include_dirs:

                    try:

                        yield relpath.decode(osutils._fs_enc)

                    except UnicodeDecodeError:

                        raise errors.BadFilenameEncoding(

                            relpath, osutils._fs_enc)

                    if not self._has_dir(relpath):

                        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:

            if e.errno == errno.ENOENT:

                raise errors.NoSuchFile(path)

            raise

    def is_ignored(self, filename):

        r"""Check whether the filename matches an ignore pattern.

        If the file is ignored, returns the pattern which caused it to

        be ignored, otherwise None.  So this can simply be used as a

        boolean if desired."""

        if getattr(self, '_global_ignoreglobster', None) is None:

            ignore_globs = set()

            ignore_globs.update(ignores.get_runtime_ignores())

            ignore_globs.update(ignores.get_user_ignores())

            self._global_ignoreglobster = globbing.ExceptionGlobster(

                ignore_globs)

        match = self._global_ignoreglobster.match(filename)

        if match is not None:

            return match

        try:

            if self.kind(filename) == 'directory':

                filename += '/'

        except errors.NoSuchFile:

            pass

        filename = filename.lstrip('/')