~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2006-2010 Canonical Ltd

import time

    commit,

    ui,

                           ReusingTransform, CantMoveRoot,

from bzrlib.progress import ProgressPhase

    def __init__(self, tree, pb=None,

        :param pb: ignored

        if parent is None:

            raise ValueError("Parent trans-id may not be None")

    def fixup_new_roots(self):

        """Reinterpret requests to change the root directory

 

        Instead of creating a root directory, or moving an existing directory,

        all the attributes and children of the new root are applied to the

        existing root directory.

 

        This means that the old root trans-id becomes obsolete, so it is

        recommended only to invoke this after the root trans-id has become

        irrelevant.

        """

        new_roots = [k for k, v in self._new_parent.iteritems() if v is

                     ROOT_PARENT]

        if len(new_roots) < 1:

            return

        if len(new_roots) != 1:

            raise ValueError('A tree cannot have two roots!')

        if self._new_root is None:

            self._new_root = new_roots[0]

            return

        old_new_root = new_roots[0]

        # TODO: What to do if a old_new_root is present, but self._new_root is

        #       not listed as being removed? This code explicitly unversions

        #       the old root and versions it with the new file_id. Though that

        #       seems like an incomplete delta

 

        # unversion the new root's directory.

        file_id = self.final_file_id(old_new_root)

        if old_new_root in self._new_id:

            self.cancel_versioning(old_new_root)

        else:

            self.unversion_file(old_new_root)

        # if, at this stage, root still has an old file_id, zap it so we can

        # stick a new one in.

        if (self.tree_file_id(self._new_root) is not None and

            self._new_root not in self._removed_id):

            self.unversion_file(self._new_root)

        self.version_file(file_id, self._new_root)

 

        # Now move children of new root into old root directory.

        # Ensure all children are registered with the transaction, but don't

        # use directly-- some tree children have new parents

        list(self.iter_tree_children(old_new_root))

        # Move all children of new root into old root directory.

        for child in self.by_parent().get(old_new_root, []):

            self.adjust_path(self.final_name(child), self._new_root, child)

 

        # Ensure old_new_root has no directory.

        if old_new_root in self._new_contents:

            self.cancel_creation(old_new_root)

        else:

            self.delete_contents(old_new_root)

 

        # prevent deletion of root directory.

        if self._new_root in self._removed_contents:

            self.cancel_deletion(self._new_root)

 

        # destroy path info for old_new_root.

        del self._new_parent[old_new_root]

        del self._new_name[old_new_root]

 

        kind = self.tree_kind(trans_id)

        if kind is not None:

            self._removed_contents.add(trans_id)

        changed_kind = (t for t in changed_kind

                        if self.tree_kind(t) != self.final_kind(t))

        :return: None if the file does not exist/has no contents.  (It is

            conceivable that a path would be created without the corresponding

            contents insertion command)

            return None

    def _check_malformed(self):

        conflicts = self.find_conflicts()

        if len(conflicts) != 0:

            raise MalformedTransform(conflicts=conflicts)

 

            kind = self.final_kind(trans_id)

            if kind is None:

                if self.final_kind(trans_id) != "file":

            if self.tree_kind(trans_id) is None:

                kind = self.final_kind(trans_id)

            kind = self.final_kind(parent_id)

            if self.final_kind(trans_id) is not None:

                return True

        to_kind = self.final_kind(to_trans_id)

        The tree is a snapshot, and altering the TreeTransform will invalidate

        it.

    def commit(self, branch, message, merge_parents=None, strict=False,

               timestamp=None, timezone=None, committer=None, authors=None,

               revprops=None, revision_id=None):

        """Commit the result of this TreeTransform to a branch.

 

        :param branch: The branch to commit to.

        :param message: The message to attach to the commit.

        :param merge_parents: Additional parent revision-ids specified by

            pending merges.

        :param strict: If True, abort the commit if there are unversioned

            files.

        :param timestamp: if not None, seconds-since-epoch for the time and

            date.  (May be a float.)

        :param timezone: Optional timezone for timestamp, as an offset in

            seconds.

        :param committer: Optional committer in email-id format.

            (e.g. "J Random Hacker <jrandom@example.com>")

        :param authors: Optional list of authors in email-id format.

        :param revprops: Optional dictionary of revision properties.

        :param revision_id: Optional revision id.  (Specifying a revision-id

            may reduce performance for some non-native formats.)

        :return: The revision_id of the revision committed.

        """

        self._check_malformed()

        if strict:

            unversioned = set(self._new_contents).difference(set(self._new_id))

            for trans_id in unversioned:

                if self.final_file_id(trans_id) is None:

                    raise errors.StrictCommitFailed()

 

        revno, last_rev_id = branch.last_revision_info()

        if last_rev_id == _mod_revision.NULL_REVISION:

            if merge_parents is not None:

                raise ValueError('Cannot supply merge parents for first'

                                 ' commit.')

            parent_ids = []

        else:

            parent_ids = [last_rev_id]

            if merge_parents is not None:

                parent_ids.extend(merge_parents)

        if self._tree.get_revision_id() != last_rev_id:

            raise ValueError('TreeTransform not based on branch basis: %s' %

                             self._tree.get_revision_id())

        revprops = commit.Commit.update_revprops(revprops, branch, authors)

        builder = branch.get_commit_builder(parent_ids,

                                            timestamp=timestamp,

                                            timezone=timezone,

                                            committer=committer,

                                            revprops=revprops,

                                            revision_id=revision_id)

        preview = self.get_preview_tree()

        list(builder.record_iter_changes(preview, last_rev_id,

                                         self.iter_changes()))

        builder.finish_inventory()

        revision_id = builder.commit(message)

        branch.set_last_revision_info(revno + 1, revision_id)

        return revision_id

 

    def __init__(self, tree, limbodir, pb=None,

        :param pb: ignored

        self._creation_mtime = None

                delete_any(path)

                delete_any(self._limbodir)

                    delete_any(self._deletiondir)

        if limbo_name is None:

            limbo_name = self._generate_limbo_path(trans_id)

            self._limbo_files[trans_id] = limbo_name

    def _generate_limbo_path(self, trans_id):

        """Generate a limbo path using the trans_id as the relative path.

 

        This is suitable as a fallback, and when the transform should not be

        sensitive to the path encoding of the limbo directory.

        """

        self._needs_rename.add(trans_id)

        return pathjoin(self._limbodir, trans_id)

 

            if previous_parent != parent:

                self._limbo_children[previous_parent].remove(trans_id)

            if previous_parent != parent or previous_name != name:

                del self._limbo_children_names[previous_parent][previous_name]

            for descendant in self._limbo_descendants(trans_id):

                desc_path = self._limbo_files[descendant]

                desc_path = new_path + desc_path[len(old_path):]

                self._limbo_files[descendant] = desc_path

 

    def _limbo_descendants(self, trans_id):

        """Return the set of trans_ids whose limbo paths descend from this."""

        descendants = set(self._limbo_children.get(trans_id, []))

        for descendant in list(descendants):

            descendants.update(self._limbo_descendants(descendant))

        return descendants

        self._set_mtime(name)

    def _set_mtime(self, path):

        """All files that are created get the same mtime.

 

        This time is set by the first object to be created.

        """

        if self._creation_mtime is None:

            self._creation_mtime = time.time()

        os.utime(path, (self._creation_mtime, self._creation_mtime))

 

    def __init__(self, tree, pb=None):

        :returns: The file kind or None if the file does not exist

            return None

        except errors.NoSuchFile:

            return None

    def _generate_limbo_path(self, trans_id):

        """Generate a limbo path using the final path if possible.

 

        This optimizes the performance of applying the tree transform by

        avoiding renames.  These renames can be avoided only when the parent

        directory is already scheduled for creation.

 

        If the final path cannot be used, falls back to using the trans_id as

        the relpath.

        """

        parent = self._new_parent.get(trans_id)

        # if the parent directory is already in limbo (e.g. when building a

        # tree), choose a limbo name inside the parent, to reduce further

        # renames.

        use_direct_path = False

        if self._new_contents.get(parent) == 'directory':

            filename = self._new_name.get(trans_id)

            if filename is not None:

                if parent not in self._limbo_children:

                    self._limbo_children[parent] = set()

                    self._limbo_children_names[parent] = {}

                    use_direct_path = True

                # the direct path can only be used if no other file has

                # already taken this pathname, i.e. if the name is unused, or

                # if it is already associated with this trans_id.

                elif self._case_sensitive_target:

                    if (self._limbo_children_names[parent].get(filename)

                        in (trans_id, None)):

                        use_direct_path = True

                else:

                    for l_filename, l_trans_id in\

                        self._limbo_children_names[parent].iteritems():

                        if l_trans_id == trans_id:

                            continue

                        if l_filename.lower() == filename.lower():

                            break

                    else:

                        use_direct_path = True

 

        if not use_direct_path:

            return DiskTreeTransform._generate_limbo_path(self, trans_id)

 

        limbo_name = pathjoin(self._limbo_files[parent], filename)

        self._limbo_children[parent].add(trans_id)

        self._limbo_children_names[parent][filename] = trans_id

        return limbo_name

 

            self._check_malformed()

                kind = self.final_kind(trans_id)

                if kind is None:

                    delete_path = os.path.join(self._deletiondir, trans_id)

                    mover.pre_delete(full_path, delete_path)

                elif (trans_id in self._new_name

                      or trans_id in self._new_parent):

                    except errors.TransformRenameFailed, e:

                    except errors.TransformRenameFailed, e:

    def __init__(self, tree, pb=None, case_sensitive=True):

            return None

        try:

            return self._tree.kind(file_id)

        except errors.NoSuchFile:

            return None

        self._iter_changes_cache = dict((c[0], c) for c in

                                        self._transform.iter_changes())

        changes = self._iter_changes_cache.get(file_id)

                     self.get_file_lines(file_id))

    def is_locked(self):

        return False

 

        return self

    def _has_id(self, file_id, fallback_check):

            return fallback_check(file_id)

 

    def has_id(self, file_id):

        return self._has_id(file_id, self._transform._tree.has_id)

 

    def has_or_had_id(self, file_id):

        return self._has_id(file_id, self._transform._tree.has_or_had_id)

    def _make_inv_entries(self, ordered_entries, specific_file_ids=None,

        yield_parents=False):

            kind = self._transform.final_kind(trans_id)

            if kind is None:

    def iter_entries_by_dir(self, specific_file_ids=None, yield_parents=False):

            specific_file_ids, yield_parents=yield_parents):

            return self._transform._tree.get_file_mtime(file_id)

                    executable = False

        executable = tt._new_executability.get(trans_id, executable)

        changes = self._iter_changes_cache.get(file_id)

        # TODO: This is doing something similar to what WT.annotate_iter is

        #       doing, however it fails slightly because it doesn't know what

        #       the *other* revision_id is, so it doesn't know how to give the

        #       other as the origin for some lines, they all get

        #       'default_revision'

        #       It would be nice to be able to use the new Annotator based

        #       approach, as well.

                kind  = self._transform.final_kind(child_id)

                if kind is not None:

                else:

        if (trans_id == self.transform.root or trans_id == ROOT_PARENT):

            # Check if tree inventory has content. If so, we populate

            # existing_files with the directory content. If there are no

            # entries we skip populating existing_files as its not used.

            # This improves performance and unncessary work on large

            # directory trees. (#501307)

            if total > 0:

                existing_files = set()

                for dir, files in wt.walkdirs():

                    existing_files.update(f[0] for f in files)

        unchanged = [(f, p[1]) for (f, p, c, v, d, n, k, e)

                     in iter if not (c or e[0] != e[1])]

        if accelerator_tree.supports_content_filtering():

            unchanged = [(f, p) for (f, p) in unchanged

                         if not accelerator_tree.iter_search_rules([p]).next()]

        unchanged = dict(unchanged)

        f = file(target_path, 'rb')

        try:

            if tree.get_file_text(file_id) == f.read():

                return True

        finally:

            f.close()

def create_from_tree(tt, trans_id, tree, file_id, bytes=None,

    filter_tree_path=None):

    """Create new file contents according to tree contents.

    

    :param filter_tree_path: the tree path to use to lookup

      content filters to apply to the bytes output in the working tree.

      This only applies if the working tree supports content filtering.

        wt = tt._tree

        if wt.supports_content_filtering() and filter_tree_path is not None:

            filters = wt._content_filter_stack(filter_tree_path)

            bytes = filtered_output_bytes(bytes, filters,