API & Fixtures

To write tests, the only function you need to use is the scenario() decorator.

molotov.scenario(weight=1, delay=0.0, name=None)

Decorator to register a function as a Molotov test.

Options:

  • weight used by Molotov when the scenarii are randomly picked. The functions with the highest values are more likely to be picked. Integer, defaults to 1. This value is ignored when the scenario_picker decorator is used.
  • delay once the scenario is done, the worker will sleep delay seconds. Float, defaults to 0. The general –delay argument you can pass to Molotov will be summed with this delay.
  • name name of the scenario. If not provided, will use the function __name___ attribute.

The decorated function receives an aiohttp.ClientSession instance.

If you don’t want scenarii to be picked randomly given the weights, you can provide your own scenario picker function, by decorating it with the scenario_picker() decorator.

molotov.scenario_picker()

Called to chose a scenario.

Arguments received by the decorated function:

  • worker_id the worker number
  • step_id the loop counter

The decorated function should return the name of the scenario the worker should execute next.

When used, the weights are ignored.

The decorated function should not be a coroutine.

Molotov also provides optional decorators to deal with test fixtures.

The lifecycle of a test is shown in the diagram below, and test fixtures can be used to run functions at various stages.

../_images/lifecycle.jpg
molotov.global_setup()

Called once when the test starts.

The decorated function is called before processes and workers are created.

Arguments received by the decorated function:

  • args arguments used to start Molotov.

This decorator is useful if you need to set up some fixtures that are shared by all workers.

The decorated function should not be a coroutine.

molotov.setup()

Called once per worker startup.

Arguments received by the decorated function:

  • worker_id the worker number
  • args arguments used to start Molotov.

The decorated function can send back a dict. This dict will be passed to the aiohttp.ClientSession class as keywords 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.

The decorated function should be a coroutine.

molotov.setup_session()

Called once per worker startup.

Arguments received by the decorated function:

  • worker_id the worker number
  • session the aiohttp.ClientSession instance created

The function can attach extra attributes to the session and use session.loop if needed.

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 decorated function should be a coroutine.

molotov.teardown_session()

Called once per worker when the session is closing.

Arguments received by the decorated function:

  • worker_id the worker number
  • session the aiohttp.ClientSession instance

The decorated function should be a coroutine.

molotov.teardown()

Called when a worker is done.

Arguments received by the decorated function:

  • worker_id the worker number

The decorated function should not be a coroutine.

molotov.global_teardown()

Called when everything is done.

The decorated function should not be a coroutine.

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

"""

This Molotov script has:

- a global setup fixture that sets variables
- an init worker fixture that sets the session headers
- an init session that attachs an object to the current session
- 1 scenario
- 2 tear downs fixtures

"""
import molotov


class SomeObject(object):
    """Does something smart in real life 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):
    molotov.get_context(session).attach("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):
    molotov.get_context(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.")