diff options
Diffstat (limited to 'testapp/exam/code_server.py')
| -rwxr-xr-x | testapp/exam/code_server.py | 198 |
1 files changed, 11 insertions, 187 deletions
diff --git a/testapp/exam/code_server.py b/testapp/exam/code_server.py index f5c5698..db30798 100755 --- a/testapp/exam/code_server.py +++ b/testapp/exam/code_server.py @@ -19,7 +19,6 @@ settings.py:SERVER_POOL_PORT. This port exposes a `get_server_port` function that returns an available server. """ import sys -import traceback from SimpleXMLRPCServer import SimpleXMLRPCServer import pwd import os @@ -34,13 +33,15 @@ import importlib # Local imports. from settings import SERVER_PORTS, SERVER_TIMEOUT, SERVER_POOL_PORT from registry import registry - +from evaluate_python import EvaluatePython +from evaluate_c import EvaluateC +from evaluate_cpp import EvaluateCpp +from evaluate_java import EvaluateJava +from evaluate_scilab import EvaluateScilab +from evaluate_bash import EvaluateBash MY_DIR = abspath(dirname(__file__)) -registry.register('python', ) -registry.register('py', MyTestCode) - def run_as_nobody(): """Runs the current process as nobody.""" # Set the effective uid and to that of nobody. @@ -48,166 +49,6 @@ def run_as_nobody(): os.setegid(nobody.pw_gid) os.seteuid(nobody.pw_uid) - -# Raised when the code times-out. -# c.f. http://pguides.net/python/timeout-a-function -class TimeoutException(Exception): - pass - - -def timeout_handler(signum, frame): - """A handler for the ALARM signal.""" - raise TimeoutException('Code took too long to run.') - -def create_signal_handler(): - """Add a new signal handler for the execution of this code.""" - prev_handler = signal.signal(signal.SIGALRM, timeout_handler) - signal.alarm(SERVER_TIMEOUT) - return prev_handler - -def set_original_signal_handler(old_handler=None): - """Set back any original signal handler.""" - if old_handler is not None: - signal.signal(signal.SIGALRM, old_handler) - return - else: - raise Exception("Signal Handler: object cannot be NoneType") - -def delete_signal_handler(): - signal.alarm(0) - return - - - -############################################################################### -# `TestCode` class. -############################################################################### -class TestCode(object): - """Tests the code obtained from Code Server""" - def __init__(self, test_parameter, language, user_answer, ref_code_path=None, in_dir=None): - msg = 'Code took more than %s seconds to run. You probably '\ - 'have an infinite loop in your code.' % SERVER_TIMEOUT - self.timeout_msg = msg - self.test_parameter = test_parameter - self.language = language.lower() - self.user_answer = user_answer - self.ref_code_path = ref_code_path - self.in_dir = in_dir - - def run_code(self): - """Tests given code (`answer`) with the test cases based on - given arguments. - - The ref_code_path is a path to the reference code. - The reference code will call the function submitted by the student. - The reference code will check for the expected output. - - If the path's start with a "/" then we assume they are absolute paths. - If not, we assume they are relative paths w.r.t. the location of this - code_server script. - - If the optional `in_dir` keyword argument is supplied it changes the - directory to that directory (it does not change it back to the original - when done). - - Returns - ------- - - A tuple: (success, error message). - """ - self._change_dir(self.in_dir) - - # Add a new signal handler for the execution of this code. - prev_handler = create_signal_handler() - success = False - - # Do whatever testing needed. - try: - success, err = self.evaluate_code() - - except TimeoutException: - err = self.timeout_msg - except: - type, value = sys.exc_info()[:2] - err = "Error: {0}".format(repr(value)) - finally: - # Set back any original signal handler. - set_original_signal_handler(prev_handler) - - # Cancel the signal - delete_signal_handler() - - result = {'success': success, 'error': err} - return result - - def evaluate_code(self): - raise NotImplementedError("evaluate_code method not implemented") - - def _create_submit_code_file(self, file_name): - """ Write the code (`answer`) to a file and set the file path""" - # File name/extension depending on the question language - submit_f = open(file_name, 'w') - submit_f.write(self.user_answer.lstrip()) - submit_f.close() - submit_path = abspath(submit_f.name) - if sfile_elf.language == "bash": - self._set_file_as_executable(submit_path) - - return submit_path - - def _set_file_as_executable(self, fname): - os.chmod(fname, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR - | stat.S_IRGRP | stat.S_IWGRP | stat.S_IXGRP - | stat.S_IROTH | stat.S_IWOTH | stat.S_IXOTH) - - def _set_test_code_file_path(self, ref_path=None, test_case_path=None): - # ref_path, test_case_path = self.ref_code_path.split(',') - - if ref_path and not ref_path.startswith('/'): - ref_path = join(MY_DIR, ref_path) - if test_case_path and not test_case_path.startswith('/'): - test_case_path = join(MY_DIR, test_case_path) - - return ref_path, test_case_path - - def _run_command(self, cmd_args, *args, **kw): - """Run a command in a subprocess while blocking, the process is killed - if it takes more than 2 seconds to run. Return the Popen object, the - stdout and stderr. - """ - try: - proc = subprocess.Popen(cmd_args, *args, **kw) - stdout, stderr = proc.communicate() - except TimeoutException: - # Runaway code, so kill it. - proc.kill() - # Re-raise exception. - raise - return proc, stdout, stderr - - def _compile_command(self, cmd, *args, **kw): - """Compiles C/C++/java code and returns errors if any. - Run a command in a subprocess while blocking, the process is killed - if it takes more than 2 seconds to run. Return the Popen object, the - stderr. - """ - try: - proc_compile = subprocess.Popen(cmd, shell=True, stdin=None, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - out, err = proc_compile.communicate() - except TimeoutException: - # Runaway code, so kill it. - proc_compile.kill() - # Re-raise exception. - raise - return proc_compile, err - - def _change_dir(self, in_dir): - if in_dir is not None and isdir(in_dir): - os.chdir(in_dir) - - ############################################################################### # `CodeServer` class. ############################################################################### @@ -219,34 +60,17 @@ class CodeServer(object): self.port = port self.queue = queue - def check_code(self, info_parameter, in_dir=None): - """Calls the TestCode Class to test the current code""" - info_parameter = json.loads(info_parameter) - test_parameter = info_parameter.get("test_parameter") - language = info_parameter.get("language") - user_answer = info_parameter.get("user_answer") - ref_code_path = info_parameter.get("ref_code_path") - - eval_module_name = "evaluate_{0}".format(language.lower()) - eval_class_name = "Evaluate{0}".format(language.capitalize()) - - get_class = self._sub_class_factory(eval_module_name, eval_class_name) + def check_code(self, info_parameter, language, in_dir=None): + """Calls the TestCode SUb Class based on language to test the current code""" + evaluate_code_class = registry.get_class(language) + evaluate_code_instance = evaluate_code_class.from_json(info_parameter, language, in_dir) + result = evaluate_code_instance.run_code() - test_code_class = get_class(test_parameter, language, user_answer, ref_code_path, in_dir) - result = test_code_class.run_code() # Put us back into the server pool queue since we are free now. self.queue.put(self.port) return json.dumps(result) - def _sub_class_factory(self, module_name, class_name): - # load the module, will raise ImportError if module cannot be loaded - get_module = importlib.import_module(module_name) - # get the class, will raise AttributeError if class cannot be found - get_class = getattr(get_module, class_name) - - return get_class - def run(self): """Run XMLRPC server, serving our methods.""" server = SimpleXMLRPCServer(("localhost", self.port)) |