check.py

#!/usr/bin/python3

from os import path, listdir
import sys
from datetime import datetime
from subprocess import run, PIPE, TimeoutExpired, CalledProcessError
if path.exists("./custom_evaluator.py"):
    sys.path.append(".")
    from custom_evaluator import evaluator
else:
    from evaluator import evaluator

from common import *


def check_test(test_name, solution_name, timeout=None):
    input_file_name = input_path(test_name)
    output_file_name = output_path(test_name)

    solution_path = path.join(SOLS_DIR, solution_name)

    with open(input_file_name) as input_file:
        try:
            start_time = datetime.now()
            comp = run(solution_path, stdin=input_file, stdout=PIPE, stderr=PIPE, timeout=timeout)
            end_time = datetime.now()
            comp.check_returncode()
        except CalledProcessError as err:
            verdict = "Error"
        except TimeoutExpired as err:
            end_time = datetime.now()
            comp = None
            verdict = "Timeout"
        else:
            verdict = "Okay"
    
    elapsed_seconds = (end_time - start_time).total_seconds()

    if verdict == "Okay":
        got = comp.stdout.decode()
        with open(output_file_name) as output_file:
            expected = output_file.read()
        if not evaluator(got, expected):
            verdict = "Wrong Answer"
    
    return verdict, elapsed_seconds, comp
    
def check_tests(test_names, solution_name, timeout=None, verbose=True):
    test_names = sorted(test_names)
    if len(test_names)==0: return set()
    print("Testing:", ", ".join(test_names))
    print()
    failed = set()
    for test_name in test_names:
        if verbose: print(f"{test_name}: ", end="", flush=True)
        verdict, elapsed, description = check_test(test_name, solution_name, timeout=timeout)
        if verdict!="Okay": failed.add(test_name)
        if verbose:
            print(f"{elapsed}s. {verdict}")
            if verdict == "Error":
                print(description.stderr.decode())
    print()
    if len(failed) == 0: print(f"Testing All Clear! ({len(test_names)} cases)")
    else:
        print(f"{len(failed)}/{len(test_names)} Test Cases Failed!")
        print(f"Failing Cases:", ", ".join(failed))
    print()

    return failed

def verify_test(test_name, verifier_name):
    input_file_name = input_path(test_name)

    verifier_path = path.join(VERS_DIR, verifier_name)

    with open(input_file_name) as input_file:
        description = run(verifier_path, stdin=input_file, stdout=PIPE, stderr=PIPE)

    passed = (description.returncode == 0)
    
    return passed, description


def verify_tests(test_names, verifier_name, verbose=True):
    test_names = sorted(test_names)
    if len(test_names)==0: return set()

    print("Verifying:", ", ".join(test_names))
    print()
    failed = set()
    for test_name in test_names:
        passed, description = verify_test(test_name, verifier_name)
        
        if verbose:
            print(f"{test_name}: ", end="", flush=True)
            if passed:
                print("Okay.")
            else:
                print("Failed!")
                print(description.stderr.decode())
        
        if not passed:
            failed.add(test_name)
    
    print()
    if len(failed) == 0: print(f"Verification All Clear! ({len(test_names)} cases)")
    else:
        print(f"{len(failed)}/{len(test_names)} Failed Input Verification!")
        print(f"Failing Cases:", ", ".join(failed)) 
    print()
    return failed


def tests(re_list, verifier_name=None, solution_name=None, timeout=None, verbose=True):
    input_file_names = get_files_from_re_list(re_list, IN_DIR, IN_SUFFIX)
    output_file_names = get_files_from_re_list(re_list, OUT_DIR, OUT_SUFFIX)

    missing_outputs = input_file_names.difference(output_file_names)
    missing_inputs = output_file_names.difference(input_file_names)
    test_names = input_file_names.intersection(output_file_names)

    if len(missing_outputs) > 0:
        print(f"{len(missing_outputs)} inputs missing outputs:", ", ".join(sorted(missing_outputs)))
    
    if len(missing_inputs) > 0:
        print(f"{len(missing_inputs)} outputs missing inputs:", ", ".join(sorted(missing_inputs)))
    
    if verbose: print(f"{len(test_names)} tests:", ", ".join(sorted(test_names)))
    print()

    if verifier_name is not None:
        failed = verify_tests(test_names, verifier_name, verbose=verbose)
        test_names.difference_update(failed)
    
    if solution_name is not None:
        check_tests(test_names, solution_name, timeout=timeout, verbose=verbose)
    
    print("Done.")


if __name__ == "__main__":
    from getopt import getopt, GetoptError
    import sys

    short_opts = "s:v:t:qa"
    try:
        opts, args = getopt(sys.argv[1:], short_opts)
    except GetoptError as err:
        print(err)
        sys.exit(2)
    
    opt_dict = {opt:value for opt, value in opts}

    verifier_name = opt_dict.get("-v")
    solution_name = opt_dict.get("-s")
    timeout_str = opt_dict.get("-t")
    verbose = "-q" not in opt_dict

    abort_on_error = "a" in opt_dict
    failure = False

    if verifier_name is not None:
        if not path.isfile(path.join(VERS_DIR, verifier_name)):
            print(f"No such verifier ({verifier_name})!")
            failure = True

    if solution_name is not None:
        if not path.isfile(path.join(SOLS_DIR, solution_name)):
            print(f"No such solution ({solution_name})!")
            failure = True
    
    if timeout_str is not None:
        try:
            timeout = float(timeout_str)
            assert timeout > 0
        except (ValueError, AssertionError):
            print(f"Invalid Timeout ({timeout_str})")
            failure=True
    else:
        timeout=None

    patterns = set(arg.rstrip(",") for arg in args)

    if len(args)==0:
        patterns = {".*"}
    
    re_list = get_re_list(patterns)
    if len(re_list) == 0:
        print("No Valid Patterns")
        failure=True
    elif abort_on_error and len(re_list) < len(patterns):
        failure=True
    
    if failure:
        print("Aborting.")
        sys.exit(2)
    else:
        tests(re_list, verifier_name=verifier_name, solution_name=solution_name, timeout=timeout, verbose=verbose)