source: tracdarcs/changeset.py @ 3

# -*- coding: iso-8859-1 -*-
#
# Copyright (C) 2005 Edgewall Software
# Copyright (C) 2005,2006 Lele Gaifax <lele@metapensiero.it>
#
# This software is licensed as described in the file COPYING, which
# you should have received as part of this distribution. The terms
# are also available at http://trac.edgewall.com/license.html.
#
# This software consists of voluntary contributions made by many
# individuals. For the exact contribution history, see the revision
# history and logs, available at http://projects.edgewall.com/trac/.
#
# Author: Lele Gaifax <lele@metapensiero.it>

from time import mktime, timezone

from xml.sax import parseString, SAXException
from xml.sax.handler import ContentHandler

from trac.util import TracError
from trac.versioncontrol import Changeset
from trac.versioncontrol.cache import CachedChangeset

from node import Node, DarcsNode
from repos import reversed

class DarcsChangeset(Changeset):
    """
    Represents a set of changes of a repository.
    """

    def __init__(self, rev, patchname, message, author, date, changes, hash):
        if message:
            log = patchname + '\n' + message
        else:
            log = patchname
        Changeset.__init__(self, rev, log, author, date)
        self.patchname = patchname
        self.changes = changes
        self.hash = hash
        # fix up changes rev slot
        for c in self.changes:
            c.rev = rev

    def get_changes(self):
        """
        Generator that produces a (path, kind, change, base_path, base_rev)
        """

        moves = {}
        for c in self.changes:
            last = c.get_history(limit=2)
            try:
                last.next()
                basepath,baserev,basechg = last.next()
            except StopIteration:
                basepath = None
                baserev = -1
            yield (c.path, c.kind, c.change, c.oldpath or basepath, baserev)

    def get_node(self, path, maybedir=False):
        """
        Find and return the node relative to given path.
        """

        for c in self.changes:
            if c.path == path or c.oldpath == path:
                return c
            if maybedir and not path.endswith('/'):
                path += '/'
                if c.path == path or c.oldpath == path:
                    return c

    def insert_in_cache(self, cursor, kindmap, actionmap, log):
        """
        Augment standard metadata with darcs patch hash.
        """

        Changeset.insert_in_cache(self, cursor, kindmap, actionmap, log)
        cursor.execute("UPDATE revision SET hash = %s "
                       "WHERE rev = %s", (self.hash, self.rev))


def changesets_from_darcschanges(changes, repository, start_revision,
                                 force_encoding=None):
    """
    Parse XML output of ``darcs changes``.

    Return a list of ``Changeset`` instances.
    """

    class DarcsXMLChangesHandler(ContentHandler):
        def __init__(self):
            self.changesets = []
            self.index = start_revision-1
            self.current = None
            self.current_field = []

        def startElement(self, name, attributes):
            if name == 'patch':
                self.current = {}
                self.current['author'] = attributes['author']
                date = attributes['date']
                # 20040619130027
                y = int(date[:4])
                m = int(date[4:6])
                d = int(date[6:8])
                hh = int(date[8:10])
                mm = int(date[10:12])
                ss = int(date[12:14])
                unixtime = int(mktime((y, m, d, hh, mm, ss, 0, 0, 0)))-timezone
                self.current['date'] = unixtime
                self.current['comment'] = ''
                self.current['hash'] = attributes['hash']
                self.current['entries'] = []
            elif name in ['name', 'comment', 'add_file', 'add_directory',
                          'remove_directory', 'modify_file', 'remove_file']:
                self.current_field = []
            elif name == 'move':
                self.old_name = attributes['from']
                self.new_name = attributes['to']

        def endElement(self, name):
            if name == 'patch':
                # Sort the paths to make tests easier
                self.current['entries'].sort()
                self.index += 1
                cset = DarcsChangeset(self.index,
                                      self.current['name'],
                                      self.current['comment'],
                                      self.current['author'],
                                      self.current['date'],
                                      self.current['entries'],
                                      self.current['hash'])
                self.changesets.append(cset)
                self.current = None
            elif name in ['name', 'comment']:
                self.current[name] = ''.join(self.current_field)
            elif name == 'move':
                kind = None
                for cs in reversed(self.changesets):
                    node = cs.get_node(self.old_name, maybedir=True)
                    if node:
                        kind = node.kind
                        break
                if kind is None:
                    kind = repository._getNodeKind(self.old_name, self.index)
                if kind == Node.DIRECTORY:
                    self.new_name += '/'
                    self.old_name += '/'
                entry = DarcsNode(self.new_name, None, kind, Changeset.MOVE,
                                  repository, self.old_name)
                self.current['entries'].append(entry)
            elif name in ['add_file', 'add_directory', 'modify_file',
                          'remove_file', 'remove_directory']:
                path = ''.join(self.current_field).strip()
                change = { 'add_file': Changeset.ADD,
                           'add_directory': Changeset.ADD,
                           'modify_file': Changeset.EDIT,
                           'remove_file': Changeset.DELETE,
                           'remove_directory': Changeset.DELETE
                         }[name]
                isdir = name in ('add_directory', 'remove_directory')
                kind = isdir and Node.DIRECTORY or Node.FILE
                # Eventually add one final '/' to identify directories.
                # This is because Trac brings around simple tuples at times,
                # that cannot carry that flag with them.
                if isdir:
                    path += '/'
                entry = DarcsNode(path, None, kind, change, repository)
                self.current['entries'].append(entry)

        def characters(self, data):
            self.current_field.append(data)

    if force_encoding:
        changes = changes.decode(force_encoding, 'replace').encode('utf-8')
        changes = '<?xml version="1.0" encoding="utf-8"?>\n' + changes

    handler = DarcsXMLChangesHandler()
    try:
        parseString(changes, handler)
    except SAXException, le:
        raise TracError('Unable to parse "darcs changes" output: ' + str(le))

    return handler.changesets

class DarcsCachedChangeset(CachedChangeset):
    """
    Darcs version of the CachedChangeset that knows about the hash.
    """

    def __init__(self, rev, db, authz=None):
        CachedChangeset.__init__(self, rev, db, authz)
        cursor = self.db.cursor()
        cursor.execute("SELECT hash FROM revision "
                       "WHERE rev=%s", (rev,))
        row = cursor.fetchone()
        if row:
            self.hash = row[0]