Mercurial > ~astiob > upreckon > hgweb
diff 2.00/testcases.py @ 21:ec6f1a132109
A pretty usable version
Test groups and testconfs in non-ZIP archives or ZIP archives with comments are not yet supported.
author | Oleg Oshmyan <chortos@inbox.lv> |
---|---|
date | Fri, 06 Aug 2010 15:39:29 +0000 |
parents | f2279b7602d3 |
children | f07b7a431ea6 |
line wrap: on
line diff
--- a/2.00/testcases.py Mon Jun 14 21:02:06 2010 +0000 +++ b/2.00/testcases.py Fri Aug 06 15:39:29 2010 +0000 @@ -1,35 +1,380 @@ -#!/usr/bin/python +#! /usr/bin/env python # Copyright (c) 2010 Chortos-2 <chortos@inbox.lv> +from __future__ import division, with_statement + +try: + from compat import * + import files, problem, config +except ImportError: + import __main__ + __main__.import_error(sys.exc_info()[1]) +else: + from __main__ import clock, options + +import glob, re, sys, tempfile, time +from subprocess import Popen, PIPE, STDOUT + +import os +devnull = open(os.path.devnull, 'w+') + +try: + from signal import SIGTERM, SIGKILL +except ImportError: + SIGTERM = 15 + SIGKILL = 9 + try: - import files as _files, problem as _problem -except ImportError as e: - import __main__ - __main__.import_error(e) + from _subprocess import TerminateProcess +except ImportError: + # CPython 2.5 does define _subprocess.TerminateProcess even though it is + # not used in the subprocess module, but maybe something else does not + try: + import ctypes + TerminateProcess = ctypes.windll.kernel32.TerminateProcess + except (ImportError, AttributeError): + TerminateProcess = None + +__all__ = ('TestCase', 'load_problem', 'TestCaseNotPassed', + 'TimeLimitExceeded', 'WrongAnswer', 'NonZeroExitCode', + 'CannotStartTestee', 'CannotStartValidator', + 'CannotReadOutputFile') + + + +# Exceptions + +class TestCaseNotPassed(Exception): __slots__ = () +class TimeLimitExceeded(TestCaseNotPassed): __slots__ = () + +class WrongAnswer(TestCaseNotPassed): + __slots__ = 'comment' + def __init__(self, comment=''): + self.comment = comment + +class NonZeroExitCode(TestCaseNotPassed): + __slots__ = 'exitcode' + def __init__(self, exitcode): + self.exitcode = exitcode + +class ExceptionWrapper(TestCaseNotPassed): + __slots__ = 'upstream' + def __init__(self, upstream): + self.upstream = upstream + +class CannotStartTestee(ExceptionWrapper): __slots__ = () +class CannotStartValidator(ExceptionWrapper): __slots__ = () +class CannotReadOutputFile(ExceptionWrapper): __slots__ = () +class CannotReadInputFile(ExceptionWrapper): __slots__ = () +class CannotReadAnswerFile(ExceptionWrapper): __slots__ = () + + + +# Test case types class TestCase(object): - __slots__ = 'problem', 'infile', 'outfile' + __slots__ = ('problem', 'id', 'isdummy', 'infile', 'outfile', 'points', + 'process', 'time_started', 'time_stopped', 'time_limit_string', + 'realinname', 'realoutname', 'maxtime', 'maxmemory') + + if ABCMeta: + __metaclass__ = ABCMeta - def __init__(case, prob, infile, outfile): + def __init__(case, prob, id, isdummy, points): case.problem = prob - case.infile = infile - case.outfile = outfile + case.id = id + case.isdummy = isdummy + case.points = points + case.maxtime = case.problem.config.maxtime + case.maxmemory = case.problem.config.maxmemory + if case.maxtime: + case.time_limit_string = '/%.3f' % case.maxtime + else: + case.time_limit_string = '' + if not isdummy: + case.realinname = case.problem.config.testcaseinname + case.realoutname = case.problem.config.testcaseoutname + else: + case.realinname = case.problem.config.dummyinname + case.realoutname = case.problem.config.dummyoutname + + @abstractmethod + def test(case): raise NotImplementedError def __call__(case): - os.copy() + try: + return case.test() + finally: + case.cleanup() + + def cleanup(case): + if not getattr(case, 'time_started', None): + case.time_started = case.time_stopped = clock() + elif not getattr(case, 'time_stopped', None): + case.time_stopped = clock() + #if getattr(case, 'infile', None): + # case.infile.close() + #if getattr(case, 'outfile', None): + # case.outfile.close() + if getattr(case, 'process', None): + # Try killing after three unsuccessful TERM attempts in a row + # (except on Windows, where TERMing is killing) + for i in range(3): + try: + try: + case.process.terminate() + except AttributeError: + # Python 2.5 + if TerminateProcess and hasattr(proc, '_handle'): + # Windows API + TerminateProcess(proc._handle, 1) + else: + # POSIX + os.kill(proc.pid, SIGTERM) + except Exception: + time.sleep(0) + case.process.poll() + else: + break + else: + # If killing the process is unsuccessful three times in a row, + # just silently stop trying + for i in range(3): + try: + try: + case.process.kill() + except AttributeError: + # Python 2.5 + if TerminateProcess and hasattr(proc, '_handle'): + # Windows API + TerminateProcess(proc._handle, 1) + else: + # POSIX + os.kill(proc.pid, SIGKILL) + except Exception: + time.sleep(0) + case.process.poll() + else: + break + + def open_infile(case): + try: + case.infile = files.File('/'.join((case.problem.name, case.realinname.replace('$', case.id)))) + except IOError: + e = sys.exc_info()[1] + raise CannotReadInputFile(e) + + def open_outfile(case): + try: + case.outfile = files.File('/'.join((case.problem.name, case.realoutname.replace('$', case.id)))) + except IOError: + e = sys.exc_info()[1] + raise CannotReadAnswerFile(e) + -def load_problem(prob): +class ValidatedTestCase(TestCase): + __slots__ = 'validator' + + def __init__(case, *args): + TestCase.__init__(case, *args) + if not case.problem.config.tester: + case.validator = None + else: + case.validator = case.problem.config.tester + + # TODO + def validate(case, output): + if not case.validator: + # Compare the output with the reference output + case.open_outfile() + with case.outfile.open() as refoutput: + for line, refline in zip(output, refoutput): + if not isinstance(refline, basestring): + line = bytes(line, sys.getdefaultencoding()) + if line != refline: + raise WrongAnswer() + try: + try: + next(output) + except NameError: + output.next() + except StopIteration: + pass + else: + raise WrongAnswer() + try: + try: + next(refoutput) + except NameError: + refoutput.next() + except StopIteration: + pass + else: + raise WrongAnswer() + return case.points + elif callable(case.validator): + return case.validator(output) + else: + # Call the validator program + output.close() + case.open_outfile() + if case.problem.config.ansname: + case.outfile.copy(case.problem.config.ansname) + case.process = Popen(case.validator, stdin=devnull, stdout=PIPE, stderr=STDOUT, universal_newlines=True, bufsize=-1) + comment = case.process.communicate()[0].strip() + lower = comment.lower() + match = re.match(r'(ok|correct|wrong(?:(?:\s|_)*answer)?)(?:$|\s+|[.,!:]+\s*)', lower) + if match: + comment = comment[match.end():] + if not case.problem.config.maxexitcode: + if case.process.returncode: + raise WrongAnswer(comment) + else: + return case.points, comment + else: + return case.points * case.process.returncode / case.problem.config.maxexitcode, comment + + +class BatchTestCase(ValidatedTestCase): + __slots__ = () + + def test(case): + if sys.platform == 'win32' or not case.maxmemory: + preexec_fn = None + else: + def preexec_fn(): + try: + import resource + maxmemory = int(case.maxmemory * 1048576) + resource.setrlimit(resource.RLIMIT_AS, (maxmemory, maxmemory)) + # I would also set a CPU time limit but I do not want the time + # that passes between the calls to fork and exec to be counted in + except MemoryError: + # We do not have enough memory for ourselves; + # let the parent know about this + raise + except Exception: + # Well, at least we tried + pass + case.open_infile() + case.time_started = None + if case.problem.config.stdio: + if options.erase and not case.validator: + # FIXME: 2.5 lacks the delete parameter + with tempfile.NamedTemporaryFile(delete=False) as f: + inputdatafname = f.name + else: + inputdatafname = case.problem.config.inname + case.infile.copy(inputdatafname) + # FIXME: inputdatafname should be deleted on __exit__ + with open(inputdatafname, 'rU') as infile: + with tempfile.TemporaryFile('w+') if options.erase and not case.validator else open(case.problem.config.outname, 'w+') as outfile: + try: + try: + case.process = Popen(case.problem.config.path, stdin=infile, stdout=outfile, stderr=devnull, universal_newlines=True, bufsize=-1, preexec_fn=preexec_fn) + except MemoryError: + # If there is not enough memory for the forked test.py, + # opt for silent dropping of the limit + case.process = Popen(case.problem.config.path, stdin=infile, stdout=outfile, stderr=devnull, universal_newlines=True, bufsize=-1) + except OSError: + raise CannotStartTestee(sys.exc_info()[1]) + case.time_started = clock() + # If we use a temporary file, it may not be a true file object, + # and if so, Popen will relay the standard output through pipes + if not case.maxtime: + case.process.communicate() + case.time_stopped = clock() + else: + time_end = case.time_started + case.maxtime + # FIXME: emulate communicate() + while True: + exitcode = case.process.poll() + now = clock() + if exitcode is not None: + case.time_stopped = now + break + elif now >= time_end: + raise TimeLimitExceeded() + if config.globalconf.force_zero_exitcode and case.process.returncode: + raise NonZeroExitCode(case.process.returncode) + outfile.seek(0) + return case.validate(outfile) + else: + if case.problem.config.inname: + case.infile.copy(case.problem.config.inname) + try: + try: + case.process = Popen(case.problem.config.path, stdin=devnull, stdout=devnull, stderr=STDOUT, preexec_fn=preexec_fn) + except MemoryError: + # If there is not enough memory for the forked test.py, + # opt for silent dropping of the limit + case.process = Popen(case.problem.config.path, stdin=devnull, stdout=devnull, stderr=STDOUT) + except OSError: + raise CannotStartTestee(sys.exc_info()[1]) + case.time_started = clock() + if not case.maxtime: + case.process.wait() + case.time_stopped = clock() + else: + time_end = case.time_started + case.maxtime + while True: + exitcode = case.process.poll() + now = clock() + if exitcode is not None: + case.time_stopped = now + break + elif now >= time_end: + raise TimeLimitExceeded() + if config.globalconf.force_zero_exitcode and case.process.returncode: + raise NonZeroExitCode(case.process.returncode) + with open(case.problem.config.outname, 'rU') as output: + return case.validate(output) + + +# This is the only test case type not executing any programs to be tested +class OutputOnlyTestCase(ValidatedTestCase): + __slots__ = () + def cleanup(case): pass + +class BestOutputTestCase(ValidatedTestCase): + __slots__ = () + +# This is the only test case type executing two programs simultaneously +class ReactiveTestCase(TestCase): + __slots__ = () + # The basic idea is to launch the program to be tested and the grader + # and to pipe their standard I/O from and to each other, + # and then to capture the grader's exit code and use it + # like the exit code of a test validator is used. + + +def load_problem(prob, _types={'batch' : BatchTestCase, + 'outonly' : OutputOnlyTestCase, + 'bestout' : BestOutputTestCase, + 'reactive': ReactiveTestCase}): if prob.config.usegroups: pass else: + # We will need to iterate over these configuration variables twice + try: + len(prob.config.dummies) + except Exception: + prob.config.dummies = tuple(prob.config.dummies) + try: + len(prob.config.tests) + except Exception: + prob.config.dummies = tuple(prob.config.tests) + # First get prob.cache.padoutput right + for i in prob.config.dummies: + s = 'sample ' + str(i).zfill(prob.config.paddummies) + prob.cache.padoutput = max(prob.cache.padoutput, len(s)) for i in prob.config.tests: - s = str(i).zfill(prob.config.padwithzeroestolength) - prob.cache.padoutputtolength = max(prob.cache.padoutputtolength, len(s)) - infile = _files.TestCaseFile(prob, prob.config.testcaseinname.replace('$', s)) - if infile: - if prob.config.kind != _problem.BATCH: - yield TestCase(prob, infile, None) - else: - outfile = _files.TestCaseFile(prob, prob.config.testcaseoutname.replace('$', s)) - if outfile: - yield TestCase(prob, infile, outfile) \ No newline at end of file + s = str(i).zfill(prob.config.padtests) + prob.cache.padoutput = max(prob.cache.padoutput, len(s)) + # Now yield the actual test cases + for i in prob.config.dummies: + s = str(i).zfill(prob.config.paddummies) + yield _types[prob.config.kind](prob, s, True, 0) + for i in prob.config.tests: + s = str(i).zfill(prob.config.padtests) + yield _types[prob.config.kind](prob, s, False, prob.config.pointmap.get(i, prob.config.pointmap.get(None, prob.config.maxexitcode if prob.config.maxexitcode else 1))) \ No newline at end of file