1 import os, os.path as osp

     2 import signal

     3 from tempfile import mkdtemp, NamedTemporaryFile

     4 import tempfile

     5 from Queue import Queue, Empty

     6 from subprocess import Popen, check_call

     7 from shutil import rmtree, copy as copyfile

     8 from uuid import uuid4 

     9 

    10 # imported by default to simplify further import statements

    11 from logilab.common.testlib import unittest_main, with_tempdir, InnerTest

    12 

    13 import os

    14 import cubicweb

    15 from cubicweb.view import StartupView

    16 from cubicweb.web.controller import Controller

    17 from cubicweb.devtools.httptest import CubicWebServerTC

    18 

    19 class FirefoxHelper(object):

    20 

    21     profile_name_mask = 'PYTEST_PROFILE_%(uid)s'

    22 

    23     def __init__(self, url=None):

    24         self._process = None

    25         self._tmp_dir = mkdtemp()

    26         self._profile_data = {'uid': uuid4()}

    27         self._profile_name = self.profile_name_mask % self._profile_data

    28         fnull = open(os.devnull, 'w')

    29         check_call(['firefox', '-no-remote', '-CreateProfile',

    30                     '%s %s' % (self._profile_name, self._tmp_dir)],

    31                               stdout=fnull, stderr=fnull)

    32         if url is not None:

    33             self.start(url)

    34 

    35 

    36     def start(self, url):

    37         self.stop()

    38         fnull = open(os.devnull, 'w')

    39         self._process = Popen(['firefox', '-no-remote', '-P', self._profile_name, url],

    40                               stdout=fnull, stderr=fnull)

    41 

    42     def stop(self):

    43         if self._process is not None:

    44             os.kill(self._process.pid, signal.SIGTERM)

    45             self._process.wait()

    46             self._process = None

    47 

    48     def __del__(self):

    49         self.stop()

    50         rmtree(self._tmp_dir)

    51 

    52 

    53 class QUnitTestCase(CubicWebServerTC):

    54 

    55     # testfile, (dep_a, dep_b)

    56     all_js_tests = ()

    57 

    58     def setUp(self):

    59         super(QUnitTestCase, self).setUp()

    60         self.test_queue = Queue()

    61         class MyQUnitResultController(QUnitResultController):

    62             tc = self

    63             test_queue = self.test_queue

    64         self._qunit_controller = MyQUnitResultController

    65         self.vreg.register(MyQUnitResultController)

    66 

    67     def tearDown(self):

    68         super(QUnitTestCase, self).tearDown()

    69         self.vreg.unregister(self._qunit_controller)

    70 

    71 

    72     def abspath(self, path):

    73         """use self.__module__ to build absolute path if necessary"""

    74         if not osp.isabs(path):

    75            dirname = osp.dirname(__import__(self.__module__).__file__)

    76            return osp.abspath(osp.join(dirname,path))

    77         return path

    78 

    79 

    80 

    81     def test_javascripts(self):

    82         for args in self.all_js_tests:

    83             test_file = self.abspath(args[0])

    84             if len(args) > 1:

    85                 depends   = [self.abspath(dep) for dep in args[1]]

    86             else:

    87                 depends = ()

    88             if len(args) > 2:

    89                 data   = [self.abspath(data) for data in args[2]]

    90             else:

    91                 data = ()

    92             for js_test in self._test_qunit(test_file, depends, data):

    93                 yield js_test

    94 

    95     @with_tempdir

    96     def _test_qunit(self, test_file, depends=(), data_files=(), timeout=30):

    97         assert osp.exists(test_file), test_file

    98         for dep in depends:

    99             assert osp.exists(dep), dep

   100         for data in data_files:

   101             assert osp.exists(data), data

   102 

   103 

   104         # generate html test file

   105         html_test_file = NamedTemporaryFile(suffix='.html')

   106         html_test_file.write(make_qunit_html(test_file, depends,

   107                              server_data=(self.test_host, self.test_port)))

   108         html_test_file.flush()

   109         # copying data file

   110         for data in data_files:

   111             copyfile(data, tempfile.tempdir)

   112 

   113         while not self.test_queue.empty():

   114             self.test_queue.get(False)

   115 

   116         browser = FirefoxHelper()

   117         browser.start(html_test_file.name)

   118         test_count = 0

   119         error = False

   120         def raise_exception(cls, *data):

   121             raise cls(*data)

   122         while not error:

   123             try:

   124                 result, test_name, msg = self.test_queue.get(timeout=timeout)

   125                 test_name = '%s (%s)' % (test_name, test_file)

   126                 self.set_description(test_name)

   127                 if result is None:

   128                     break

   129                 test_count += 1

   130                 if result:

   131                     yield InnerTest(test_name, lambda : 1)

   132                 else:

   133                     yield InnerTest(test_name, self.fail, msg)

   134             except Empty:

   135                 error = True

   136                 yield InnerTest(test_file, raise_exception, RuntimeError, "%s did not report execution end. %i test processed so far." % (test_file, test_count))

   137 

   138         browser.stop()

   139         if test_count <= 0 and not error:

   140             yield InnerTest(test_name, raise_exception, RuntimeError, 'No test yielded by qunit for %s' % test_file)

   141 

   142 class QUnitResultController(Controller):

   143 

   144     __regid__ = 'qunit_result'

   145 

   146 

   147     # Class variables to circumvent the instantiation of a new Controller for each request.

   148     _log_stack = [] # store QUnit log messages

   149     _current_module_name = '' # store the current QUnit module name

   150 

   151     def publish(self, rset=None):

   152         event = self._cw.form['event']

   153         getattr(self, 'handle_%s' % event)()

   154 

   155     def handle_module_start(self):

   156         self.__class__._current_module_name = self._cw.form.get('name', '')

   157 

   158     def handle_test_done(self):

   159         name = '%s // %s' %  (self._current_module_name, self._cw.form.get('name', ''))

   160         failures = int(self._cw.form.get('failures', 0))

   161         total = int(self._cw.form.get('total', 0))

   162 

   163         self._log_stack.append('%i/%i assertions failed' % (failures, total))

   164         msg = '\n'.join(self._log_stack)

   165 

   166         if failures:

   167             self.tc.test_queue.put((False, name, msg))

   168         else:

   169             self.tc.test_queue.put((True, name, msg))

   170         self._log_stack[:] = []

   171 

   172     def handle_done(self):

   173         self.tc.test_queue.put((None, None, None))

   174 

   175     def handle_log(self):

   176         result = self._cw.form['result']

   177         message = self._cw.form['message']

   178         self._log_stack.append('%s: %s' % (result, message))

   179 

   180 

   181 

   182 def cw_path(*paths):

   183   return file_path(osp.join(cubicweb.CW_SOFTWARE_ROOT, *paths))

   184 

   185 def file_path(path):

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

   187 

   188 def build_js_script( host, port):

   189     return """

   190     var host = '%s';

   191     var port = '%s';

   192 

   193     QUnit.moduleStart = function (name) {

   194       jQuery.ajax({

   195                   url: 'http://'+host+':'+port+'/qunit_result',

   196                  data: {"event": "module_start",

   197                         "name": name},

   198                  async: false});

   199     }

   200 

   201     QUnit.testDone = function (name, failures, total) {

   202       jQuery.ajax({

   203                   url: 'http://'+host+':'+port+'/qunit_result',

   204                  data: {"event": "test_done",

   205                         "name": name,

   206                         "failures": failures,

   207                         "total":total},

   208                  async: false});

   209     }

   210 

   211     QUnit.done = function (failures, total) {

   212       jQuery.ajax({

   213                    url: 'http://'+host+':'+port+'/qunit_result',

   214                    data: {"event": "done",

   215                           "failures": failures,

   216                           "total":total},

   217                    async: false});

   218       window.close();

   219     }

   220 

   221     QUnit.log = function (result, message) {

   222       jQuery.ajax({

   223                    url: 'http://'+host+':'+port+'/qunit_result',

   224                    data: {"event": "log",

   225                           "result": result,

   226                           "message": message},

   227                    async: false});

   228     }

   229     """ % (host, port)

   230 

   231 def make_qunit_html(test_file, depends=(), server_data=None):

   232     """"""

   233     data = {

   234             'web_data': cw_path('web', 'data'),

   235             'web_test': cw_path('web', 'test', 'jstests'),

   236         }

   237 

   238     html = ['''<html>

   239   <head>

   240     <!-- JS lib used as testing framework -->

   241     <link rel="stylesheet" type="text/css" media="all" href="%(web_test)s/qunit.css" />

   242     <script src="%(web_data)s/jquery.js" type="text/javascript"></script>

   243     <script src="%(web_test)s/qunit.js" type="text/javascript"></script>'''

   244     % data]

   245     if server_data is not None:

   246         host, port = server_data

   247         html.append('<!-- result report tools -->')

   248         html.append('<script type="text/javascript">')

   249         html.append(build_js_script(host, port))

   250         html.append('</script>')

   251     html.append('<!-- Test script dependencies (tested code for example) -->')

   252 

   253     for dep in depends:

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

   255 

   256     html.append('    <!-- Test script itself -->')

   257     html.append('    <script src="%s" type="text/javascript"></script>'% (file_path(test_file),))

   258     html.append('''  </head>

   259   <body>

   260     <div id="main">

   261     </div>

   262     <h1 id="qunit-header">QUnit example</h1>

   263     <h2 id="qunit-banner"></h2>

   264     <h2 id="qunit-userAgent"></h2>

   265     <ol id="qunit-tests">

   266   </body>

   267 </html>''')

   268     return u'\n'.join(html)

   269 

   270 

   271 

   272 

   273 

   274 

   275 

   276 if __name__ == '__main__':

   277     unittest_main()