dogeystamp/testr
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

format with black

Browse Source
This commit is contained in:
dogeystamp 2023-11-04 17:35:18 -04:00
parent a8adcab0a1
commit f247fb6019
Signed by: dogeystamp
GPG Key ID: 7225FE3592EFFA38
3 changed files with 51 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
testr/cli.py
View File

@ -18,13 +18,16 @@ async def run_cli():
test_data_group = parser.add_mutually_exclusive_group(required=True)
test_data_group.add_argument("--testdir", help="Directory for test cases")
test_runner_group.add_argument("--exec", help="Executable to run test cases against")
test_runner_group.add_argument(
"--exec", help="Executable to run test cases against"
)
parser.add_argument("--timelim", help="Time limit in seconds", type=float, default=5.0)
parser.add_argument(
"--timelim", help="Time limit in seconds", type=float, default=5.0
)
args = parser.parse_args()
code_styles = {
StatusCode.AC: Fore.GREEN,
StatusCode.WA: Fore.RED,
@ -35,29 +38,21 @@ async def run_cli():
test_suite = DirectorySuite(Path(args.testdir))
test_runner = ExecutableRunner(Path(args.exec))
status_counts = { k: 0 for k in StatusCode }
status_counts = {k: 0 for k in StatusCode}
test_case_count = 0
async for test_case in test_runner.run_test_suite(
test_suite,
TestOptions(
time_limit=args.timelim
)
):
test_suite, TestOptions(time_limit=args.timelim)
):
test_case_count += 1
status_counts[test_case.code] += 1
print(
f"{test_case.test_data.name : <15} "
f"[ {code_styles.get(test_case.code, '')}{test_case.code.name}{Style.RESET_ALL} ]"
)
f"{test_case.test_data.name : <15} "
f"[ {code_styles.get(test_case.code, '')}{test_case.code.name}{Style.RESET_ALL} ]"
)
if test_case.code == StatusCode.IR:
print(
"---\n" +
Style.BRIGHT +
"Program stderr:"
+ Style.RESET_ALL
)
print("---\n" + Style.BRIGHT + "Program stderr:" + Style.RESET_ALL)
print("\n".join([f" {line}" for line in test_case.stderr.split("\n")]))
if test_case_count > 15:
@ -66,9 +61,9 @@ async def run_cli():
for code in StatusCode:
if status_counts[code] > 0:
print(
f"{code_styles.get(code, '')}{code.name : >7}{Style.RESET_ALL}: "
f"x{status_counts[code]}"
)
f"{code_styles.get(code, '')}{code.name : >7}{Style.RESET_ALL}: "
f"x{status_counts[code]}"
)
print(f"\n Finished in{time.perf_counter() - start_time : .2f} seconds.\n")
@ -76,5 +71,6 @@ async def run_cli():
def main():
asyncio.run(run_cli())
if __name__ == "__main__":
main()

33
testr/file_data.py
View File

@ -1,5 +1,12 @@
import asyncio
from testr.runner import TestData, TestOptions, TestRunner, TestStatus, TestSuite, StatusCode
from testr.runner import (
TestData,
TestOptions,
TestRunner,
TestStatus,
TestSuite,
StatusCode,
)
from pathlib import Path
@ -33,26 +40,29 @@ class ExecutableRunner(TestRunner):
async def run_test(self, data: TestData, opts: TestOptions) -> TestStatus:
proc = await asyncio.create_subprocess_shell(
str(self.executable),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
stdin=asyncio.subprocess.PIPE,
)
str(self.executable),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
stdin=asyncio.subprocess.PIPE,
)
input_data = await data.get_input()
try:
out_stream, err_stream = await asyncio.wait_for(
proc.communicate(input=input_data.encode()), timeout=opts.time_limit)
proc.communicate(input=input_data.encode()), timeout=opts.time_limit
)
except TimeoutError:
proc.kill()
return TestStatus(code=StatusCode.TLE, stderr="", stdout="", test_data = data)
return TestStatus(code=StatusCode.TLE, stderr="", stdout="", test_data=data)
stdout: str = out_stream.decode()
stderr: str = err_stream.decode()
if proc.returncode != 0:
return TestStatus(code=StatusCode.IR, stdout=stdout, stderr=stderr, test_data=data)
return TestStatus(
code=StatusCode.IR, stdout=stdout, stderr=stderr, test_data=data
)
correct: bool = await data.validate_output(stdout)
ret_code = StatusCode.AC if correct else StatusCode.WA
@ -73,8 +83,9 @@ class DirectorySuite(TestSuite):
outp_file = inp_file.with_suffix(".out")
if not outp_file.is_file():
raise ValueError(f"output file '{outp_file}' is not a valid file")
self.test_cases.append(FileData(inp_file, outp_file, name=inp_file.with_suffix("").name))
self.test_cases.append(
FileData(inp_file, outp_file, name=inp_file.with_suffix("").name)
)
def __iter__(self):
return self.test_cases.__iter__()

18
testr/runner.py
View File

@ -31,14 +31,16 @@ class TestInput(ABC):
"""Input provider for single test case."""
@abstractmethod
async def get_input(self) -> str: pass
async def get_input(self) -> str:
pass
class TestValidator(ABC):
"""Output validator for single test case."""
@abstractmethod
async def validate_output(self, output: str) -> bool: pass
async def validate_output(self, output: str) -> bool:
pass
@dataclass
@ -71,7 +73,8 @@ class TestSuite(ABC):
"""Loader for multiple test cases."""
@abstractmethod
def __iter__(self) -> Iterator[TestData]: pass
def __iter__(self) -> Iterator[TestData]:
pass
@dataclass
@ -79,7 +82,7 @@ class TestStatus:
"""
Status of an individual test case.
"""
code: StatusCode
test_data: TestData
@ -92,8 +95,11 @@ class TestRunner(ABC):
"""Runner for test cases."""
@abstractmethod
async def run_test(self, data: TestData, opts: TestOptions) -> TestStatus: pass
async def run_test(self, data: TestData, opts: TestOptions) -> TestStatus:
pass
async def run_test_suite(self, data: TestSuite, opts: TestOptions) -> AsyncIterator[TestStatus]:
async def run_test_suite(
self, data: TestSuite, opts: TestOptions
) -> AsyncIterator[TestStatus]:
for test_case in data:
yield await self.run_test(test_case, opts)