Testing¶
This section covers how to test some parts of Tocketry’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 tocketry.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 tocketry.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 tocketry import Tocketry
app = Tocketry()
@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 tocketry import Tocketry
from tocketry.conds import scheduler_running
app = Tocketry(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 Tocketry 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 tocketry import Tocketry
from tocketry.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 = Tocketry(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 = Tocketry(time_func=get_time)