blob: 348798aa08d452eb170388e24e2945dc10632cbc [file] [log] [blame]
import contextlib
import time
class DeadlineExceeded(Exception):
pass
class TimedContext(contextlib.ContextDecorator):
"""
A context that will raise DeadlineExceeded if the
max duration is reached during the execution.
>>> TimedContext(1)(time.sleep)(.1)
>>> TimedContext(0)(time.sleep)(.1)
Traceback (most recent call last):
...
tests._context.DeadlineExceeded: (..., 0)
"""
def __init__(self, max_duration: int):
self.max_duration = max_duration
def __enter__(self):
self.start = time.monotonic()
def __exit__(self, *err):
duration = time.monotonic() - self.start
if duration > self.max_duration:
raise DeadlineExceeded(duration, self.max_duration)