Testing

This section covers how to test some parts of Rocketry’s system including the conditions and the scheduler.

Conditions

The conditions have method observe that is used by the system to evaluate the state (whether the condition is true or false).

Here is a simple example:

from rocketry.conds import true, false

cond = true | false

assert cond.observe()

However, some conditions also require the task or the session to determine their state. Conditions that don’t require them still accepts them.

from rocketry.conds import daily

@app.task()
def do_things():
    ...

assert daily.observe(task=do_things, session=app.session)

Tasks

You can run a specific once task using the scheduler. This can be done with session.run:

from rocketry import Rocketry

app = Rocketry()

@app.task()
def do_things():
    ...

if __name__ == "__main__":
    app.session.run(do_things)

Scheduler

Sometimes you might want to test the full system in a unit test. However, unlike often with production, you don’t want your unit tests to run indefinitely. To prevent this, you can set a shutdown condition to the scheduler.

This scheduler runs 5 minutes before automatically ending itself:

from rocketry import Rocketry
from rocketry.conds import scheduler_running

app = Rocketry(shut_cond=scheduler_running(more_than="5 minutes"))

@app.task()
def do_things():
    ...

if __name__ == "__main__":
    app.run()

After this, you can test the task logs or other items.

Setting Custom Time

You can also test scheduling and task triggering by forcing Rocketry to use custom time.

For example, if you wish to fix scheduling to start at 12:00 on 12th of May 2014:

import datetime
import time

from rocketry import Rocketry
from rocketry.conds import daily

def fix_time(dt):
    "Get new time measurement function"
    start_time = time.time()
    def get_time():
        sec_since_start = time.time() - start_time
        return dt.timestamp() + sec_since_start
    return get_time

app = Rocketry(config={"time_func": fix_time(datetime.datetime(2014, 5, 31, 12, 00))})

@app.task(daily)
def do_things():
    ...

if __name__ == "__main__":
    app.run()

Note

We used nested functions to reuse fixing the time. If it looks confusing, you can also use a flat function with globals:

START_TIME = time.time()
DATETIME = datetime.datetime(2014, 5, 31, 12, 00)
def get_time():
    sec_since_start = time.time() - START_TIME
    return DATETIME.timestamp() + sec_since_start

app = Rocketry(time_func=get_time)