/brz/remove-bazaar

#

#

#

from binascii import hexlify

import collections

import re

from time import time

import warnings

import bzrlib

from bzrlib import bzrdir, errors, ignores, osutils, urlutils

import bzrlib.branch

                            rand_chars,

from bzrlib.symbol_versioning import (deprecated_passed,

        deprecated_method,

        deprecated_function,

        DEPRECATED_PARAMETER,

        zero_eight,

        )

from bzrlib.trace import mutter, note

from bzrlib.textui import show_status

import bzrlib.tree

# the regex removes any weird characters; we don't escape them 

# but rather just pull them out

_gen_file_id_re = re.compile(r'[^\w.]')

_gen_id_suffix = None

_gen_id_serial = 0

 

 

def _next_id_suffix():

    """Create a new file id suffix that is reasonably unique.

    

    On the first call we combine the current time with 64 bits of randomness

    to give a highly probably globally unique number. Then each call in the same

    process adds 1 to a serial number we append to that unique value.

    """

    # XXX TODO: change bzrlib.add.smart_add to call workingtree.add() rather 

    # than having to move the id randomness out of the inner loop like this.

    # XXX TODO: for the global randomness this uses we should add the thread-id

    # before the serial #.

    global _gen_id_suffix, _gen_id_serial

    if _gen_id_suffix is None:

        _gen_id_suffix = "-%s-%s-" % (compact_date(time()), rand_chars(16))

    _gen_id_serial += 1

    return _gen_id_suffix + str(_gen_id_serial)

 

 

    """Return new file id for the basename 'name'.

 

    The uniqueness is supplied from _next_id_suffix.

    """

    # The real randomness is in the _next_id_suffix, the

    # rest of the identifier is just to be nice.

    # So we:

    # 1) Remove non-ascii word characters to keep the ids portable

    # 2) squash to lowercase, so the file id doesn't have to

    #    be escaped (case insensitive filesystems would bork for ids

    #    that only differred in case without escaping).

    # 3) truncate the filename to 20 chars. Long filenames also bork on some

    #    filesystems

    # 4) Removing starting '.' characters to prevent the file ids from

    #    being considered hidden.

    ascii_word_only = _gen_file_id_re.sub('', name.lower())

    short_no_dots = ascii_word_only.lstrip('.')[:20]

    return short_no_dots + _next_id_suffix()

    """An entry that implements the minimum interface used by commands.

            warnings.warn("WorkingTree() is deprecated as of bzr version 0.8. "

            self._branch = wt.branch

                warnings.warn("WorkingTree(..., branch=XXX) is deprecated as of bzr 0.8."

            self._branch = branch

            self._branch = self.bzrdir.open_branch()

            # assume all other formats have their own control files.

        cache_filename = self.bzrdir.get_workingtree_transport(None).local_abspath('stat-cache')

        #hc.scan()

    branch = property(

        fget=lambda self: self._branch,

        doc="""The branch this WorkingTree is connected to.

 

            This cannot be set - it is reflective of the actual disk structure

            the working tree has been constructed from.

            """)

 

    def break_lock(self):

        """Break a lock if one is present from another instance.

 

        Uses the ui factory to ask for confirmation if the lock may be from

        an active process.

 

        This will probe the repository for its lock as well.

        """

        self._control_files.break_lock()

        self.branch.break_lock()

 

        assert inv.root is not None

        :param filename: A filename within the tree. This is a relative path

        from the root of this tree.

 

        return self.bzrdir.is_control_filename(filename)

 

        :return: The WorkingTree that contains 'path', and the rest of path

            path = osutils.getcwd()

 

            if osutils.lexists(self.abspath(path)):

                inv.root.revision = revision_id

        warnings.warn('delete WorkingTree.create', stacklevel=3)

    def relpath(self, path):

        """Return the local path portion from a given path.

        

        The path may be absolute or relative. If its a relative path it is 

        interpreted relative to the python current working directory.

        """

        return relpath(self.basedir, path)

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

    def get_file_text(self, file_id):

        return self.get_file(file_id).read()

 

    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 = self.last_revision()

        if last_rev is None:

            parents = []

        else:

            parents = [last_rev]

        other_parents = self.pending_merges()

        return parents + other_parents

 

        committed_id = Commit().commit( working_tree=self, revprops=revprops,

            *args, **kwargs)

        return committed_id

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

    def get_file_sha1(self, file_id, path=None):

        if not path:

            path = self._inventory.id2path(file_id)

    def get_file_mtime(self, file_id, path=None):

        if not path:

            path = self._inventory.id2path(file_id)

        return os.lstat(self.abspath(path)).st_mtime

 

    if not supports_executable():

        def is_executable(self, file_id, path=None):

    else:

        def is_executable(self, file_id, path=None):

            if not path:

                path = self._inventory.id2path(file_id)

            return bool(stat.S_ISREG(mode) and stat.S_IEXEC & mode)

                raise errors.ForbiddenControlFileError(filename=f)

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

                inv.add_path(f, kind=kind)

            else:

                inv.add_path(f, kind=kind, file_id=file_id)

    def add_parent_tree_id(self, revision_id):

        """Add revision_id as a parent.

 

        This is equivalent to retrieving the current list of parent ids

        and setting the list to its value plus revision_id.

 

        :param revision_id: The revision id to add to the parent list. It may

        be a ghost revision.

        """

        self.set_parent_ids(self.get_parent_ids() + [revision_id])

 

    @needs_write_lock

    def add_parent_tree(self, parent_tuple):

        """Add revision_id, tree tuple as a parent.

 

        This is equivalent to retrieving the current list of parent trees

        and setting the list to its value plus parent_tuple. See also

        add_parent_tree_id - if you only have a parent id available it will be

        simpler to use that api. If you have the parent already available, using

        this api is preferred.

 

        :param parent_tuple: The (revision id, tree) to add to the parent list.             If the revision_id is a ghost, pass None for the tree.

        """

        self.set_parent_ids(self.get_parent_ids() + [parent_tuple[0]])

 

    @needs_write_lock

        except NoSuchFile:

    def set_parent_ids(self, revision_ids):

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

        """

        trees = []

        for rev_id in revision_ids:

            try:

                trees.append(

                    (rev_id, self.branch.repository.revision_tree(rev_id)))

            except errors.RevisionNotPresent:

                trees.append((rev_id, None))

                pass

        self.set_parent_trees(trees)

 

    @needs_write_lock

    def set_parent_trees(self, parents_list):

        """Set the parents of the working tree.

 

        :param parents_list: A list of (revision_id, tree) tuples. 

            If tree is None, then that element is treated as an unreachable

            parent tree - i.e. a ghost.

        """

        parent = parents_list[:1]

        if len(parent):

            self.set_last_revision(parent[0][0])

        else:

            self.set_last_revision(None)

        merges = parents_list[1:]

        self.set_pending_merges([revid for revid, tree in merges])

 

    @needs_write_lock

        """Recursively list all files as (path, class, kind, id, entry).

        # Convert these into local objects to save lookup times

        pathjoin = osutils.pathjoin

        file_kind = osutils.file_kind

 

        # transport.base ends in a slash, we want the piece

        # between the last two slashes

        transport_base_dir = self.bzrdir.transport.base.rsplit('/', 2)[1]

 

        fk_entries = {'directory':TreeDirectory, 'file':TreeFile, 'symlink':TreeLink}

 

        # directory file_id, relative path, absolute path, reverse sorted children

        children = os.listdir(self.basedir)

        children.sort()

        # jam 20060527 The kernel sized tree seems equivalent whether we 

        # use a deque and popleft to keep them sorted, or if we use a plain

        # list and just reverse() them.

        children = collections.deque(children)

        stack = [(inv.root.file_id, u'', self.basedir, children)]

        while stack:

            from_dir_id, from_dir_relpath, from_dir_abspath, children = stack[-1]

 

            while children:

                f = children.popleft()

                if transport_base_dir == f:

                # we know that from_dir_relpath and from_dir_abspath never end in a slash

                # and 'f' doesn't begin with one, we can do a string op, rather

                # than the checks of pathjoin(), all relative paths will have an extra slash

                # at the beginning

                fp = from_dir_relpath + '/' + f

                fap = from_dir_abspath + '/' + f

                elif self.is_ignored(fp[1:]):

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

                    f_norm, can_access = osutils.normalized_filename(f)

                    if f == f_norm or not can_access:

                        # No change, so treat this file normally

                        c = '?'

                    else:

                        # this file can be accessed by a normalized path

                        # check again if it is versioned

                        # these lines are repeated here for performance

                        f = f_norm

                        fp = from_dir_relpath + '/' + f

                        fap = from_dir_abspath + '/' + f

                        f_ie = inv.get_child(from_dir_id, f)

                        if f_ie:

                            c = 'V'

                        elif self.is_ignored(fp[1:]):

                            c = 'I'

                        else:

                            c = '?'

                    yield fp[1:], c, fk, f_ie.file_id, f_ie

                    try:

                        yield fp[1:], c, fk, None, fk_entries[fk]()

                    except KeyError:

                        yield fp[1:], c, fk, None, TreeEntry()

                    continue

                # But do this child first

                new_children = os.listdir(fap)

                new_children.sort()

                new_children = collections.deque(new_children)

                stack.append((f_ie.file_id, fp, fap, new_children))

                # Break out of inner loop, so that we start outer loop with child

                break

            else:

                # if we finished all children, pop it off the stack

                stack.pop()

        if to_dir_ie.kind != 'directory':

            dest_path = pathjoin(to_name, name_tail)

                dest_path = pathjoin(to_name, name_tail)

        for info in self.list_files():

            path = info[0]

            # mutter("search for unknowns in %r", path)

                if subf == '.bzr':

                    continue

                if subf not in dir_entry.children:

                    subf_norm, can_access = osutils.normalized_filename(subf)

                    if subf_norm != subf and can_access:

                        if subf_norm not in dir_entry.children:

                            fl.append(subf_norm)

                    else:

                        fl.append(subf)

                subp = pathjoin(path, subf)

    def _translate_ignore_rule(self, rule):

        """Translate a single ignore rule to a regex.

 

        There are two types of ignore rules.  Those that do not contain a / are

        matched against the tail of the filename (that is, they do not care

        what directory the file is in.)  Rules which do contain a slash must

        match the entire path.  As a special case, './' at the start of the

        string counts as a slash in the string but is removed before matching

        (e.g. ./foo.c, ./src/foo.c)

 

        :return: The translated regex.

        """

        if rule[:2] in ('./', '.\\'):

            # rootdir rule

            result = fnmatch.translate(rule[2:])

        elif '/' in rule or '\\' in rule:

            # path prefix 

            result = fnmatch.translate(rule)

        else:

            # default rule style.

            result = "(?:.*/)?(?!.*/)" + fnmatch.translate(rule)

        assert result[-1] == '$', "fnmatch.translate did not add the expected $"

        return "(" + result + ")"

 

    def _combine_ignore_rules(self, rules):

        """Combine a list of ignore rules into a single regex object.

 

        Each individual rule is combined with | to form a big regex, which then

        has $ added to it to form something like ()|()|()$. The group index for

        each subregex's outermost group is placed in a dictionary mapping back 

        to the rule. This allows quick identification of the matching rule that

        triggered a match.

        :return: a list of the compiled regex and the matching-group index 

        dictionaries. We return a list because python complains if you try to 

        combine more than 100 regexes.

        """

        result = []

        groups = {}

        next_group = 0

        translated_rules = []

        for rule in rules:

            translated_rule = self._translate_ignore_rule(rule)

            compiled_rule = re.compile(translated_rule)

            groups[next_group] = rule

            next_group += compiled_rule.groups

            translated_rules.append(translated_rule)

            if next_group == 99:

                result.append((re.compile("|".join(translated_rules)), groups))

                groups = {}

                next_group = 0

                translated_rules = []

        if len(translated_rules):

            result.append((re.compile("|".join(translated_rules)), groups))

        return result

        ignoreset = getattr(self, '_ignoreset', None)

        if ignoreset is not None:

            return ignoreset

 

        ignore_globs = set(bzrlib.DEFAULT_IGNORE)

        ignore_globs.update(ignores.get_runtime_ignores())

 

        ignore_globs.update(ignores.get_user_ignores())

 

            try:

                ignore_globs.update(ignores.parse_ignore_file(f))

            finally:

                f.close()

 

        self._ignoreset = ignore_globs

        self._ignore_regex = self._combine_ignore_rules(ignore_globs)

        return ignore_globs

 

    def _get_ignore_rules_as_regex(self):

        """Return a regex of the ignore rules and a mapping dict.

 

        :return: (ignore rules compiled regex, dictionary mapping rule group 

        indices to original rule.)

        """

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

            self.get_ignore_list()

        return self._ignore_regex

    

        rules = self._get_ignore_rules_as_regex()

        for regex, mapping in rules:

            match = regex.match(filename)

            if match is not None:

                # one or more of the groups in mapping will have a non-None group 

                # match.

                groups = match.groups()

                rules = [mapping[group] for group in 

                    mapping if groups[group] is not None]

                return rules[0]

        return None

    def is_locked(self):

        return self._control_files.is_locked()

 

    def get_physical_lock_status(self):

        return self._control_files.get_physical_lock_status()

 

            self.branch.generate_revision_history(new_revision)

        except errors.NoSuchRevision:

            # not present in the repo - dont try to set it deeper than the tip

            self.branch.set_revision_history([new_revision])

        # TODO: this should allow the ready-to-use inventory to be passed in,

        # as commit already has that ready-to-use [while the format is the

        # same, that is].

            # the fast path is to copy the raw xml from the repository. If the

            # xml contains 'revision_id="', then we assume the right 

            # revision_id is set. We must check for this full string, because a

            # root node id can legitimately look like 'revision_id' but cannot

            # contain a '"'.

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

            if not 'revision_id="' in xml.split('\n', 1)[0]:

                inv = self.branch.repository.deserialise_inventory(

                    new_revision, xml)

                inv.revision_id = new_revision

                xml = bzrlib.xml5.serializer_v5.write_inventory_to_string(inv)

            assert isinstance(xml, str), 'serialised xml must be bytestring.'

            sio = StringIO(xml)

            self._control_files.put(path, sio)

        except (errors.NoSuchRevision, errors.RevisionNotPresent):

        return self._control_files.get(path).read()

    def remove(self, files, verbose=False, to_file=None):

                show_status(new_status, inv[fid].kind, f, to_file=to_file)

        raise NotImplementedError(self.unlock)

                # our last revision was not the prior branch last revision

    def add_conflicts(self, arg):

        raise UnsupportedOperation(self.add_conflicts, self)

 

            except errors.NoSuchFile:

                text = False

                        if kind != "file":

                    except errors.NoSuchFile:

                    if text == False:

class WorkingTree2(WorkingTree):

    """This is the Format 2 working tree.

 

    This was the first weave based working tree. 

     - uses os locks for locking.

     - uses the branch last-revision.

    """

 

    def unlock(self):

        # we share control files:

        if self._hashcache.needs_write and self._control_files._lock_count==3:

            self._hashcache.write()

        # reverse order of locking.

        try:

            return self._control_files.unlock()

        finally:

            self.branch.unlock()

 

 

    @needs_write_lock

    def add_conflicts(self, new_conflicts):

        conflict_set = set(self.conflicts())

        conflict_set.update(set(list(new_conflicts)))

        self.set_conflicts(ConflictList(sorted(conflict_set,

                                               key=Conflict.sort_key)))

 

    def unlock(self):

        if self._hashcache.needs_write and self._control_files._lock_count==1:

            self._hashcache.write()

        # reverse order of locking.

        try:

            return self._control_files.unlock()

        finally:

            self.branch.unlock()

 

            raise errors.UnknownFormatError(format=format_string)

    def stub_initialize_remote(self, control_files):

        """As a special workaround create critical control files for a remote working tree

        

        This ensures that it can later be updated and dealt with locally,

        since BzrDirFormat6 and BzrDirFormat5 cannot represent dirs with 

        no working tree.  (See bug #43064).

        """

        sio = StringIO()

        inv = Inventory()

        bzrlib.xml5.serializer_v5.write_inventory(inv, sio)

        sio.seek(0)

        control_files.put('inventory', sio)

 

        control_files.put_utf8('pending-merges', '')

        

 

        wt = WorkingTree2(a_bzrdir.root_transport.local_abspath('.'),

        return WorkingTree2(a_bzrdir.root_transport.local_abspath('.'),

        - uses a LockDir to guard access for writes.

        revision_id allows creating a working tree at a different

        control_files.lock_write()

        wt = WorkingTree3(a_bzrdir.root_transport.local_abspath('.'),

        wt.lock_write()

        try:

            wt._write_inventory(inv)

            wt.set_root_id(inv.root.file_id)

            wt.set_last_revision(revision_id)

            wt.set_pending_merges([])

            build_tree(wt.basis_tree(), wt)

        finally:

            wt.unlock()

            control_files.unlock()

        return self._open(a_bzrdir, self._open_control_files(a_bzrdir))

 

    def _open(self, a_bzrdir, control_files):

        """Open the tree itself.

        

        :param a_bzrdir: the dir for the tree.

        :param control_files: the control files for the tree.

        """

        return WorkingTree3(a_bzrdir.root_transport.local_abspath('.'),

    def _clone_test(self, test, bzrdir_format, workingtree_format, variation):

        """Clone test for adaption."""

        new_test = deepcopy(test)

        new_test.transport_server = self._transport_server

        new_test.transport_readonly_server = self._transport_readonly_server

        new_test.bzrdir_format = bzrdir_format

        new_test.workingtree_format = workingtree_format

        def make_new_test_id():

            new_id = "%s(%s)" % (test.id(), variation)

            return lambda: new_id

        new_test.id = make_new_test_id()

        return new_test

    

            new_test = self._clone_test(

                test,

                bzrdir_format,

                workingtree_format, workingtree_format.__class__.__name__)