Source

django-triager-bot / djangotriager / trac / tickets.py

from djangotriager.trac.server import get_proxy
from djangotriager.patches import Patch

from djangotriager.tests import TestRunner
from djangotriager.reporter import Reporter

class Ticket(object):
    """
    Class representing a SCM ticket.
    """
    def __init__(self, ticket_id):
        self._ticket_id = int(ticket_id)
        self.__patches = []
        self._reporter = Reporter()
        self._reporter.info("=== REPORT FOR TICKET #%d ===" % self._ticket_id)

    def send_report(self, config):
        """Send the generated report to a ticket's comment."""
        proxy = get_proxy(config.get("trac"))
        proxy.ticket.update(
            self._ticket_id, ("""
'''This ticket has been checked through django-triager-bot, below is the
report:'''

{{{
%s
}}}""" % self.get_report()))

    def get_report(self):
        """Print the report generated so far.

        You can use this to retrieve the report instead of sending to the
        ticket's page at the SCM instance.
        """
        return self._reporter.get_buffer()

    @classmethod
    def get_by_query(cls, config, query):
        """
        Retrieve all tickets that matches the query string given.
        """
        proxy = get_proxy(config.get("trac"))
        return [cls(ticket_id) for ticket_id in proxy.ticket.query(query)]

    def get_patches(self, config):
        """
        Return all patches available for the ticket.
        """
        if not self.__patches:
            proxy = get_proxy(config.get("trac"))
            attachments = proxy.ticket.listAttachments(self._ticket_id)
            getAttachment = proxy.ticket.getAttachment
            self.__patches = \
                [Patch(file[0], str(getAttachment(self._ticket_id, file[0])))
                     for file in attachments if Patch.validate(file[0])]
            self._reporter.warn("Patches retrieved: %r" % self.__patches)
        return self.__patches

    def try_patch(self, wc, patch):
        """Check if the given patch is valid.

        The patch is applied in the working copy and validate through the test
        suite, after this the changes in the working copy is reverted.
        """
        self._reporter.info("Trying patch %r" % patch)
        self._reporter.info("Applying patch")
        is_valid = False
        if not patch.apply(wc.dir, logger=self._reporter):
            self._reporter.warn("Patch %r could not be applied on r%s"
                                % (patch, wc.get_revision()))
        else:
            self._reporter.warn("Patch %r applied successfully on r%s"
                                % (patch, wc.get_revision()))
            self._reporter.info("Running tests")
            is_valid = TestRunner(wc).run(logger=self._reporter)
        self._reporter.info("Restoring working copy")
        wc.revert_changes()
        return is_valid

    def try_patches(self, config, wc, limit=None):
        patches = self.get_patches(config)
        if not patches:
            self._reporter.info("No patch to process")
            return
        for idx, patch in enumerate(reversed(patches)):
            self.try_patch(config, patch)
            if limit and idx == limit - 1:
                self._reporter.warn(
                    "Limit of patches (%d) reached, finishing" % limit)
                return

    def apply_patches(self, config, wc, limit=None):
        patches = self.get_patches(config)
        if not patches:
            self._reporter.info("No patch to process")
            return
        for idx, patch in enumerate(reversed(patches)):
            self._reporter.info("Applying patch")
            if not patch.apply(wc.dir, logger=self._reporter):
                self._reporter.warn("Patch %r could not be applied on r%s"
                                    % (patch, wc.get_revision()))

            else:
                self._reporter.warn("Patch %r applied successfully on r%s"
                                    % (patch, wc.get_revision()))
            self._reporter.info("Restoring working copy")
            wc.revert_changes()
            if limit and idx == limit - 1:
                self._reporter.warn(
                    "Limit of patches (%d) reached, finishing" % limit)
                return