testr/runner.py: add TestRunner.run_test_suite

This commit is contained in:
dogeystamp 2023-11-02 14:15:23 -04:00
parent daa94bfa7e
commit ce369b67e3
Signed by: dogeystamp
GPG Key ID: 7225FE3592EFFA38
2 changed files with 19 additions and 14 deletions

View File

@ -38,24 +38,24 @@ class ExecutableRunner(TestRunner):
stdin=asyncio.subprocess.PIPE,
)
input_data = (await data.get_input()).encode()
input_data = await data.get_input()
try:
out_stream, err_stream = await asyncio.wait_for(
proc.communicate(input=input_data), timeout=5.0)
proc.communicate(input=input_data.encode()), timeout=5.0)
except TimeoutError:
proc.kill()
return TestStatus(code=StatusCode.TLE, stderr="", stdout="")
return TestStatus(code=StatusCode.TLE, stderr="", stdout="", stdin=input_data)
stdout: str = out_stream.decode()
stderr: str = err_stream.decode()
if proc.returncode != 0:
return TestStatus(code=StatusCode.IR, stdout=stdout, stderr=stderr)
return TestStatus(code=StatusCode.IR, stdout=stdout, stderr=stderr, stdin=input_data)
correct: bool = await data.validate_output(stdout)
ret_code = StatusCode.AC if correct else StatusCode.WA
return TestStatus(code=ret_code, stdout=stdout, stderr=stderr)
return TestStatus(code=ret_code, stdout=stdout, stderr=stderr, stdin=input_data)
class DirectorySuite(TestSuite):

View File

@ -1,7 +1,7 @@
from dataclasses import dataclass
from abc import ABC, abstractmethod
from collections.abc import Iterable
from enum import Enum, auto
from typing import AsyncIterator, Iterator
class StatusCode(Enum):
@ -34,6 +34,7 @@ class TestStatus:
"""
code: StatusCode
stdin: str
stderr: str
stdout: str
@ -58,15 +59,19 @@ class TestData(TestInput, TestValidator):
pass
class TestRunner(ABC):
"""Runner for test cases."""
@abstractmethod
async def run_test(self, data: TestData) -> StatusCode: pass
class TestSuite(ABC):
"""Loader for multiple test cases."""
@abstractmethod
def __iter__(self) -> Iterable[TestData]: pass
def __iter__(self) -> Iterator[TestData]: pass
class TestRunner(ABC):
"""Runner for test cases."""
@abstractmethod
async def run_test(self, data: TestData) -> TestStatus: pass
async def run_test_suite(self, data: TestSuite) -> AsyncIterator[TestStatus]:
for test_case in data:
yield await self.run_test(test_case)