Documentation for pulsar 0.9.2. For development docs, go here.

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.

Introduction

To get started with pulsar asynchronous test suite is easy. The first thing to do is to create a python script on the top level directory of your library, let’s call it runtests.py:

from pulsar.apps import TestSuite

if __name__ == '__main__':
    TestSuite(description='Test suite for my library',
              modules=('regression',
                       ('examples', 'tests'))).start()

where modules is a tuple/list which specifies where to search for test cases to run. In the above example the test suite will look for all python files in the regression module (in a recursive fashion), and for modules called tests in the examples module.

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.

Note

Pulsar test suite help you speed up your tests! Use setUpClass rather than setUp, use asynchronous tests and if not possible add fire power with more test workers.

Try with the -w 8 command line input for a laugh.

The next step is to actually write the tests.

Writing a Test Case

Only subclasses of TestCase are collected by this application. When running a test, pulsar looks for two extra method: _pre_setup and _post_teardown. If the former is available, it is run just before the setUp() method while if the latter is available, it is run just after the tearDown() method. In addition, if the setUpClass() class methods is available, it is run just before all tests functions are run and the tearDownClass(), if available, is run just after all tests functions are run.

An example test case:

import unittest

class MyTest(unittest.TestCase):

    def test_async_test(self):
        result = yield maybe_async_function()
        self.assertEqual(result, ...)

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

Note

Test functions are asynchronous, when they return a generator or 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','tests'))).start()

The TestSuite loads tests via the TestLoader class.

In the context of explaining the rules for loading tests, we refer to an object as

  • a module (including a directory module)
  • a python class.

with this in mind:

  • Directories that aren’t packages are not inspected. This means that if a directory does not have a __init__.py file, it won’t be inspected.
  • Any class that is a TestCase subclass is collected.
  • If an object starts with _ or . it won’t be collected, nor will any objects it contains.
  • If an object defines a __test__ attribute that does not evaluate to True, that object will not be collected, nor will any objects it contains.

The modules parameter is a tuple or list of entries where each single entry follows the rules:

  • If the entry is a string, it indicates the dotted path relative to the root directory (the directory of the script running the tests). For example:

    modules = ['tests', 'moretests.here']
    
  • If an entry is two elements tuple, than the first element represents a dotted path relative to the root directory and the second element a pattern which modules (files or directories) must match in order to be included in the search. For example:

    modules = [...,
               ('examples', 'test*')
               ]
    

    load modules, from inside the examples module, starting with test.

  • If an entry is a three elements tuple, it is the same as the two elements tuple rule with the third element which specifies the top level tag for all tests in this entry. For example:

    modules = [...,
                ('bla', '*', 'foo')
              ]
    

For example, the following:

modules = ['test', ('bla', '*', 'foo'), ('examples','test*')]

loads

  • all tests from modules in the test directory,
  • all tests from the bla directory with top level tag foo,
  • all tests from modules which starts with test from the examples directory.

All top level modules will be added to the python path.

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):
    ...

Using the sequential option, does not mean only one test function is executed by the TestSuite at a given time. Indeed, several TestCase are executed at the same time and therefore each one of the may have one test function running.

In order to run only one function at any time, the sequential option should be used in conjunction with –concurrent-tasks option:

python runtests.py --sequential --concurrent-tasks 1

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 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:

  • Included it in the test Suite:

    from pulsar.apps.test import TestSuite
    from pulsar.apps.test.plugins import bench
    
    def suite():
        TestSuite(..., plugins=(..., bench.BenchMark()))
    
  • 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.

The test class can implement additional methods to fine-tune how the benchmark plugin evaluate the perfomance 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]

pulsar.apps.test.TestPlugin for profiling test cases.

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. Alternatively it can be a callable returning the iterable over modules. For example:

    suite = TestSuite(modules=('regression',
                               ('examples','tests'),
                               ('apps','test_*')))
    
    def get_modules(suite):
        ...
    
    suite = TestSuite(modules=get_modules)
    

    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.

  • result_class – Optional class for collecting test results. By default it used the standard TestResult.
  • plugins – Optional list of TestPlugin instances.
new_runner()[source]

The TestRunner driving test cases.

monitor_start(*args, **kwargs)[source]

When the monitor starts load all test classes into the queue

Test Loader

class pulsar.apps.test.loader.TestLoader(root, modules, runner, logger=None)[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:
checktag(tag, import_tags, exclude_tags)[source]

Return True if tag is in import_tags.

testmodules(tags=None, exclude_tags=None)[source]

Generator 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.
get_tests(path, dotted_path, pattern, import_tags=None, tags=(), exclude_tags=None, parent=None)[source]

Collect python modules for testing and return a generator of tag,module pairs.

Parameters:
  • path – directory path where to search. Files starting with _ or . are excluded from the search, as well as non-python files.
  • dotted_path – the dotted python path equivalent of path.
  • parent – the parent module for the current one. This parameter is passed by this function recursively.

Plugin

class pulsar.apps.test.result.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.result.TestRunner(plugins, stream, writercls=None, descriptions=True, logger=None)[source]

A Plugin for asynchronously running tests.

loadTestsFromTestCase(test_cls)[source]

Load all test functions for the test_cls

add(result)[source]

Add a new result to the result container

before_test_function_run(test)[source]

Called just before the test is run, in the test process domain.

after_test_function_run(test)[source]

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

Test Result

class pulsar.apps.test.result.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 occured.

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.plugins.base.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 = pulsar.Setting(type=int,
                        default=1,
                        validator=pulsar.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.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

Utilities

run on arbiter

pulsar.apps.test.utils.run_on_arbiter(f)[source]

Decorator for running a test function in the Arbiter context domain.

This can be useful to test Arbiter mechanics.

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(*args, **kwargs)[source]

Spawn a new actor and perform some tests

AsyncAssert

class pulsar.apps.test.utils.AsyncAssert(test)[source]

A descriptor added by the test-suite to all python TestCase loaded.

It can be used to invoke the same assertXXX methods available in the TestCase in an asynchronous fashion.

The descriptor is available via the async attribute. For example:

class MyTest(unittest.TestCase):

    def test1(self):
        yield self.async.assertEqual(3, Future().callback(3))
        ...

check server

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

Check if server name is available at the address specified <name>_server config value.

Return type:boolean


Table Of Contents

Previous topic

Distributed Task Queue

Next topic

Asynchronous Shell

This Page