source: mergebot/trunk/utils/test.py @ 57

#!/usr/bin/python
"""Automated tests for MergeBot

Run from a Trac source tree with mergebot installed system-wide.  (This needs
to be reworked to be less cumbersome.)
"""

import os
import unittest
import time
import shutil

from subprocess import call, Popen #, PIPE, STDOUT
from twill.errors import TwillAssertionError


from trac.tests.functional import FunctionalTestSuite, FunctionalTester, FunctionalTwillTestCaseSetup, tc, b, logfile
from trac.tests.functional.svntestenv import SvnFunctionalTestEnvironment
from trac.tests.contentgen import random_page, random_sentence #, random_word


#class MergeBotTestEnvironment(FunctionalTestEnvironment):
#    """Slight change to FunctionalTestEnvironment to keep the PYTHONPATH from
#    our environment.
#    """
#    def start(self):
#        """Starts the webserver"""
#        server = Popen(["python", "./trac/web/standalone.py",
#                        "--port=%s" % self.port, "-s",
#                        "--basic-auth=trac,%s," % self.htpasswd,
#                        self.tracdir],
#                       #env={'PYTHONPATH':'.'},
#                       stdout=logfile, stderr=logfile,
#                      )
#        self.pid = server.pid
#        time.sleep(1) # Give the server time to come up
#
#    def _tracadmin(self, *args):
#        """Internal utility method for calling trac-admin"""
#        if call(["python", "./trac/admin/console.py", self.tracdir] +
#                list(args),
#                #env={'PYTHONPATH':'.'},
#                stdout=logfile, stderr=logfile):
#            raise Exception('Failed running trac-admin with %r' % (args, ))
#
#
#FunctionalTestEnvironment = MergeBotTestEnvironment


class MergeBotFunctionalTester(FunctionalTester):
    """Adds some MergeBot functionality to the functional tester."""
    # FIXME: the tc.find( <various actions> ) checks are bogus: any ticket can
    # satisfy them, not just the one we're working on.
    def __init__(self, trac_url, repo_url):
        FunctionalTester.__init__(self, trac_url)
        self.repo_url = repo_url
        self.mergeboturl = self.url + '/mergebot'

    def wait_until_find(self, search, timeout=5):
        start = time.time()
        while time.time() - start < timeout:
            try:
                #tc.reload() # This appears to re-POST
                tc.go(b.get_url())
                tc.find(search)
                return
            except TwillAssertionError:
                pass
        raise TwillAssertionError("Unable to find %r within %s seconds" % (search, timeout))

    def wait_until_notfind(self, search, timeout=5):
        start = time.time()
        while time.time() - start < timeout:
            try:
                #tc.reload() # This appears to re-POST
                tc.go(b.get_url())
                tc.notfind(search)
                return
            except TwillAssertionError:
                pass
        raise TwillAssertionError("Unable to notfind %r within %s seconds" % (search, timeout))

    def go_to_mergebot(self):
        tc.go(self.mergeboturl)
        tc.url(self.mergeboturl)
        tc.notfind('No handler matched request to /mergebot')

    def branch(self, ticket_id, component, timeout=1):
        """timeout is in seconds."""
        self.go_to_mergebot()
        tc.formvalue('ops-%s' % ticket_id, 'ticket', ticket_id) # Essentially a noop to select the right form
        tc.submit('Branch')
        self.wait_until_find('Nothing in the queue', timeout)
        tc.find('Rebranch')
        tc.find('Merge')
        tc.find('CheckMerge')
        self.go_to_ticket(ticket_id)
        tc.find('Created branch from .* for .*')
        retval = call(['svn', 'ls', self.repo_url + '/' + component + '/branches/ticket-%s' % ticket_id],
                    stdout=logfile, stderr=logfile)
        if retval:
            raise Exception('svn ls failed with exit code %s' % retval)

    def _rebranch(self, ticket_id, component, search, timeout=15):
        """timeout is in seconds."""
        self.go_to_mergebot()
        tc.formvalue('ops-%s' % ticket_id, 'ticket', ticket_id) # Essentially a noop to select the right form
        tc.submit('Rebranch')
        self.wait_until_find('Nothing in the queue', timeout)
        tc.find('Rebranch')
        tc.find('Merge')
        tc.find('CheckMerge')
        self.go_to_ticket(ticket_id)
        tc.find(search)
        retval = call(['svn', 'ls', self.repo_url + '/' + component + '/branches/ticket-%s' % ticket_id],
                    stdout=logfile, stderr=logfile)
        if retval:
            raise Exception('svn ls failed with exit code %s' % retval)

    def rebranch(self, ticket_id, component, timeout=15):
        self._rebranch(ticket_id, component, 'Rebranched from .* for .*', timeout)

    def rebranch_conflict(self, ticket_id, component, timeout=15):
        self._rebranch(ticket_id, component, 'There were conflicts on rebranching', timeout)

    def merge(self, ticket_id, component, timeout=5):
        """timeout is in seconds."""
        self.go_to_mergebot()
        tc.formvalue('ops-%s' % ticket_id, 'ticket', ticket_id) # Essentially a noop to select the right form
        tc.submit('Merge')
        self.wait_until_find('Nothing in the queue', timeout)
        tc.find('Branch')
        self.go_to_ticket(ticket_id)
        tc.find('Merged .* to .* for')
        # TODO: We may want to change this to remove the "dead" branch
        retval = call(['svn', 'ls', self.repo_url + '/' + component + '/branches/ticket-%s' % ticket_id],
                    stdout=logfile, stderr=logfile)
        if retval:
            raise Exception('svn ls failed with exit code %s' % retval)

    def checkmerge(self, ticket_id, component, timeout=5):
        """timeout is in seconds."""
        self.go_to_mergebot()
        tc.formvalue('ops-%s' % ticket_id, 'ticket', ticket_id) # Essentially a noop to select the right form
        tc.submit('CheckMerge')
        self.wait_until_find('Nothing in the queue', timeout)
        tc.find('Rebranch')
        tc.find('Merge')
        tc.find('CheckMerge')
        self.go_to_ticket(ticket_id)
        tc.find('while checking merge of')
        # TODO: We may want to change this to remove the "dead" branch
        retval = call(['svn', 'ls', self.repo_url + '/' + component + '/branches/ticket-%s' % ticket_id],
                    stdout=logfile, stderr=logfile)
        if retval:
            raise Exception('svn ls failed with exit code %s' % retval)


class MergeBotTestSuite(FunctionalTestSuite):
    def setUp(self):
        port = 8889
        baseurl = "http://localhost:%s" % port
        self._testenv = SvnFunctionalTestEnvironment("testenv%s" % port, port, baseurl)

        # Configure mergebot
        env = self._testenv.get_trac_environment()
        env.config.set('components', 'mergebot.web_ui.mergebotmodule', 'enabled')
        env.config.save()
        os.mkdir(os.path.join("testenv%s" % port, 'trac', 'mergebot'))
        self._testenv._tracadmin('upgrade') # sets up the bulk of the mergebot config
        env.config.parse_if_needed()
        env.config.set('mergebot', 'repository_url', self._testenv.repo_url())
        env.config.set('logging', 'log_type', 'file')
        env.config.save()
        env.config.parse_if_needed()

        self._testenv.start()
        self._tester = MergeBotFunctionalTester(baseurl, self._testenv.repo_url())
        os.system('mergebotdaemon -f "%s" > %s/mergebotdaemon.log 2>&1 &' % (self._testenv.tracdir, self._testenv.tracdir))
        self.fixture = (self._testenv, self._tester)

        # Setup some common component stuff for MergeBot's use:
        svnurl = self._testenv.repo_url()
        for component in ['stuff', 'flagship', 'submarine']:
            self._tester.create_component(component)
            if call(['svn', '-m', 'Create tree for "%s".' % component, 'mkdir',
                     svnurl + '/' + component,
                     svnurl + '/' + component + '/trunk',
                     svnurl + '/' + component + '/tags',
                     svnurl + '/' + component + '/branches'],
                    stdout=logfile, stderr=logfile):
                raise Exception("svn mkdir failed")

        self._tester.create_version('trunk')


class FunctionalSvnTestCaseSetup(FunctionalTwillTestCaseSetup):
    def get_workdir(self):
        return os.path.join(self._testenv.dirname, self.__class__.__name__)

    def checkout(self, ticket_id=None):
        """checkout a working copy of the branch for the given ticket, or trunk if none given"""
        if ticket_id is None:
            svnurl = self._testenv.repo_url() + '/stuff/trunk'
        else:
            svnurl = self._testenv.repo_url() + '/stuff/branches/ticket-%s' % ticket_id
        retval = call(['svn', 'checkout', svnurl, self.get_workdir()],
            stdout=logfile, stderr=logfile)
        self.assertEqual(retval, 0, "svn checkout failed with error %s" % (retval))

    def switch(self, ticket_id=None):
        if ticket_id is None:
            svnurl = self._testenv.repo_url() + '/stuff/trunk'
        else:
            svnurl = self._testenv.repo_url() + '/stuff/branches/ticket-%s' % ticket_id
        retval = call(['svn', 'switch', svnurl, self.get_workdir()],
            stdout=logfile, stderr=logfile)
        self.assertEqual(retval, 0, "svn checkout failed with error %s" % (retval))

    def add_new_file(self, filename=None):
        workdir = self.get_workdir()
        if filename is None:
            newfile = os.path.join(workdir, self.__class__.__name__)
        else:
            newfile = os.path.join(workdir, filename)
        open(newfile, 'w').write(random_page())
        retval = call(['svn', 'add', newfile],
            cwd=workdir,
            stdout=logfile, stderr=logfile)
        self.assertEqual(retval, 0, "svn add failed with error %s" % (retval))

    def commit(self, message, files=None):
        if files is None:
            files = ['.']
        commit_message = self.__class__.__name__ + ": " + message
        retval = call(['svn', 'commit', '-m', commit_message] + list(files),
            cwd=self.get_workdir(),
            stdout=logfile, stderr=logfile)
        self.assertEqual(retval, 0, "svn commit failed with error %s" % (retval))

    def mv(self, oldname, newname):
        retval = call(['svn', 'mv', oldname, newname],
            cwd=self.get_workdir(),
            stdout=logfile, stderr=logfile)
        self.assertEqual(retval, 0, "svn mv failed with error %s" % (retval))


class MergeBotTestEnabled(FunctionalTwillTestCaseSetup):
    def runTest(self):
        self._tester.logout()
        tc.go(self._tester.url)
        self._tester.login('admin')
        tc.follow('MergeBot')
        mergeboturl = self._tester.url + '/mergebot'
        tc.url(mergeboturl)
        tc.notfind('No handler matched request to /mergebot')


class MergeBotTestNoVersion(FunctionalTwillTestCaseSetup):
    """Verify that if a ticket does not have the version field set, it will not
    appear in the MergeBot list.
    """
    def runTest(self):
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':''})
        tc.follow('MergeBot')
        tc.notfind(self.__class__.__name__)


class MergeBotTestBranch(FunctionalTwillTestCaseSetup):
    def runTest(self):
        """Verify that the 'branch' button works"""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        self._tester.branch(ticket_id, 'stuff')


class MergeBotTestRebranch(FunctionalTwillTestCaseSetup):
    def runTest(self):
        """Verify that the 'rebranch' button works"""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        self._tester.branch(ticket_id, 'stuff')
        self._tester.rebranch(ticket_id, 'stuff')


class MergeBotTestMerge(FunctionalTwillTestCaseSetup):
    def runTest(self):
        """Verify that the 'merge' button works"""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        self._tester.branch(ticket_id, 'stuff')
        self._tester.merge(ticket_id, 'stuff')


class MergeBotTestCheckMerge(FunctionalTwillTestCaseSetup):
    def runTest(self):
        """Verify that the 'checkmerge' button works"""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        self._tester.branch(ticket_id, 'stuff')
        self._tester.checkmerge(ticket_id, 'stuff')


class MergeBotTestRebranchWithChange(FunctionalSvnTestCaseSetup):
    def runTest(self):
        """Verify that the 'rebranch' button works with changes on the branch"""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        self._tester.branch(ticket_id, 'stuff')

        # checkout a working copy & make a change
        self.checkout(ticket_id)
        # Create & add a new file
        self.add_new_file()
        self.commit('Add a new file')

        self._tester.rebranch(ticket_id, 'stuff')


class MergeBotTestRebranchWithChangeAndTrunkChange(FunctionalSvnTestCaseSetup):
    def runTest(self):
        """Verify that the 'rebranch' button works with changes on the branch and trunk"""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        self._tester.branch(ticket_id, 'stuff')

        # checkout a working copy & make a change
        self.checkout(ticket_id)
        # Create & add a new file
        basename = self.__class__.__name__
        self.add_new_file(basename + '-ticket')
        self.commit('Add a new file on ticket')
        self.switch()
        self.add_new_file(basename + '-trunk')
        self.commit('Add a new file on trunk')

        self._tester.rebranch(ticket_id, 'stuff')


class MergeBotTestRebranchWithConflict(FunctionalSvnTestCaseSetup):
    def runTest(self):
        """Verify that the 'rebranch' button works with conflicts on the branch and trunk"""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        basename = self.__class__.__name__

        # create a file in which to have conflicts
        self.checkout()
        self.add_new_file(basename)
        self.commit('Add a new file on trunk')

        # create the branch
        self._tester.branch(ticket_id, 'stuff')

        # modify the file on trunk
        open(os.path.join(self.get_workdir(), basename), 'a').write(random_sentence())
        self.commit('Modify the file on trunk')

        # modify the file on the branch
        self.switch(ticket_id)
        open(os.path.join(self.get_workdir(), basename), 'a').write(random_sentence())
        self.commit('Modify the file on branch')

        # rebranch, make sure it shows a conflict
        self._tester.rebranch_conflict(ticket_id, 'stuff')


class MergeBotTestRebranchWithBranchRenameConflict(FunctionalSvnTestCaseSetup):
    def runTest(self):
        """Verify that the 'rebranch' button works when a file renamed on the branch was modified on trunk"""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        basename = self.__class__.__name__

        # create a file in which to have conflicts
        self.checkout()
        self.add_new_file(basename)
        self.commit('Add a new file on trunk')

        # create the branch
        self._tester.branch(ticket_id, 'stuff')

        # modify the file on trunk
        open(os.path.join(self.get_workdir(), basename), 'a').write(random_sentence())
        self.commit('Modify the file on trunk')

        # rename the file on the branch
        self.switch(ticket_id)
        self.mv(basename, basename + '-renamed')
        self.commit('Rename the file on branch')

        self._tester.rebranch_conflict(ticket_id, 'stuff')


class MergeBotTestSingleUseCase(FunctionalTwillTestCaseSetup):
    def runTest(self):
        """Create a branch, make a change, checkmerge, and merge it."""
        ticket_id = self._tester.create_ticket(summary=self.__class__.__name__,
            info={'component':'stuff', 'version':'trunk'})
        self._tester.branch(ticket_id, 'stuff')
        # checkout a working copy & make a change
        svnurl = self._testenv.repo_url()
        workdir = os.path.join(self._testenv.dirname, self.__class__.__name__)
        retval = call(['svn', 'checkout', svnurl + '/stuff/branches/ticket-%s' % ticket_id, workdir],
            stdout=logfile, stderr=logfile)
        self.assertEqual(retval, 0, "svn checkout failed with error %s" % (retval))
        # Create & add a new file
        newfile = os.path.join(workdir, self.__class__.__name__)
        open(newfile, 'w').write(random_page())
        retval = call(['svn', 'add', self.__class__.__name__],
            cwd=workdir,
            stdout=logfile, stderr=logfile)
        self.assertEqual(retval, 0, "svn add failed with error %s" % (retval))
        retval = call(['svn', 'commit', '-m', 'Add a new file', self.__class__.__name__],
            cwd=workdir,
            stdout=logfile, stderr=logfile)
        self.assertEqual(retval, 0, "svn commit failed with error %s" % (retval))

        self._tester.checkmerge(ticket_id, 'stuff')
        self._tester.merge(ticket_id, 'stuff')

        shutil.rmtree(workdir) # cleanup working copy


def suite():
    suite = MergeBotTestSuite()
    suite.addTest(MergeBotTestEnabled())
    suite.addTest(MergeBotTestNoVersion())
    suite.addTest(MergeBotTestBranch())
    suite.addTest(MergeBotTestRebranch())
    suite.addTest(MergeBotTestCheckMerge())
    suite.addTest(MergeBotTestMerge())
    suite.addTest(MergeBotTestRebranchWithChange())
    suite.addTest(MergeBotTestRebranchWithChangeAndTrunkChange())
    suite.addTest(MergeBotTestRebranchWithConflict())
    suite.addTest(MergeBotTestRebranchWithBranchRenameConflict())
    suite.addTest(MergeBotTestSingleUseCase())
    return suite

if __name__ == '__main__':
    unittest.main(defaultTest='suite')