~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005-2010 Canonical Ltd

    controldir,

    graph as _mod_graph,

    inventory,

    transport,

from bzrlib.lock import LogicalLockResult

from bzrlib.revision import CURRENT_REVISION

from bzrlib.symbol_versioning import (

    deprecated_passed,

    DEPRECATED_PARAMETER,

    )

# TODO: Modifying the conflict objects or their type is currently nearly

# impossible as there is no clear relationship between the working tree format

# and the conflict list file format.

class WorkingTree(bzrlib.mutabletree.MutableTree,

    controldir.ControlComponent):

 

    :ivar basedir: The root of the tree on disk. This is a unicode path object

        (as opposed to a URL).

    @property

    def user_transport(self):

        return self.bzrdir.user_transport

 

    @property

    def control_transport(self):

        return self._transport

 

    def _get_check_refs(self):

        """Return the references needed to perform a check of this tree.

        

        The default implementation returns no refs, and is only suitable for

        trees that have no local caching and can commit on ghosts at any time.

 

        :seealso: bzrlib.check for details about check_refs.

        """

        return []

 

    def open_containing_paths(file_list, default_directory='.',

        canonicalize=True, apply_view=True):

        """Open the WorkingTree that contains a set of paths.

 

        Fail if the paths given are not all in a single tree.

 

        This is used for the many command-line interfaces that take a list of

        any number of files and that require they all be in the same tree.

        """

        # recommended replacement for builtins.internal_tree_files

        if file_list is None or len(file_list) == 0:

            tree = WorkingTree.open_containing(default_directory)[0]

            # XXX: doesn't really belong here, and seems to have the strange

            # side effect of making it return a bunch of files, not the whole

            # tree -- mbp 20100716

            if tree.supports_views() and apply_view:

                view_files = tree.views.lookup_view()

                if view_files:

                    file_list = view_files

                    view_str = views.view_display_str(view_files)

                    note("Ignoring files outside view. View is %s" % view_str)

            return tree, file_list

        tree = WorkingTree.open_containing(file_list[0])[0]

        return tree, tree.safe_relpath_files(file_list, canonicalize,

            apply_view=apply_view)

 

    def safe_relpath_files(self, file_list, canonicalize=True, apply_view=True):

        """Convert file_list into a list of relpaths in tree.

 

        :param self: A tree to operate on.

        :param file_list: A list of user provided paths or None.

        :param apply_view: if True and a view is set, apply it or check that

            specified files are within it

        :return: A list of relative paths.

        :raises errors.PathNotChild: When a provided path is in a different self

            than self.

        """

        if file_list is None:

            return None

        if self.supports_views() and apply_view:

            view_files = self.views.lookup_view()

        else:

            view_files = []

        new_list = []

        # self.relpath exists as a "thunk" to osutils, but canonical_relpath

        # doesn't - fix that up here before we enter the loop.

        if canonicalize:

            fixer = lambda p: osutils.canonical_relpath(self.basedir, p)

        else:

            fixer = self.relpath

        for filename in file_list:

            relpath = fixer(osutils.dereference_path(filename))

            if view_files and not osutils.is_inside_any(view_files, relpath):

                raise errors.FileOutsideView(filename, view_files)

            new_list.append(relpath)

        return new_list

 

    @staticmethod

        t = transport.get_transport(location)

        iterator = bzrdir.BzrDir.find_bzrdirs(t, evaluate=evaluate,

        return [tr for tr in iterator if tr is not None]

        """See Tree.get_file_with_stat."""

        if filtered and self.supports_content_filtering():

        my_file = self.get_file(file_id, path=path, filtered=filtered)

        try:

            return my_file.read()

        finally:

            my_file.close()

        if filtered and self.supports_content_filtering():

        maybe_file_parent_keys = []

        for parent_id in self.get_parent_ids():

            try:

                parent_tree = self.revision_tree(parent_id)

            except errors.NoSuchRevisionInTree:

                parent_tree = self.branch.repository.revision_tree(parent_id)

            parent_tree.lock_read()

            try:

                if file_id not in parent_tree:

                    continue

                ie = parent_tree.inventory[file_id]

                if ie.kind != 'file':

                    # Note: this is slightly unnecessary, because symlinks and

                    # directories have a "text" which is the empty text, and we

                    # know that won't mess up annotations. But it seems cleaner

                    continue

                parent_text_key = (file_id, ie.revision)

                if parent_text_key not in maybe_file_parent_keys:

                    maybe_file_parent_keys.append(parent_text_key)

            finally:

                parent_tree.unlock()

        graph = _mod_graph.Graph(self.branch.repository.texts)

        heads = graph.heads(maybe_file_parent_keys)

        file_parent_keys = []

        for key in maybe_file_parent_keys:

            if key in heads:

                file_parent_keys.append(key)

 

        # Now we have the parents of this content

        annotator = self.branch.repository.texts.get_annotator()

        text = self.get_file_text(file_id)

        this_key =(file_id, default_revision)

        annotator.add_special_text(this_key, file_parent_keys, text)

        annotations = [(key[-1], line)

                       for key, line in annotator.annotate_flat(this_key)]

        return annotations

            merges_bytes = self._transport.get_bytes('pending-merges')

            for l in osutils.split_lines(merges_bytes):

        # XXX: this returns the on-disk size; it should probably return the

        # canonical size

        if file_id is None:

            # For unversioned files on win32, we just assume they are not

            # executable

            return False

            return self._file_content_summary(path, stat_result)

    def _file_content_summary(self, path, stat_result):

        size = stat_result.st_size

        executable = self._is_executable_from_path_and_stat(path, stat_result)

        # try for a stat cache lookup

        return ('file', size, executable, self._sha_from_stat(

            path, stat_result))

 

                          merge_type=None, force=False):

        merger = Merger(self.branch, this_tree=self)

        # check that there are no local alterations

        if not force and self.has_changes():

            raise errors.UncommittedChanges(self)

        if to_revision is None:

            to_revision = _mod_revision.ensure_null(branch.last_revision())

        merger.other_rev_id = to_revision

        if _mod_revision.is_null(merger.other_rev_id):

            raise errors.NoCommits(branch)

        self.branch.fetch(branch, last_revision=merger.other_rev_id)

        merger.other_basis = merger.other_rev_id

        merger.other_tree = self.branch.repository.revision_tree(

            merger.other_rev_id)

        merger.other_branch = branch

        if from_revision is None:

            merger.find_base()

        else:

            merger.set_base_revision(from_revision, branch)

        if merger.base_rev_id == merger.other_rev_id:

            raise errors.PointlessMerge

        merger.backup_files = False

        if merge_type is None:

            merger.merge_type = Merge3Merger

        else:

            merger.merge_type = merge_type

        merger.set_interesting_files(None)

        merger.show_base = False

        merger.reprocess = False

        conflicts = merger.do_merge()

        merger.set_pending()

            branch.BranchReferenceFormat().initialize(tree_bzrdir,

                target_branch=new_branch)

        wt = tree_bzrdir.create_workingtree(_mod_revision.NULL_REVISION)

        child_inv = inventory.Inventory(root_id=None)

        :param include_root: if True, return an entry for the root

            if from_dir_id is None:

                # Directory not versioned

                return

                dir_ie = inv[from_dir_id]

                if dir_ie.kind == 'directory':

                    f_ie = dir_ie.children.get(f)

                else:

                    f_ie = None

                    # we may not have found this file, because of a unicode

                    # issue, or because the directory was actually a symlink.

    def move(self, from_paths, to_dir=None, after=False):

            raise TypeError('You must supply a target directory')

                self._move_entry(WorkingTree._RenameEntry(

                    entry.to_rel, entry.from_id,

            # if file is missing in the inventory maybe it's in the basis_tree

            basis_tree = self.branch.basis_tree()

            from_id = basis_tree.path2id(from_rel)

            if from_id is None:

                raise errors.BzrRenameFailedError(from_rel,to_rel,

                    errors.NotVersionedError(path=str(from_rel)))

            # put entry back in the inventory so we can rename it

            from_entry = basis_tree.inventory[from_id].copy()

            inv.add(from_entry)

        else:

            from_entry = inv[from_id]

                                pb=None,

                    basis_root_id = basis_tree.get_root_id()

                    new_root_id = new_basis_tree.get_root_id()

                    if basis_root_id != new_root_id:

                        self.set_root_id(new_root_id)

        others match against only the last component.  Patterns starting

        with '!' are ignore exceptions.  Exceptions take precedence

        over regular patterns and cause the filename to not be ignored.

            self._ignoreglobster = globbing.ExceptionGlobster(self.get_ignore_list())

        """Lock the tree for reading.

 

        This also locks the branch, and can be unlocked via self.unlock().

 

        :return: A bzrlib.lock.LogicalLockResult.

        """

            self._control_files.lock_read()

            return LogicalLockResult(self.unlock)

        """See MutableTree.lock_tree_write, and WorkingTree.unlock.

 

        :return: A bzrlib.lock.LogicalLockResult.

        """

            self._control_files.lock_write()

            return LogicalLockResult(self.unlock)

        """See MutableTree.lock_write, and WorkingTree.unlock.

 

        :return: A bzrlib.lock.LogicalLockResult.

        """

            self._control_files.lock_write()

            return LogicalLockResult(self.unlock)

        f = self._transport.get('inventory')

        try:

            result = self._deserialize(f)

        finally:

            f.close()

            xml = self.branch.repository._get_inventory_xml(new_revision)

                inv = self.branch.repository._serializer.read_inventory_from_string(

                    xml, new_revision)

        f = self._transport.get('inventory')

        try:

            result = self._deserialize(f)

        finally:

            f.close()

        all_files = set() # specified and nested files 

        if to_file is None:

            to_file = sys.stdout

 

        files_to_backup = []

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

                    if self.path2id(relpath):

                        all_files.add(relpath)

                        # Files which are not versioned

                        files_to_backup.append(relpath)

                all_files.add(filename)

        files = list(all_files)

            for (file_id, path, content_change, versioned, parent_id, name,

                 kind, executable) in self.iter_changes(self.basis_tree(),

                     include_unchanged=True, require_versioned=False,

                     want_unversioned=True, specific_files=files):

                if versioned[0] == False:

                    # The record is unknown or newly added

                    files_to_backup.append(path[1])

                elif (content_change and (kind[1] is not None) and

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

                    # Versioned and changed, but not deleted, and still

                    # in one of the dirs to be deleted.

                    files_to_backup.append(path[1])

        def backup(file_to_backup):

            backup_name = self.bzrdir.generate_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)

                    # XXX: Really should be a more abstract reporter interface

                    kind_ch = osutils.kind_marker(self.kind(fid))

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

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

                            message = backup(f)

                        if f in files_to_backup:

                            message = backup(f)

                        else:

                            osutils.delete_any(abs_path)

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

               pb=None, report_changes=False):

                if last_revision != _mod_revision.NULL_REVISION:

    _marker = object()

 

    def update(self, change_reporter=None, possible_transports=None,

               revision=None, old_tip=_marker):