Fixtures¶

Molotov provides some decorators to deal with test fixtures:

  • @global_setup() called once when the test starts, before processes and workers are created. Receives the arguments used to start Molotov. The decorated function should not be a coroutine.
  • @setup() called once per worker startup. Receives the worker number and the arguments used to start Molotov. The decorated function should be a coroutine.
  • @setup_session() called once per worker startup when the aoihttp.ClienSession instance has been created. The decorated function should be a coroutine and receives the worker id and the session.
  • @teardown_session() called once per worker startup when the aoihttp.ClienSession instance is going to be closed. The decorated function should be a coroutine and receives the worker id and the session.
  • @teardown() called when a worker is done. Receives the worker id. The decorated function should not be a coroutine.
  • @global_teardown() called when everything is done. The decorated function should not be a coroutine.

When a function is decorated with the setup() decorator, it will be called with the worker id and the command-line arguments and needs to send back a dict.

This dict will be passed to the ClientSession class when it’s created. This is useful when you need to set up session-wide options like Authorization headers, or do whatever you need on startup.

Once the session is created, it’s passed to setup_session(), where you can attach extra attributes to the session and use session.loop if you need to. It’s a good place to attache an object that interacts with the event loop, so you are sure to use the same one that the session’s.

The global_setup() decorator is useful if you need to set up some fixtures that are shared by all workers.

Here’s a full example, in order of calls:

import molotov


class SomeObject(object):
    """Does something smart in real lif with the async loop.
    """
    def __init__(self, loop):
        self.loop = loop

    def cleanup(self):
        pass


@molotov.global_setup()
def init_test(args):
    molotov.set_var('SomeHeader', '1')
    molotov.set_var('endpoint', 'http://localhost:8080')


@molotov.setup()
async def init_worker(worker_num, args):
    headers = {'AnotherHeader': '1',
               'SomeHeader': molotov.get_var('SomeHeader')}
    return {'headers': headers}


@molotov.setup_session()
async def init_session(worker_num, session):
    session.ob = SomeObject(loop=session.loop)


@molotov.scenario(100)
async def scenario_one(session):
    endpoint = molotov.get_var('endpoint')
    async with session.get(endpoint) as resp:
        res = await resp.json()
        assert res['result'] == 'OK'
        assert resp.status == 200


@molotov.teardown_session()
async def end_session(worker_num, session):
    session.ob.cleanup()


@molotov.teardown()
def end_worker(worker_num):
    print("This is the end for %d" % worker_num)


@molotov.global_teardown()
def end_test():
    print("This is the end of the test.")