58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
import multiprocessing
|
|
from typing import Any, Callable, Iterator
|
|
|
|
|
|
class TimeoutException(Exception):
|
|
pass
|
|
|
|
def get_tests_for(cls: type) -> Iterator[Callable[[], None]]:
|
|
"""get all functions that are declared as tests"""
|
|
return [e for e in cls.__dict__.values() if hasattr(e, 'is_test')]
|
|
|
|
def test_wrapper(test: Callable[[], None]):
|
|
try:
|
|
test()
|
|
except Exception as e:
|
|
test.has_failed = True
|
|
test.cause = e
|
|
|
|
|
|
def run_test(test: Callable[[], None]) -> None:
|
|
"""run a test and catch any unexpected behavior and mark it as failed with the exception as cause"""
|
|
if not hasattr(test, 'is_test'):
|
|
return
|
|
timeout = test.timeout
|
|
p = multiprocessing.Process(target=test_wrapper, args=[test])
|
|
p.start()
|
|
p.join(timeout)
|
|
if p.is_alive():
|
|
test.has_failed = True
|
|
test.cause = TimeoutException(f"test failed after {timeout} seconds")
|
|
|
|
|
|
def run_tests_for_task(task: object) -> None:
|
|
for test in task.tests:
|
|
run_test(test)
|
|
|
|
|
|
def eip_task(name: str, max_points: float) -> Callable[[type], type]:
|
|
def wrapper(cls: type) -> type:
|
|
cls.is_task: bool = True
|
|
cls.name: str = name
|
|
cls.max_points: float = max_points
|
|
cls.tests: Callable[..., None] = get_tests_for(cls)
|
|
return cls
|
|
return wrapper
|
|
|
|
|
|
def eip_test(msg: str, points: float, timeout = 180) -> Callable[[], None]:
|
|
def wrapper(test: Callable[[], None]) -> Callable[[], None]:
|
|
test.is_test: bool = True
|
|
test.msg: str = msg
|
|
test.has_failed: bool = False
|
|
test.points: float = points
|
|
test.timeout: int = timeout
|
|
test.cause: Any | None = None
|
|
return test
|
|
return wrapper
|