devtools/qunit.py
author Pierre-Yves David <pierre-yves.david@logilab.fr>
Fri, 18 Jun 2010 18:16:36 +0200
changeset 5794 a6b81d106775
parent 5790 26680cbc507b
child 5795 7947b1af2e82
permissions -rw-r--r--
[devtools/qunit] Locate jquery through cubicwew config to handle installed setup

import os, os.path as osp
import signal
from tempfile import mkdtemp, NamedTemporaryFile, TemporaryFile
import tempfile
from Queue import Queue, Empty
from subprocess import Popen, check_call, CalledProcessError
from shutil import rmtree, copy as copyfile
from uuid import uuid4 

# imported by default to simplify further import statements
from logilab.common.testlib import unittest_main, with_tempdir, InnerTest

import os
import cubicweb
from cubicweb.view import StartupView
from cubicweb.web.controller import Controller
from cubicweb.devtools.httptest import CubicWebServerTC


class VerboseCalledProcessError(CalledProcessError):

    def __init__(self, returncode, command, stdout, stderr):
        super(VerboseCalledProcessError, self).__init__(returncode, command)
        self.stdout = stdout
        self.stderr = stderr

    def __str__(self):
        str = [ super(VerboseCalledProcessError, self).__str__()]
        if self.stdout.strip():
            str.append('******************')
            str.append('* process stdout *')
            str.append('******************')
            str.append(self.stdout)
        if self.stderr.strip():
            str.append('******************')
            str.append('* process stderr *')
            str.append('******************')
            str.append(self.stderr)
        return '\n'.join(str)



class FirefoxHelper(object):

    profile_name_mask = 'PYTEST_PROFILE_%(uid)s'

    def __init__(self, url=None):
        self._process = None
        self._tmp_dir = mkdtemp()
        self._profile_data = {'uid': uuid4()}
        self._profile_name = self.profile_name_mask % self._profile_data
        fnull = open(os.devnull, 'w')
        stdout = TemporaryFile()
        stderr = TemporaryFile()
        try:
          check_call(['firefox', '-no-remote', '-CreateProfile',
                      '%s %s' % (self._profile_name, self._tmp_dir)],
                                stdout=stdout, stderr=stderr)
        except CalledProcessError, cpe:
            stdout.seek(0)
            stderr.seek(0)
            raise VerboseCalledProcessError(cpe.returncode, cpe.cmd, stdout.read(), stderr.read())


    def start(self, url):
        self.stop()
        fnull = open(os.devnull, 'w')
        self._process = Popen(['firefox', '-no-remote', '-P', self._profile_name, url],
                              stdout=fnull, stderr=fnull)

    def stop(self):
        if self._process is not None:
            assert self._process.returncode is None,  self._process.returncode
            os.kill(self._process.pid, signal.SIGTERM)
            self._process.wait()
            self._process = None

    def __del__(self):
        self.stop()
        rmtree(self._tmp_dir)


class QUnitTestCase(CubicWebServerTC):

    # testfile, (dep_a, dep_b)
    all_js_tests = ()

    def setUp(self):
        super(QUnitTestCase, self).setUp()
        self.test_queue = Queue()
        class MyQUnitResultController(QUnitResultController):
            tc = self
            test_queue = self.test_queue
        self._qunit_controller = MyQUnitResultController
        self.vreg.register(MyQUnitResultController)

    def tearDown(self):
        super(QUnitTestCase, self).tearDown()
        self.vreg.unregister(self._qunit_controller)


    def abspath(self, path):
        """use self.__module__ to build absolute path if necessary"""
        if not osp.isabs(path):
           dirname = osp.dirname(__import__(self.__module__).__file__)
           return osp.abspath(osp.join(dirname,path))
        return path



    def test_javascripts(self):
        for args in self.all_js_tests:
            test_file = self.abspath(args[0])
            if len(args) > 1:
                depends   = [self.abspath(dep) for dep in args[1]]
            else:
                depends = ()
            if len(args) > 2:
                data   = [self.abspath(data) for data in args[2]]
            else:
                data = ()
            for js_test in self._test_qunit(test_file, depends, data):
                yield js_test

    @with_tempdir
    def _test_qunit(self, test_file, depends=(), data_files=(), timeout=30):
        assert osp.exists(test_file), test_file
        for dep in depends:
            assert osp.exists(dep), dep
        for data in data_files:
            assert osp.exists(data), data


        # generate html test file
        jquery_dir = self.config.locate_resource('jquery.js')
        html_test_file = NamedTemporaryFile(suffix='.html')
        html_test_file.write(make_qunit_html(test_file, depends,
                             server_data=(self.test_host, self.test_port),
                             web_data_path=jquery_dir))
        html_test_file.flush()
        # copying data file
        for data in data_files:
            copyfile(data, tempfile.tempdir)

        while not self.test_queue.empty():
            self.test_queue.get(False)

        browser = FirefoxHelper()
        browser.start(html_test_file.name)
        test_count = 0
        error = False
        def raise_exception(cls, *data):
            raise cls(*data)
        while not error:
            try:
                result, test_name, msg = self.test_queue.get(timeout=timeout)
                test_name = '%s (%s)' % (test_name, test_file)
                self.set_description(test_name)
                if result is None:
                    break
                test_count += 1
                if result:
                    yield InnerTest(test_name, lambda : 1)
                else:
                    yield InnerTest(test_name, self.fail, msg)
            except Empty:
                error = True
                yield InnerTest(test_file, raise_exception, RuntimeError, "%s did not report execution end. %i test processed so far." % (test_file, test_count))

        browser.stop()
        if test_count <= 0 and not error:
            yield InnerTest(test_name, raise_exception, RuntimeError, 'No test yielded by qunit for %s' % test_file)

class QUnitResultController(Controller):

    __regid__ = 'qunit_result'


    # Class variables to circumvent the instantiation of a new Controller for each request.
    _log_stack = [] # store QUnit log messages
    _current_module_name = '' # store the current QUnit module name

    def publish(self, rset=None):
        event = self._cw.form['event']
        getattr(self, 'handle_%s' % event)()

    def handle_module_start(self):
        self.__class__._current_module_name = self._cw.form.get('name', '')

    def handle_test_done(self):
        name = '%s // %s' %  (self._current_module_name, self._cw.form.get('name', ''))
        failures = int(self._cw.form.get('failures', 0))
        total = int(self._cw.form.get('total', 0))

        self._log_stack.append('%i/%i assertions failed' % (failures, total))
        msg = '\n'.join(self._log_stack)

        if failures:
            self.tc.test_queue.put((False, name, msg))
        else:
            self.tc.test_queue.put((True, name, msg))
        self._log_stack[:] = []

    def handle_done(self):
        self.tc.test_queue.put((None, None, None))

    def handle_log(self):
        result = self._cw.form['result']
        message = self._cw.form['message']
        self._log_stack.append('%s: %s' % (result, message))



def cw_path(*paths):
  return file_path(osp.join(cubicweb.CW_SOFTWARE_ROOT, *paths))

def file_path(path):
    return 'file://' + osp.abspath(path)

def build_js_script( host, port):
    return """
    var host = '%s';
    var port = '%s';

    QUnit.moduleStart = function (name) {
      jQuery.ajax({
                  url: 'http://'+host+':'+port+'/qunit_result',
                 data: {"event": "module_start",
                        "name": name},
                 async: false});
    }

    QUnit.testDone = function (name, failures, total) {
      jQuery.ajax({
                  url: 'http://'+host+':'+port+'/qunit_result',
                 data: {"event": "test_done",
                        "name": name,
                        "failures": failures,
                        "total":total},
                 async: false});
    }

    QUnit.done = function (failures, total) {
      jQuery.ajax({
                   url: 'http://'+host+':'+port+'/qunit_result',
                   data: {"event": "done",
                          "failures": failures,
                          "total":total},
                   async: false});
      window.close();
    }

    QUnit.log = function (result, message) {
      jQuery.ajax({
                   url: 'http://'+host+':'+port+'/qunit_result',
                   data: {"event": "log",
                          "result": result,
                          "message": message},
                   async: false});
    }
    """ % (host, port)

def make_qunit_html(test_file, depends=(), server_data=None,
                    web_data_path=cw_path('web', 'data')):
    """"""
    data = {
            'web_data': web_data_path,
            'web_test': cw_path('devtools', 'data'),
        }

    html = ['''<html>
  <head>
    <!-- JS lib used as testing framework -->
    <link rel="stylesheet" type="text/css" media="all" href="%(web_test)s/qunit.css" />
    <script src="%(web_data)s/jquery.js" type="text/javascript"></script>
    <script src="%(web_test)s/qunit.js" type="text/javascript"></script>'''
    % data]
    if server_data is not None:
        host, port = server_data
        html.append('<!-- result report tools -->')
        html.append('<script type="text/javascript">')
        html.append(build_js_script(host, port))
        html.append('</script>')
    html.append('<!-- Test script dependencies (tested code for example) -->')

    for dep in depends:
        html.append('    <script src="%s" type="text/javascript"></script>' % file_path(dep))

    html.append('    <!-- Test script itself -->')
    html.append('    <script src="%s" type="text/javascript"></script>'% (file_path(test_file),))
    html.append('''  </head>
  <body>
    <div id="main">
    </div>
    <h1 id="qunit-header">QUnit example</h1>
    <h2 id="qunit-banner"></h2>
    <h2 id="qunit-userAgent"></h2>
    <ol id="qunit-tests">
  </body>
</html>''')
    return u'\n'.join(html)







if __name__ == '__main__':
    unittest_main()