Hiro - time manipulation utilities for testing in python#
Often testing code that can be time dependent can become either fragile or
slow. Hiro provides context managers and utilities to either freeze, accelerate
or decelerate and jump between different points in time. Functions exposed by the
standard library’s time
, datetime
and date
modules are patched within
the contexts exposed.
Hiro context manager and utilities#
Timeline context#
The Timeline
context manager hijacks a few commonly used time functions
to allow time manipulation within its context. Specifically sleep()
, time()
, time.time_ns()
,
monotonic()
, time.monotonic_ns()
, time.localtime()
, gmtime()
,
datetime.datetime.now()
, datetime.datetime.utcnow()
and datetime.datetime.today()
behave according the configuration of the context.
The context provides the following manipulation options:
rewind()
: accepts seconds as an integer or antimedelta
instance.forward()
: accepts seconds as an integer or antimedelta
instance.freeze()
: accepts a floating point time since epoch or adatetime
ordate
instance to freeze the time at.unfreeze()
: resumes time from the point it was frozen at.scale()
: accepts a floating point to accelerate/decelerate time by.> 1 = acceleration, < 1 = deceleration
reset()
: resets all time alterations.
import hiro
from datetime import timedelta, datetime
import time
datetime.now().isoformat()
# OUT: '2013-12-01T06:55:41.706060'
with hiro.Timeline() as timeline:
# forward by an hour
timeline.forward(60*60)
datetime.now().isoformat()
# OUT: '2013-12-01T07:55:41.707383'
# jump forward by 10 minutes
timeline.forward(timedelta(minutes=10))
datetime.now().isoformat()
# OUT: '2013-12-01T08:05:41.707425'
# jump to yesterday and freeze
timeline.freeze(datetime.now() - timedelta(hours=24))
datetime.now().isoformat()
# OUT: '2013-11-30T09:15:41'
timeline.scale(5) # scale time by 5x
time.sleep(5) # this will effectively only sleep for 1 second
# since time is frozen the sleep has no effect
datetime.now().isoformat()
# OUT: '2013-11-30T09:15:41'
timeline.rewind(timedelta(days=365))
datetime.now().isoformat()
# OUT: '2012-11-30T09:15:41'
To reduce the amount of statements inside the context, certain timeline setup tasks can be done via the constructor and/or by using the fluent interface.
import hiro
import time
from datetime import timedelta, datetime
start_point = datetime(2012,12,12,0,0,0)
my_timeline = hiro.Timeline(scale=5).forward(60*60).freeze()
with my_timeline as timeline:
print datetime.now()
# OUT: '2012-12-12 01:00:00.000315'
time.sleep(5) # effectively 1 second
# no effect as time is frozen
datetime.now()
# OUT: '2012-12-12 01:00:00.000315'
timeline.unfreeze()
# back to starting point
datetime.now()
# OUT: '2012-12-12 01:00:00.000317'
time.sleep(5) # effectively 1 second
# takes effect (+5 seconds)
datetime.now()
# OUT: '2012-12-12 01:00:05.003100'
Timeline
can additionally be used as a decorator. If the decorated
function expects a timeline
argument, the Timeline
will be
passed to it.
import hiro
import time, datetime
@hiro.Timeline(scale=50000)
def sleeper():
datetime.datetime.now()
# OUT: '2013-11-30 14:27:43.409291'
time.sleep(60*60) # effectively 72 ms
datetime.datetime.now()
# OUT: '2013-11-30 15:28:36.240675'
@hiro.Timeline()
def sleeper_aware(timeline):
datetime.datetime.now()
# OUT: '2013-11-30 14:27:43.409291'
timeline.forward(60*60)
datetime.datetime.now()
# OUT: '2013-11-30 15:28:36.240675'
run_sync and run_async#
In order to execute certain callables within a Timeline
context, two
shortcut functions are provided.
Both functions return a ScaledRunner
object which provides the following methods
get_execution_time()
: The actual execution time of thecallable
get_response()
(will either return the actual return value ofcallable
or raise the exception that was thrown)
run_async()
returns a derived class of hiro.core.ScaledRunner
that additionally provides the following methods
is_running()
:True/False
depending on whether the callable has completed executionjoin()
: blocks until thecallable
completes execution
import hiro
import time
def _slow_function(n):
time.sleep(n)
if n > 10:
raise RuntimeError()
return n
runner = hiro.run_sync(10, _slow_function, 10)
runner.get_response()
# OUT: 10
# due to the scale factor 10 it only took 1s to execute
runner.get_execution_time()
# OUT: 1.1052658557891846
runner = hiro.run_async(10, _slow_function, 11)
runner.is_running()
# OUT: True
runner.join()
runner.get_execution_time()
# OUT: 1.1052658557891846
runner.get_response()
# OUT: Traceback (most recent call last):
# ....
# OUT: File "<input>", line 4, in _slow_function
# OUT: RuntimeError