Test Suite

The TestSuite is a testing framework for both synchronous and asynchronous applications and for running tests in parallel on multiple threads or processes. It is used for testing pulsar but it can be used as a test suite for any other library.

Integration

Pulsar test suite can be used in conjunction with setuptools or stand-alone.

Setuptools Integration

Pulsar asynchronous test suite can be used as testing framework in your setuptools based project. Add this to setup.py file:

from setuptools import setup

setup(
    #...,
    setup_requires=['pulsar', ...],
    #...,
)

And create an alias into setup.cfg file:

[aliases]
test=pulsar_test

If you now type:

python setup.py test

this will execute your tests using pulsar test runner. As this is a standalone version of pulsar no prior installation whatsoever is required for calling the test command. You can also pass additional arguments to pulsar test, such as your test directory or other options using -a.

Manual Integration

If for some reason you don’t want/can’t use the setuptools integration, you can write a standalone script, for example runtests.py:

from pulsar.apps import TestSuite

if __name__ == '__main__':
    TestSuite(description='Test suite for my library').start()

To run the test suite:

python runtests.py

Type:

python runtests.py --help

For a list different options/parameters which can be used when running tests.

Running Tests

Writing a Test Case

Only subclasses of TestCase are collected by this application. An example test case:

import unittest

class MyTest(unittest.TestCase):

    async def test_async_test(self):
        result = await async_function()
        self.assertEqual(result, ...)

    def test_simple_test(self):
        self.assertEqual(1, 1)

Note

Test functions are asynchronous, when they are coroutine functions or return a Future, synchronous, when they return anything else.

Loading Tests

The loading of test cases is controlled by the modules parameter when initialising the TestSuite:

from pulsar.apps import TestSuite

if __name__ == '__main__':
    TestSuite(modules=('tests', 'examples')).start()

When using pulsar test suite - setuptools integration, test modules are specified in the setup.cfg:

[test]
test_modules = tests examples

Test files

The TestSuite loads tests via the loader property. By default it recursively loads test files inside the modules directories matching:

  • test_<label>.py
  • <label>_test.py
  • tests.py

Test labels

To run a specific label:

python runtests.py <label>

When using the setuptools integration:

python setup.py test -a <label>

To list all labels:

python setup.py test -l

It is also possible to run a single test function within a label:

python setup.py test -a <label>.<test_function_name>

Options

All standard settings can be applied to the test application. In addition, the following options are test suite specific:

sequential

By default, test functions within a TestCase are run in asynchronous fashion. This means that several test functions may be executed at once depending on their return values.

By specifying the –sequential command line option, the TestSuite forces test functions from a given TestCase to be run in a sequential way, one after the other:

python runtests.py --sequential

Alternatively, if you need to specify a TestCase which always runs its test functions in a sequential way, you can use the sequential() decorator:

from pulsar.apps.test import sequential

@sequential
class MyTestCase(unittest.TestCase):
    ...

list labels

By passing the -l or –list-labels flag to the command line, the full list of test labels available is displayed:

python runtests.py -l

test timeout

When running asynchronous tests, it can be useful to set a cap on how long a test function can wait for results. This is what the –test-timeout command line flag does:

python runtests.py --test-timeout 10

Set the test timeout to 10 seconds. Test timeout is only meaningful for asynchronous test function.

Http TestClient

The test_wsgi_request() can be used to test wsgi middleware without going through socket connections.

To use the client in a test function:

async def test_my_wsgi_test(self):
   request = await test_wsgi_request('/')
   response = await http.get('http://bla.com/...')

The host part of the url is irrelevant, it can be anything you like.

API

Test Suite

class pulsar.apps.test.TestSuite(callable=None, load_config=True, **params)[source]

An asynchronous test suite which works like a task queue.

Each task is a group of test methods in a python TestCase class.

Parameters:
  • modules – An iterable over modules where to look for tests. If not provided it is set as default to ["tests"] which loads all python module from the tests module in a recursive fashion. Check the the TestLoader for detailed information.
  • plugins – Optional list of dotted path to TestPlugin classes.
loader

Instance of the TestLoader used for loading test cases

monitor_start(monitor, **kw)[source]

When the monitor starts load all test classes into the queue

Test Loader

class pulsar.apps.test.TestLoader(suite)[source]

Classes used by the TestSuite to aggregate tests from a list of paths.

The way it works is simple, you give a root directory and a list of submodules where to look for tests.

Parameters:
tags(include=None, exclude=None)[source]

Return a generator of tag information

test_files(include=None, exclude=None)[source]

List of tag, modules pairs.

Parameters:
  • tags – optional list of tags to include, if not available all tags will be included.
  • exclude_tags – optional list of tags to exclude. If not provided no tags will be excluded.

Plugin

class pulsar.apps.test.Plugin[source]

Interface for all classes which are part of the TestRunner.

Most classes used by the test application are plugins, for example the TestRunner itself, the TestResult and the TestPlugin.

result = None

An optional result

stream = None

handle for writing text on the default output.

Set by the TestRunner at runtime.

configure(cfg)[source]
Called once just after construction of a TestRunner
and before any test class is loaded.

This is a chance to configure the Plugin or global variables which may affect the way tests are run. If it returns something other than None (for example an abort message) it will stop the configuration of all subsequent plugins and quit the test.

Parameters:cfg – a Config.
Returns:None unless the tests runner must be stopped.
on_start()[source]

Called by the TestSuite once only at startup.

This callback is invoked once all tests are loaded but before the test suite starts running them.

on_end()[source]

Called by the TestSuite just before it stops.

loadTestsFromTestCase(testcls)[source]

Called when loading tests from the testcls class.

Can be used to modify the number of test functions loaded.

startTestClass(testcls)[source]

Called just before a testcls runs its tests.

stopTestClass(testcls)[source]

Called just after a testcls has run its tests.

startTest(test)[source]

Called just before a test function is executed.

This is run just before _pre_setup method.

stopTest(test)[source]

Called just after a test function has finished.

This is run just after the _post_teardown method.

before_test_function_run(test, local)[source]

Can be used by plugins to manipulate the test behaviour in the process domain where the test run.

after_test_function_run(test, local)[source]

Executed in the test process domain, after the test has finished.

addSuccess(test)[source]

Called when a test function succeed

addFailure(test, err)[source]

Called when a test function as a (test) failure

addError(test, err)[source]

Called when a test function as an (unexpected) error

Test Runner

class pulsar.apps.test.TestRunner(plugins, stream, writercls=None, descriptions=True)[source]

A Plugin for asynchronously running tests.

loadTestsFromTestCase(test_cls)[source]

Load all test functions for the test_cls

before_test_function_run(test)[source]

Called just before the test is run

after_test_function_run(test)[source]

Called just after the test has finished, in the test process domain.

Test Result

class pulsar.apps.test.TestResult(descriptions=True)[source]

A Plugin for collecting results/failures for test runs.

Each Plugin can access the TestRunner result object via the result attribute.

startTest(test)[source]

Increase the test counter

addError(test, err)[source]

Called when an unexpected error has occurred.

err is a tuple of values as returned by sys.exc_info()

addFailure(test, err)[source]

Called when an test failure has occurred.

err is a tuple of values as returned by sys.exc_info()

addSkip(test, reason)[source]

Called when a test is skipped.

addExpectedFailure(test, err)[source]

Called when an expected failure/error occurred.

addUnexpectedSuccess(test)[source]

Called when a test was expected to fail, but succeed.

wasSuccessful()[source]

Tells whether or not this result was a success

Test Plugin

class pulsar.apps.test.TestPlugin[source]

Base class for Plugin which can be added to a TestSuite to extend its functionalities.

If the class attribute name is not specified or its value validate as True, an additional setting is added to the configuration. In addition, a TestPlugin can specify several additional settings as class attributes. For example, the benchmark plugin has an additional setting for controlling the number of repetitions:

class Bench(TestPlugin):
    repeat = Setting(type=int,
                        default=1,
                        validator=validate_pos_int,
                        desc="Default number of repetition")
name

Class attribute used for adding the default plugin setting to the configuration container of the test suite application. If the attribute is not set, the class name in lower case is used. If set and validate as not True, no new setting is added to the test suite configuration parameters. For example:

class MyPlugin(TestPlugin):
    name = None

won’t add the default plugin setting.

desc

Class attribute used as the description of the setting. If name is disabled, this attribute is not relevant.

config

A Config container created by the TestPlugin metaclass. It collects the default setting, if available, and any additional settings specified as class attributes.

Populate

A useful function for populating random data:

from pulsar.apps.test import populate

data = populate('string', 100)

gives you a list of 100 random strings

pulsar.apps.test.populate(datatype='string', size=10, start=None, end=None, converter=None, choice_from=None, **kwargs)[source]

Utility function for populating lists with random data.

Useful when populating database with data for fuzzy testing.

Supported data-types

  • string

    For example:

    populate('string', 100, min_len=3, max_len=10)
    

    create a 100 elements list with random strings with random length between 3 and 10

  • date

    For example:

    from datetime import date
    populate('date', 200, start=date(1997,1,1), end=date.today())
    

    create a 200 elements list with random datetime.date objects between start and end

  • integer

    For example:

    populate('integer', 200, start=0, end=1000)
    

    create a 200 elements list with random integers between start and end

  • float

    For example:

    populate('float', 200, start=0, end=10)
    

    create a 200 elements list with random floats between start and end

  • choice (elements of an iterable)

    For example:

    populate('choice', 200, choice_from=['pippo','pluto','blob'])
    

    create a 200 elements list with random elements from choice_from

Test Plugins

A TestPlugin is a way to extend the test suite with additional options and behaviours implemented in the various plugin’s callbacks. There are two basic rules for plugins:

  • Test plugin classes should subclass TestPlugin.
  • Test plugins may implement any of the methods described in the class Plugin interface.

Pulsar ships with two battery-included plugins:

Benchmark

BenchMark is a TestPlugin for benchmarking test functions.

To use the plugin follow these three steps:

  • Flag a unittest.TestCase class with the __benchmark__ = True class attribute:

    class MyBenchmark(unittest.TestCase):
        __benchmark__ = True
    
        def test_mybenchmark_function1(self):
            ...
    
        def test_mybenchmark_function2(self):
            ...
    
  • Run the test suite with the --benchmark command line option.

Alternatively, you can use the bench command from your setup.py file. Simply add the bench entry in the setup.cfg.

[bench] test_modules = tests/bench

The test class can implement additional methods to fine-tune how the benchmark plugin evaluate the performance and display results:

  • When implemented, the startUp method is invoked before each run of a test function.
  • The time taken to run a test once can be modified by implementing the getTime method which receives as only argument the time interval taken. By default it returns the same time interval.
class pulsar.apps.test.plugins.bench.BenchMark[source]

Benchmarking addon for pulsar test suite.

Profile

Profile is a TestPlugin for profiling test cases and generating Html reports. It uses the cProfile module from the standard library.

To use the plugin follow these two steps:

  • Included it in the test Suite:

    from pulsar.apps.test import TestSuite
    from pulsar.apps.test.plugins import profile
    
    def suite():
        TestSuite(..., plugins=(..., profile.Profile()))
    
  • Run the test suite with the --profile command line option.

class pulsar.apps.test.plugins.profile.Profile[source]

TestPlugin for profiling test cases.

Utilities

sequential

pulsar.apps.test.utils.sequential(cls)[source]

Decorator for a TestCase which cause its test functions to run sequentially rather than in an asynchronous fashion.

Typical usage:

import unittest

from pulsar.apps.test import sequential

@sequenatial
class MyTests(unittest.TestCase):
    ...

You can also run test functions sequentially when using the sequential flag in the command line.

ActorTestMixin

class pulsar.apps.test.utils.ActorTestMixin[source]

A mixin for TestCase.

Useful for classes testing spawning of actors. Make sure this is the first class you derive from, before the TestCase, so that the tearDown method is overwritten.

concurrency

The concurrency model used to spawn actors via the spawn() method.

spawn_actor(concurrency=None, **kwargs)[source]

Spawn a new actor and perform some tests

check server

pulsar.apps.test.utils.check_server(name)[source]