Skip to content
Back to top
View this page

# 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 an `timedelta` instance.

- `forward()`: accepts seconds as an integer or an `timedelta` instance.

- `freeze()`: accepts a floating point time since epoch or a `datetime` or `date` 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.

- `hiro.run_sync()`

- `hiro.run_async()`

Both functions return a `ScaledRunner` object which provides the following methods

- `get_execution_time()`: The actual execution time of the `callable`

- `get_response()` (will either return the actual return value of `callable` 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 execution

- `join()`: blocks until the `callable` 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