Commit bd47f22a authored by Dom Sekotill's avatar Dom Sekotill
Browse files

Add control for test timeouts

parent c8c549ea

tests/__init__.py

+6 −6
Original line number Diff line number Diff line
@@ -3,12 +3,10 @@ A package of tests for kilter.service modules 
"""



import functools

import inspect

import os

from collections.abc import Callable

from collections.abc import Coroutine

from inspect import iscoroutinefunction

from types import CoroutineType

from types import FunctionType

from typing import Any

from unittest import TestCase
@@ -17,6 +15,8 @@ import trio 
SyncTest = Callable[[TestCase], None]

AsyncTest = Callable[[TestCase], Coroutine[Any, Any, None]]



LIMIT_SCALE_FACTOR = float(os.environ.get("LIMIT_SCALE_FACTOR", 1))





class AsyncTestCase(TestCase):

	"""
@@ -24,18 +24,18 @@ class AsyncTestCase(TestCase): 
	"""



	@classmethod

	def __init_subclass__(cls, time_limit: float = 1.0, **kwargs: Any) -> None:

	def __init_subclass__(cls, time_limit: float = 0.5, **kwargs: Any) -> None:

		super().__init_subclass__(**kwargs)

		for name, value in ((n, getattr(cls, n)) for n in dir(cls)):

			if name.startswith("test_") and iscoroutinefunction(value):

				setattr(cls, name, _syncwrap(value, time_limit))

				setattr(cls, name, _syncwrap(value, time_limit * LIMIT_SCALE_FACTOR))





def _syncwrap(test: AsyncTest, time_limit: float) -> SyncTest:

	@functools.wraps(test)

	def wrap(self: TestCase) -> None:

		async def limiter() -> None:

			with trio.move_on_after(time_limit) as cancel_scope:

			with trio.fail_after(time_limit) as cancel_scope:

				await test(self)

			if cancel_scope.cancelled_caught:

				raise TimeoutError