# -*- coding: iso-8859-1 -*- # # Copyright (C) 2005 Edgewall Software # Copyright (C) 2005-2006 Christian Boos # Copyright (C) 2005 Johan Rydberg # Copyright (C) 2006 Yann Hodique # Copyright (C) 2006 Lukas Lalinsky # Copyright (C) 2006 Marien Zwart # All rights reserved. # # This software may be used and distributed according to the terms # of the GNU General Public License, incorporated herein by reference. # # Author: Yann Hodique """Bazaar-ng backend for trac's versioncontrol.""" import datetime import StringIO from itertools import izip import time import urllib from trac import versioncontrol, core, mimeview, wiki from trac.util.html import html, Markup from bzrlib import ( branch as bzrlib_branch, bzrdir, errors, inventory, osutils, revision, transport, tsort, ) class BzrConnector(core.Component): """The necessary glue between our repository and trac.""" core.implements(versioncontrol.IRepositoryConnector, wiki.IWikiMacroProvider) # IRepositoryConnector def get_supported_types(self): """Support for `repository_type = bzr`""" yield ('bzr', 8) yield ('bzr+debug', 8) def get_repository(self, repos_type, repos_dir, authname): """Return a `BzrRepository`""" assert repos_type in ('bzr', 'bzr+debug') if repos_type == 'bzr+debug': # HACK: this causes logging to be applied to all BzrRepositories, # BzrChangesets, and BzrNodes BzrRepository.__getattribute__ = getattribute BzrChangeset.__getattribute__ = getattribute BzrNode.__getattribute__ = getattribute return BzrRepository(repos_dir, self.log) # IWikiMacroProvider def get_macros(self): yield 'Branches' def get_macro_description(self, name): assert name == 'Branches' return 'Render a list of available branches.' def render_macro(self, req, name, content): assert name == 'Branches' # This is pretty braindead but adding an option for this is too. manager = versioncontrol.RepositoryManager(self.env) if manager.repository_type != 'bzr': raise core.TracError('Configured repo is not a bzr repo') temp_branch = bzrlib_branch.Branch.open(manager.repository_dir) trans = temp_branch.repository.bzrdir.root_transport branches = sorted(self._get_branches(trans)) # Slight hack. We know all these branches are in the same # repo, so we can read lock that once. repo = bzrdir.BzrDir.open_from_transport(trans).open_repository() repo.lock_read() try: return html.TABLE(class_='listing')( html.THEAD(html.TR( html.TH('Path'), html.TH('Nick'), html.TH('Last Change'))), html.TBODY([ html.TR( html.TD(html.A(loc, href=req.href.browser( rev=':%s' % (urllib.quote(loc, ''), )))), html.TD(target.nick), html.TD( datetime.datetime.fromtimestamp( repo.get_revision( target.last_revision()).timestamp ).ctime()), ) for loc, target in branches])) finally: repo.unlock() class BzrRepository(versioncontrol.Repository): """Present a bzr branch as a trac repository.""" def __init__(self, location, log): versioncontrol.Repository.__init__(self, location, None, log) self.root_transport = transport.get_transport(location) self._tree_cache = {} self._locked_branches = [] self._branch_cache = {} self._history = None self._previous = None self._revision_cache = {} def __repr__(self): return 'BzrRepository(%r)' % self.root_transport.base def branch_path(self, branch): """Determine the relative path to a branch from the root""" repo_path = self.root_transport.base branch_path = branch.bzrdir.root_transport.base if branch_path.startswith(repo_path): return branch_path[len(repo_path):].rstrip('/') else: repo_path = osutil.normalizepath(repo_path) branch_path = osutil.normalizepath(branch_path) return osutils.relpath(repo_path, branch_path) def string_rev(self, branch, revid): """Create a trac rev string. branch is None or a bzr branch. """ if branch is None: # No "safe" chars (make sure "/" is escaped) return self._escape(revid) relpath = self.branch_path(branch) try: return '%s,%s' % (urllib.quote(relpath, ':'), branch.revision_id_to_revno(revid)) except errors.NoSuchRevision: dotted = self.dotted_revno(branch, revid) if dotted is not None: return '%s,%s' % (urllib.quote(relpath, ':'), dotted) if revid in branch.repository.get_ancestry(branch.last_revision()): return '%s,%s' % (urllib.quote(relpath, ''), self._escape(revid, ':')) else: return self._escape(revid, ':') @staticmethod def _escape(string): return urllib.quote(string, '') @staticmethod def _string_rev_revid(relpath, revid): return '%s,%s' % (urllib.quote(relpath, ''), urllib.quote(revid, '')) def _parse_rev(self, rev): """Translate a trac rev string into a (branch, revid) tuple. branch is None or a bzr branch object. Supported syntax: - "spork,123" is revno 123 in the spork branch. - "spork,revid" is a revid in the spork branch. (currently revid is assumed to be in the branch ancestry!) Branch paths and revids are urlencoded. """ # Try integer revno to revid conversion. if rev.isdigit(): raise versioncontrol.NoSuchChangeset(rev) # Try path-to-branch-in-repo. if ',' in rev: split = rev.split(',') if len(split) != 2: raise versioncontrol.NoSuchChangeset(rev) rev_branch,rev_rev = split try: branch = self.get_branch(rev_branch) except errors.NotBranchError: raise versioncontrol.NoSuchChangeset(rev) if len(split) == 2: if rev_rev.isdigit(): try: revid = branch.get_rev_id(int(rev_rev)) except errors.NoSuchRevision: raise versioncontrol.NoSuchChangeset(rev) else: dotted = rev_rev.split('.') for segment in dotted: if not segment.isdigit(): revid = urllib.unquote(rev_rev) break else: cache = self.get_branch_cache(branch) revid = cache.revid_from_dotted(rev_rev) if revid is None: raise repr(dotted) revid = urllib.unquote(rev_rev) else: revid = branch.last_revision() return branch, revid # Try raw revid. revid = urllib.unquote(rev) if revid in ('current:', 'null:'): return None, revid return None, revid if self.repo.has_revision(revid): return None, revid # Unsupported format. raise versioncontrol.NoSuchChangeset(rev) def __del__(self): # XXX Eeeeeww. Unfortunately for us trac does not actually call the # close method. So we do this. Quite silly, since bzr does the same # thing (printing a warning...) self.close() # Trac api methods. def close(self): """Release our branches. Trac does not *have* to call this!""" for branch in self._locked_branches: branch.unlock() def get_branch(self, location): if location in self._branch_cache: return self._branch_cache[location].branch trans = self.root_transport for piece in urllib.unquote(location).split('/'): if piece: trans = trans.clone(piece) target_bzrdir = bzrdir.BzrDir.open_from_transport(trans) branch = target_bzrdir.open_branch() branch.lock_read() self._locked_branches.append(branch) self._branch_cache[location] = BranchCache(self, branch) return branch def get_containing_branch(self, location): branch, relpath = containing_branch(self.root_transport, location) real_location = location[:-len(relpath)].rstrip('/') if real_location not in self._branch_cache: self._branch_cache[real_location] = BranchCache(self, branch) branch.lock_read() self._locked_branches.append(branch) # return the cached version, possibly throwing away the one we just # retrieved. return self._branch_cache[real_location].branch, relpath def _get_branches(self, trans=None, loc=()): """Find branches under a listable transport. Does not descend into control directories or branch directories. (branches inside other branches will not be listed) """ if trans is None: trans = self.root_transport try: children = trans.list_dir('.') if '.bzr' in children: children.remove('.bzr') try: target_bzrdir = bzrdir.BzrDir.open_from_transport(trans) yield '/'.join(loc), target_bzrdir.open_branch() except errors.NotBranchError: pass else: return for child in children: for child_loc, child_branch in self._get_branches( trans.clone(child), loc + (child,)): yield child_loc, child_branch except errors.NoSuchFile, e: return def get_changeset(self, rev): """Retrieve a Changeset.""" branch, revid = self._parse_rev(rev) try: return BzrChangeset(self, branch, revid, self.log) except errors.NoSuchRevision, e: assert e.revision == revid raise versioncontrol.NoSuchChangeset(rev) # TODO: get_changesets? def has_node(self, path, rev=None): """Return a boolean indicating if the node is present in a rev.""" try: self.get_node(path, rev) except versioncontrol.NoSuchNode: return False else: return True def get_node(self, path, rev=None): """Return a Node object or raise NoSuchNode or NoSuchChangeset.""" path = self.normalize_path(path) if rev is None: rev = 'current%3A' revbranch, revid = self._parse_rev(rev) try: branch, relpath = self.get_containing_branch(path) except errors.NotBranchError: if not self.root_transport.has(path): raise versioncontrol.NoSuchNode(path, rev) return UnversionedDirNode(self, path) if revbranch is None: revbranch = branch try: if revid == 'current:': tree = revbranch.basis_tree() else: tree = revbranch.repository.revision_tree(revid) except (errors.NoSuchRevision, errors.RevisionNotPresent): raise versioncontrol.NoSuchChangeset(rev) file_id = tree.inventory.path2id(relpath) if file_id is None: raise versioncontrol.NoSuchNode(path, rev) entry = tree.inventory[file_id] klass = NODE_MAP[entry.kind] return klass(self, branch, tree, entry, path) def get_oldest_rev(self): # TODO just use revno here # (by definition, this revision is always in the branch history) return self.string_rev(None, 'null:') def get_youngest_rev(self): # TODO just use revno here # (by definition, this revision is always in the branch history) return self.string_rev(None, 'current:') def _repo_history(self): revisions = {} repos = {} branches = {} seen = set() for loc, branch in self._get_branches(): repo_base = branch.repository.bzrdir.transport.base repos[repo_base] = branch.repository for revision_id in reversed(branch.revision_history()): if revision_id in seen: break revisions.setdefault(repo_base, []).append(revision_id) branches[revision_id] = branch seen.add(revision_id) revision_set = set() for repo_base, revision_ids in revisions.iteritems(): revision_set.update(repos[repo_base].get_revisions(revision_ids)) revisions = sorted(revision_set, key=lambda x: x.timestamp) return [(r.revision_id, branches[r.revision_id]) for r in revisions] def previous_rev(self, rev): branch, revid = self._parse_rev(rev) if revid == 'null:': return None if self._history is None: self._history = self._repo_history() self._previous = {} last = None for rev, branch in reversed(self._history): if rev == last: raise repr(self._history) if last is not None: self._previous[last] = (branch, rev) last = rev if revid == 'current:': return self.string_rev(self._history[-1][1], self._history[-1][0]) try: return self.string_rev(*self._previous[revid]) except KeyError: return 'null:' def next_rev(self, rev, path=''): # TODO path is ignored. branch, revid = self._parse_rev(rev) if revid == 'current:': return None if revid == 'null:': return 'current:' if branch is None: ancestry = self.repo.get_ancestry(self.branch.last_revision()) else: ancestry = branch.repository.get_ancestry(branch.last_revision()) try: idx = ancestry.index(revid) except ValueError: # XXX this revision is not in the branch ancestry. Now what? return None try: next_revid = ancestry[idx + 1] except IndexError: # There is no next rev. Now what? return None return self.string_rev(branch, next_revid) def rev_older_than(self, rev1, rev2): if rev1 == rev2: return False branch1, rrev1 = self._parse_rev(rev1) branch2, rrev2 = self._parse_rev(rev2) if rrev2 == 'current:': return False first_before_second = rrev1 in branch2.repo.get_ancestry(rrev2) second_before_first = rrev2 in branch1.repo.get_ancestry(rrev1) if first_before_second and second_before_first: raise core.TracError('%s and %s precede each other?' % (rrev1, rrev2)) if first_before_second: return True if second_before_first: return False # Bah, unrelated revisions. Fall back to comparing timestamps. return (self.repo.get_revision(rrev1).timestamp < self.repo.get_revision(rrev2).timestamp) # XXX what is get_youngest_rev_in_cache doing in here def get_path_history(self, path, rev=None, limit=None): """Shortcut for Node's get_history.""" # XXX I think the intention for this one is probably different: # it should track the state of this filesystem location across time. # That is, it should keep tracking the same path as stuff is moved # on to / away from that path. # No need to normalize/unquote, get_node is a trac api method # so it takes quoted values. return self.get_node(path, rev).get_history(limit) def normalize_path(self, path): """Remove leading and trailing '/'""" # Also turns None into '', just in case. return path and path.strip('/') or '' def normalize_rev(self, rev): """Turn a user-specified rev into a "normalized" rev. This turns None into a rev, and may convert a revid-based rev into a revno-based one. """ if rev is None: branch = None revid = 'current:' else: branch, revid = self._parse_rev(rev) if branch is not None: repository = branch.repository else: repository = None return self.string_rev(branch, revid) def short_rev(self, rev): """Attempt to shorten a rev. This returns the revno if there is one, otherwise returns a "nearby" revno with a ~ prefix. The result of this method is used above the line number columns in the diff/changeset viewer. There is *very* little room there. Our ~revno results are actually a little too wide already. Tricks like using the branch nick simply do not fit. """ branch, revid = self._parse_rev(rev) if branch is None: return '????' history = branch.revision_history() # First try if it is a revno. try: return str(history.index(revid) + 1) except ValueError: # Get the closest thing that *is* a revno. ancestry = branch.repository.get_ancestry(revid) # We've already tried the current one. ancestry.pop() for ancestor in reversed(ancestry): try: return '~%s' % (history.index(ancestor) + 1,) except ValueError: pass # XXX unrelated branch. Now what? return '????' def get_changes(self, old_path, old_rev, new_path, new_rev, ignore_ancestry=1): """yields (old_node, new_node, kind, change) tuples.""" # ignore_ancestry is ignored, don't know what it's for. if old_path != new_path: raise core.TracError( 'Currently the bzr plugin does not support this between ' 'different directories. Sorry.') old_branch, old_revid = self._parse_rev(old_rev) new_branch, new_revid = self._parse_rev(new_rev) old_tree = old_branch.repository.revision_tree(old_revid) new_tree = new_branch.repository.revision_tree(new_revid) delta = new_tree.changes_from(old_tree) for path, file_id, kind in delta.added: entry = new_tree.inventory[file_id] node = NODE_MAP[kind](self, new_branch, new_tree, entry, path) cur_path = new_tree.id2path(file_id) node._history_cache[(new_revid, old_revid, file_id)] = \ cur_path, new_rev, versioncontrol.Changeset.ADD yield None, node, node.kind, versioncontrol.Changeset.ADD for path, file_id, kind in delta.removed: entry = old_tree.inventory[file_id] node = NODE_MAP[kind](self, old_branch, old_tree, entry, path) yield node, None, node.kind, versioncontrol.Changeset.DELETE for oldpath, newpath, file_id, kind, textmod, metamod in delta.renamed: oldnode = NODE_MAP[kind](self, old_branch, old_tree, old_tree.inventory[file_id], oldpath) newnode = NODE_MAP[kind](self, new_branch, new_tree, new_tree.inventory[file_id], newpath) if oldnode.kind != newnode.kind: raise core.TracError( '%s changed kinds, I do not know how to handle that' % ( newpath,)) yield oldnode, newnode, oldnode.kind, versioncontrol.Changeset.MOVE for path, file_id, kind, textmod, metamod in delta.modified: # Bzr won't report a changed path as a rename but trac wants that. oldpath = old_tree.id2path(file_id) oldnode = NODE_MAP[kind](self, old_branch, old_tree, old_tree.inventory[file_id], oldpath) newnode = NODE_MAP[kind](self, new_branch, new_tree, new_tree.inventory[file_id], path) if oldnode.kind != newnode.kind: raise core.TracError( '%s changed kinds, I do not know how to handle that' % ( newpath,)) if oldpath != path: action = versioncontrol.Changeset.MOVE else: action = versioncontrol.Changeset.EDIT cur_path = new_tree.id2path(file_id) newnode._history_cache[(new_revid, old_revid, file_id)] = \ cur_path, new_rev, action yield oldnode, newnode, oldnode.kind, action def dotted_revno(self, branch, revid): return self.get_branch_cache(branch).dotted_revno(revid) def get_branch_cache(self, branch): branch_key = branch.bzrdir.root_transport.base if branch_key not in self._branch_cache: self._branch_cache[branch_key] = BranchCache(self, branch) return self._branch_cache[branch_key] def sorted_revision_history(self, branch): return self.get_branch_cache(branch).sorted_revision_history() def sync(self): """Dummy to satisfy interface requirements""" # XXX should we be dumping in-mem caches? Seems unlikely. self.log = None pass class BzrNode(versioncontrol.Node): pass class UnversionedDirNode(BzrNode): def __init__(self, bzr_repo, path): rev_string = urllib.quote('current:') BzrNode.__init__(self, path, rev_string, versioncontrol.Node.DIRECTORY) self.transport = bzr_repo.root_transport.clone(path) self.bzr_repo = bzr_repo self.path = path def __repr__(self): return 'UnversionedDirNode(path=%r)' % self.path def get_properties(self): return {} def get_entries(self): result = [] for name in self.transport.list_dir(''): if name == '.bzr': continue stat_mode = self.transport.stat(name).st_mode kind = osutils.file_kind_from_stat_mode(stat_mode) if not kind == 'directory': continue child_path = osutils.pathjoin(self.path, name) try: branch = self.bzr_repo.get_branch(child_path) except errors.NotBranchError: result.append(UnversionedDirNode(self.bzr_repo, child_path)) else: tree = branch.basis_tree() node = BzrDirNode(self.bzr_repo, branch, tree, tree.inventory.root, child_path) result.append(node) return result def get_content_length(self): return 0 def get_content(self): return StringIO.StringIO('') def get_content_type(self): return 'application/octet-stream' def get_history(self, limit=None): return [(self.path, 'current%3A', 'add')] def sorted_revision_history(branch, generate_revno=False): history = branch.revision_history() graph = branch.repository.get_revision_graph(history[-1]) return tsort.merge_sort(graph, history[-1], generate_revno=generate_revno) class BzrVersionedNode(BzrNode): _history_cache = {} _diff_map = { 'modified': versioncontrol.Changeset.EDIT, 'unchanged': versioncontrol.Changeset.EDIT, 'added': versioncontrol.Changeset.ADD, inventory.InventoryEntry.RENAMED: versioncontrol.Changeset.MOVE, inventory.InventoryEntry.MODIFIED_AND_RENAMED: versioncontrol.Changeset.MOVE } def __init__(self, bzr_repo, branch, revisiontree, entry, path): """Initialize. path has to be a normalized path.""" rev_string = bzr_repo.string_rev(branch, entry.revision) BzrNode.__init__(self, path, rev_string, self.kind) self.bzr_repo = bzr_repo self.log = bzr_repo.log self.repo = branch.repository self.branch = branch self.tree = revisiontree self.entry = entry # XXX I am not sure if this makes any sense but it does make # the links in the changeset viewer work. self.created_rev = self.rev self.created_path = self.path self.root_path = path[:-len(self.tree.id2path(self.entry.file_id))] def get_properties(self): # Must at least return an empty dict here (base class version raises). result = {} if self.entry.executable: result['executable'] = 'True' return result def _merging_history(self): """Iterate through history revisions that merged changes to this node This includes all revisions in which the revision_id changed. It may also include a few revisions in which the revision_id did not change, if the modification was subsequently undone. """ weave = self.tree.get_weave(self.entry.file_id) file_ancestry = weave.get_ancestry(self.entry.revision) # Can't use None here, because it's a legitimate revision id. last_yielded = 'bogus:' for num, revision_id, depth, revno, eom in \ self.bzr_repo.sorted_revision_history(self.branch): if depth == 0: last_mainline = revision_id if last_mainline == last_yielded: continue if revision_id in file_ancestry: yield last_mainline last_yielded = last_mainline yield None def get_history(self, limit=None): """Backward history. yields (path, revid, chg) tuples. path is the path to this entry. revid is the revid string. It is the revision in which the change was applied to the branch, not necessarily the revision that originated the change. In SVN terms, it is a changeset, not a file revision. chg is a Changeset.ACTION thing. First thing should be for the current revision. limit is an int cap on how many entries to return. """ history_iter = self._get_history() if limit is None: return history_iter else: return (y for x, y in izip(range(limit), history_iter)) def _get_history(self, limit=None): file_id = self.entry.file_id revision = None history = list(self._merging_history()) cache = self.bzr_repo.get_branch_cache(self.branch) trees = cache.revision_trees(history) for prev_tree in trees: previous_revision = prev_tree.get_revision_id() try: prev_file_revision = prev_tree.inventory[file_id].revision except errors.NoSuchId: prev_file_revision = None if (revision is not None and prev_file_revision != file_revision): path, rev_str, chg = \ self.get_change(revision, previous_revision, file_id) branch_revision = self.bzr_repo.string_rev(self.branch, revision) yield (osutils.pathjoin(self.root_path, path), branch_revision, chg) if prev_file_revision is None: break revision = previous_revision file_revision = prev_file_revision def get_change(self, revision, previous_revision, file_id): key = (revision, previous_revision, file_id) if key not in self._history_cache or False: self._history_cache[key] = self.calculate_history(revision, previous_revision, file_id) return self._history_cache[key] def calculate_history(self, revision, previous_revision, file_id): cache = self.bzr_repo.get_branch_cache(self.branch) tree = cache.revision_tree(revision) current_entry = tree.inventory[file_id] current_path = tree.id2path(file_id) if previous_revision not in (None, 'null:'): previous_tree = cache.revision_tree(previous_revision) previous_entry = previous_tree.inventory[file_id] else: previous_entry = None # We should only get revisions in the ancestry for which # we exist, so this should succeed.. return self.compare_entries(current_path, current_entry, previous_entry) def compare_entries(self, current_path, current_entry, previous_entry): diff = current_entry.describe_change(previous_entry, current_entry) rev = self.bzr_repo.string_rev(self.branch, current_entry.revision) try: return current_path, rev, self._diff_map[diff] except KeyError: raise Exception('unknown describe_change %r' % (diff,)) def get_last_modified(self): return self.tree.get_file_mtime(self.entry.file_id) class BzrDirNode(BzrVersionedNode): isdir = True isfile = False kind = versioncontrol.Node.DIRECTORY def __init__(self, bzr_repo, branch, revisiontree, entry, path, revcache=None): BzrVersionedNode.__init__(self, bzr_repo, branch, revisiontree, entry, path) if revcache is None: ancestry = self.repo.get_ancestry(revisiontree.get_revision_id()) ancestry.reverse() self.revcache = {} best = self._get_cache(self.revcache, ancestry, entry) self._orig_rev = ancestry[best] self.rev = bzr_repo.string_rev(self.branch, (ancestry[best])) else: self.revcache = revcache self._orig_rev = revcache[entry.file_id] self.rev = bzr_repo.string_rev(self.branch, self._orig_rev) def __repr__(self): return 'BzrDirNode(path=%r, relpath=%r)' % (self.path, self.entry.name) @classmethod def _get_cache(cls, cache, ancestry, entry, ancestry_idx=None): """Populate a file_id <-> revision_id mapping. This mapping is different from InventoryEntry.revision, but only for directories. In this scheme, directories are considered modified if their contents are modified. The revision ids are not guaranteed to be in the mainline revision history. cache: The cache to populate ancestry: A topologically-sorted list of revisions, with more recent revisions having lower indexes. entry: The InventoryEntry to start at ancestry_idx: A mapping of revision_id <-> ancestry index. """ if ancestry_idx is None: ancestry_idx = dict((r, n) for n, r in enumerate(ancestry)) # best ~= most recent revision to modify a child of this directory best = ancestry_idx[entry.revision] for child in entry.children.itervalues(): if child.kind == 'directory': index = cls._get_cache(cache, ancestry, child, ancestry_idx) cache[child.file_id] = ancestry[index] else: index = ancestry_idx[child.revision] best = min(best, index) return best def get_content(self): """Return a file-like (read(length)) for a file, None for a dir.""" return None def get_entries(self): """Yield child Nodes if a dir, return None if a file.""" for name, entry in self.entry.children.iteritems(): childpath = '/'.join((self.path, name)) klass = NODE_MAP[entry.kind] if klass is BzrDirNode: yield klass(self.bzr_repo, self.branch, self.tree, entry, childpath, self.revcache) else: yield klass(self.bzr_repo, self.branch, self.tree, entry, childpath) def get_content_length(self): return None def get_content_type(self): return None def _get_revision_history(self, limit=None): history = self.branch.revision_history() first = history[0] if limit is not None: history = history[-limit:] self.bzr_repo.get_branch_cache(self.branch).cache_revisions(history) for rev_id in reversed(history): if rev_id == first: operation = versioncontrol.Changeset.ADD else: operation = versioncontrol.Changeset.EDIT yield (self.path, self.bzr_repo.string_rev(self.branch, rev_id), operation) def get_history(self, limit=None): """Backward history. yields (path, rev, chg) tuples. path is the path to this entry, rev is the revid string. chg is a Changeset.ACTION thing. First thing should be for the current revision. This is special because it checks for changes recursively, not just to this directory. bzr only treats the dir as changed if it is renamed, not if its contents changed. Trac needs this recursive behaviour. limit is an int cap on how many entries to return. """ current_entry = self.entry file_id = current_entry.file_id if current_entry.parent_id == None: for r in self._get_revision_history(limit): yield r return count = 0 # We need the rev we were created with, not the rev the entry # specifies (our contents may have changed between that rev # and our own current rev). current_revid = self._orig_rev if self.branch is not None: history = self.branch.revision_history() else: history = [] if current_revid == 'current:': current_revid = history[-1] # If the current_revid we start from is in the branch history, # limit our view to just the history, not the full ancestry. try: index = history.index(current_revid) except ValueError: ancestry = self.repo.get_ancestry(current_revid) # The last entry is this rev, skip it. ancestry.pop() ancestry.reverse() # The last entry is None, skip it. ancestry.pop() else: ancestry = ['null:'] + history[:index] ancestry.reverse() # Load a bunch of trees in one go. We do not know how many we # need: we may end up skipping some trees because they do not # change us. chunksize = limit or 100 current_tree = self.tree current_path = current_tree.id2path(file_id) path_prefix = self.path[:-len(current_path)] while ancestry: chunk, ancestry = ancestry[:chunksize], ancestry[chunksize:] cache = self.bzr_repo.get_branch_cache(self.branch) for previous_revid, previous_tree in izip( chunk, cache.revision_trees(chunk)): if file_id in previous_tree.inventory: previous_entry = previous_tree.inventory[file_id] else: previous_entry = None delta = current_tree.changes_from( previous_tree, specific_files=[current_path]) if not delta.has_changed(): current_entry = previous_entry current_path = previous_tree.inventory.id2path(file_id) current_revid = previous_revid current_tree = previous_tree continue diff = current_entry.describe_change(previous_entry, current_entry) if diff == 'added': yield (path_prefix+current_path, self.bzr_repo.string_rev(self.branch, current_revid), versioncontrol.Changeset.ADD) # There is no history before this point, we're done. return elif diff == 'modified' or diff == 'unchanged': # We want the entry anyway. yield (path_prefix+current_path, self.bzr_repo.string_rev(self.branch, current_revid), versioncontrol.Changeset.EDIT) elif diff in (current_entry.RENAMED, current_entry.MODIFIED_AND_RENAMED): yield (path_prefix+current_path, self.bzr_repo.string_rev(self.branch, current_revid), versioncontrol.Changeset.MOVE) else: raise Exception('unknown describe_change %r' % (diff,)) count += 1 if limit is not None and count >= limit: return current_entry = previous_entry current_path = previous_tree.inventory.id2path(file_id) current_revid = previous_revid current_tree = previous_tree def get_previous(self): """Equivalent to i=iter(get_history(2));i.next();return i.next(). The default implementation does essentially that, but we specialcase it because we can skip the loading of all the trees. """ # Special case: if this is the root node it (well, its # contents) change every revision. if not self.tree.id2path(self.entry.file_id): return self.path, self.rev, versioncontrol.Changeset.EDIT if self._orig_rev != self.entry.revision: # The last change did not affect this dir directly, it changed # our contents. return self.path, self.rev, versioncontrol.Changeset.EDIT # We were affected directly. Get a delta to figure out how. delta = self.repo.get_revision_delta(self._orig_rev) for path, file_id, kind in delta.added: if file_id == self.entry.file_id: return path, self.rev, versioncontrol.Changeset.ADD for oldpath, newpath, file_id, kind, textmod, metamod in delta.renamed: if file_id == self.entry.file_id: return newpath, self.rev, versioncontrol.Changeset.MOVE # We were removed (which does not make any sense, # the tree we were constructed from is newer and has us) raise core.TracError('should not get here, %r %r %r' % (self.entry, delta, self._orig_rev)) class BzrFileNode(BzrVersionedNode): isfile = True isdir = False kind = versioncontrol.Node.FILE def __repr__(self): return 'BzrFileNode(path=%r)' % self.path def get_content(self): """Return a file-like (read(length)) for a file, None for a dir.""" return self.tree.get_file(self.entry.file_id) def get_entries(self): return None def get_content_length(self): return self.entry.text_size def get_content_type(self): return mimeview.get_mimetype(self.name) class BzrSymlinkNode(BzrVersionedNode): """Kinda like a file only not really. Empty, properties only.""" isfile = True isdir = False kind = versioncontrol.Node.FILE def __repr__(self): return 'BzrSymlinkNode(path=%r)' % self.path def get_content(self): return StringIO.StringIO('') def get_entries(self): return None def get_content_length(self): return 0 def get_content_type(self): return 'text/plain' def get_properties(self): return {'destination': self.entry.symlink_target} NODE_MAP = { 'directory': BzrDirNode, 'file': BzrFileNode, 'symlink': BzrSymlinkNode, } class BzrChangeset(versioncontrol.Changeset): def __init__(self, bzr_repo, branch, revid, log): """Initialize from a bzr repo, an unquoted revid and a logger.""" self.log = log self.bzr_repo = bzr_repo if branch is None: if revid in ('current:', 'null:'): self.revision = revision.Revision(revid, committer='', message='', timezone='') versioncontrol.Changeset.__init__(self, urllib.quote(revid), '', '', time.time()) else: raise errors.NoSuchRevision(None, revid) else: self.revision = bzr_repo.get_branch_cache(branch).get_revision(revid) versioncontrol.Changeset.__init__(self, bzr_repo.string_rev( branch, revid), self.revision.message, self.revision.committer, self.revision.timestamp) self.branch = branch def __repr__(self): return 'BzrChangeset(%r)' % (self.revision.revision_id) def get_properties(self): """Return an iterator of (name, value, is wikitext, html class).""" for name, value in self.revision.properties.iteritems(): yield name, value, False, '' if len(self.revision.parent_ids) > 1: for name, link in [('parent trees', ' * source:@%s'), ('changesets', ' * [changeset:%s]')]: yield name, '\n'.join( link % (self.bzr_repo.string_rev(self.branch, parent),) for parent in self.revision.parent_ids), True, '' def get_changes(self): """Yield changes. Return tuples are (path, kind, change, base_path, base_rev). change is self.ADD/COPY/DELETE/EDIT/MOVE. kind is Node.FILE or Node.DIRECTORY. base_path and base_rev are the location and revision of the file before "ours". """ if self.revision.revision_id in ('current:', 'null:'): return branchpath = osutils.relpath( osutils.normalizepath(self.bzr_repo.root_transport.base), osutils.normalizepath(self.branch.bzrdir.root_transport.base)) for path, kind, change, base_path, base_rev in self._get_changes(): if path is not None: path = osutils.pathjoin(branchpath, path) if base_path is not None: base_path = osutils.pathjoin(branchpath, base_path) yield (path, kind, change, base_path, base_rev) def _get_changes(self): if self.revision.revision_id in ('current:', 'null:'): return this = self.branch.repository.revision_tree(self.revision.revision_id) parents = self.revision.parent_ids if parents: parent_revid = parents[0] else: parent_revid = None other = self.branch.repository.revision_tree(parent_revid) delta = this.changes_from(other) kindmap = {'directory': versioncontrol.Node.DIRECTORY, 'file': versioncontrol.Node.FILE, 'symlink': versioncontrol.Node.FILE, # gotta do *something* } # We have to make sure our base_path/base_rev combination # exists (get_node succeeds). If we use # other.inventory[file_id].revision as base_rev we cannot use # what bzr hands us or what's in other.inventory as base_path # since the parent node may have been renamed between what we # return as base_rev and the revision the "other" inventory # corresponds to (parent_revid). # So we can either return other.id2path(file_id) and parent_revid # or use entry.revision and pull up the inventory for that revision # to get the path. Currently the code does the former, # remember to update the paths if it turns out returning the other # revision works better. for path, file_id, kind in delta.added: # afaict base_{path,rev} *should* be ignored for this one. yield path, kindmap[kind], self.ADD, None, None for path, file_id, kind in delta.removed: yield (path, kindmap[kind], self.DELETE, path, self.bzr_repo.string_rev(self.branch, parent_revid)) for oldpath, newpath, file_id, kind, textmod, metamod in delta.renamed: yield (newpath, kindmap[kind], self.MOVE, other.id2path(file_id), self.bzr_repo.string_rev(self.branch, parent_revid)) for path, file_id, kind, textmod, metamod in delta.modified: # "path" may not be accurate for base_path: the directory # it is in may have been renamed. So pull the right path from # the old inventory. yield (path, kindmap[kind], self.EDIT, other.id2path(file_id), self.bzr_repo.string_rev(self.branch, parent_revid)) def getattribute(self, attr): """If a callable is requested log the call.""" obj = object.__getattribute__(self, attr) if not callable(obj): return obj try: log = object.__getattribute__(self, 'log') except AttributeError: return obj if log is None: return obj else: def _wrap(*args, **kwargs): arg_list = [repr(a) for a in args] + \ ['%s=%r' % i for i in kwargs.iteritems() ] call_str = '%s.%s(%s)' % \ (self, attr, ', '.join(arg_list)) if len(call_str) > 200: call_str = '%s...' % call_str[:200] log.debug('CALL %s' % call_str) try: result = obj(*args, **kwargs) except Exception, e: log.debug('FAILURE of %s: %s' % (call_str, e) ) raise str_result = str(result) if len(str_result) > 200: str_result = '%s...' % str_result[:200] log.debug('RESULT of %s: %s' % (call_str, str_result)) return result return _wrap def containing_branch(transport, path): child_transport = transport.clone(path) my_bzrdir, relpath = \ bzrdir.BzrDir.open_containing_from_transport(child_transport) return my_bzrdir.open_branch(), relpath class BranchCache(object): def __init__(self, bzr_repo, branch): self.bzr_repo = bzr_repo self.branch = branch self._sorted_revision_history = None self._dotted_revno = None self._revno_revid = None def sorted_revision_history(self): if self._sorted_revision_history is None: self._sorted_revision_history = \ sorted_revision_history(self.branch, generate_revno=True) return self._sorted_revision_history def _populate_dotted_maps(self): if self._dotted_revno is None: self._dotted_revno = {} self._revno_revid = {} for s, revision_id, m, revno, e in \ self.sorted_revision_history(): dotted = '.'.join([str(s) for s in revno]) self._dotted_revno[revision_id] = dotted self._revno_revid[dotted] = revision_id def dotted_revno(self, revid): self._populate_dotted_maps() return self._dotted_revno.get(revid) def revid_from_dotted(self, dotted_revno): self._populate_dotted_maps() return self._revno_revid.get(dotted_revno) def revision_tree(self, revision_id): if revision_id not in self.bzr_repo._tree_cache: self.bzr_repo._tree_cache[revision_id] = \ self.branch.repository.revision_tree(revision_id) if revision_id == 'null:': self.bzr_repo._tree_cache[None] = \ self.bzr_repo._tree_cache['null:'] return self.bzr_repo._tree_cache[revision_id] def revision_trees(self, revision_ids): if None in revision_ids or 'null:' in revision_ids: self.revision_tree('null:') missing = [r for r in revision_ids if r not in self.bzr_repo._tree_cache] if len(missing) > 0: trees = self.branch.repository.revision_trees(missing) for tree in trees: self.bzr_repo._tree_cache[tree.get_revision_id()] = tree return [self.bzr_repo._tree_cache[r] for r in revision_ids] def cache_revisions(self, revision_ids): if self.bzr_repo.log: self.bzr_repo.log.debug('caching %d revisions' % len(revision_ids)) missing = [r for r in revision_ids if r not in self.bzr_repo._revision_cache] revisions = self.branch.repository.get_revisions(missing) self.bzr_repo._revision_cache.update(dict((r.revision_id, r) for r in revisions)) if self.bzr_repo.log: self.bzr_repo.log.debug('done caching %d revisions' % len(revision_ids)) def get_revision(self, revision_id): try: return self.bzr_repo._revision_cache[revision_id] except KeyError: self.cache_revisions([revision_id]) return self.bzr_repo._revision_cache[revision_id]