Source code for webtest_casperjs

# -*- coding: utf-8 -*-
from __future__ import unicode_literals
__doc__ = '''Allow to run an external process to test your application'''
from webtest import app as testapp
from webtest.http import StopableWSGIServer
from contextlib import contextmanager
from six import binary_type
import subprocess
import logging
import tempfile
import shutil
import time
import sys
import os

log = logging.getLogger('nose.casperjs')
stderr = sys.stderr

[docs]class TestApp(testapp.TestApp): """Run the test application in a separate thread to allow to access it via http""" def __init__(self, app=None, url=None, timeout=30000, extra_environ=None, relative_to=None, **kwargs): super(TestApp, self).__init__(app, relative_to=relative_to) self.server = StopableWSGIServer.create(app) self.server.wait() self.application_url = self.server.application_url os.environ['APPLICATION_URL'] = self.application_url self.extra_environ = extra_environ or {} self.timeout = timeout self.test_app = self def get_binary(self, name): for path in (os.getcwd(), '/usr/local', '/usr', '/opt'): filename = os.path.join(path, 'bin', name) if os.path.isfile(filename): return filename
[docs] def close(self): """Close WSGI server if needed""" if self.server: self.server.shutdown()
[docs]def casperjs(test_app, timeout=60): """A context manager to run a test with a :class:`webtest.ext.TestApp`""" app = TestApp( binary = app.get_binary('casperjs') tempdir = tempfile.mkdtemp(prefix='casperjs') def run(script, *args): dirname = os.path.dirname(sys._getframe(1).f_code.co_filename) log = os.path.join(tempdir, os.path.basename(script) + '.log') script = os.path.join(dirname, script) if binary: stdout = open(log, 'ab+') cmd = [binary, 'test'] + list(args) + [script] p = subprocess.Popen(cmd, stdout=stdout, stderr=subprocess.PIPE) end = time.time() + timeout while time.time() < end: ret = p.poll() if ret is not None: end = time.time() + 100 break time.sleep(.3) if time.time() < end: try: p.kill() except OSError: pass if os.path.isfile(log): with open(log) as fd: output = if isinstance(output, binary_type): output = output.decode('utf8', 'replace') if p.returncode != 0: print(output) raise AssertionError( 'Failure while running %s' % ' '.join(cmd)) try: yield run finally: shutil.rmtree(tempdir) app.close()