autograder.py

"""
Student autograder utilities.

This file provides a common interface for the CS 61A project
student-side autograder. Students do not need to read or understand
the contents of this file.

Usage:
This file is intended to run as the main script. Test cases should
be defined in two files:

* locked_tests.py
* unlocked_tests.py

This file supports the following primary options:

* Unlocking tests (-u): Students will receive test cases in locked
    form. Test cases can be unlocked by using the "-u" flag. Once a
    test is unlocked, students will be able to run the autograder with
    those cases (see below)
* Testing individual questions (-q): Students can test an individual
    question using the "-q" flag. Omitting the "-q" flag will test
    all unlocked test cases. By default, the autograder will stop once
    the first error is encountered (see below)
* Testing all questions regardless of errors (-a): When specified, the
    autograder continues running even if it encounters errors. The
    "-a" flag works even if the "-q" is used
* Interactive debugging (-i): by default, when the autograder
    encounters an error, it prints the stack trace and terminates. The
    "-i" will instead print the stack trace, then open an interactive
    console to allow students to inspect the environment.
"""

import argparse
from code import InteractiveConsole
import hmac
import pickle
import rlcompleter
import re
import traceback
import urllib.request
import os

__version__ = '1.0'

#############
# Utilities #
#############

class TestException(Exception):
    """Custom exception for autograder."""

    def __init__(self, test_src, outputs, explanation='', preamble='',
                 timeout=None):
        super().__init__()
        self.test_src = test_src
        self.outputs = outputs
        self.explanation = explanation
        self.preamble = preamble
        self.timeout=timeout

def underline(line, under='='):
    """Underlines a given line with the specified under style.

    PARAMETERS:
    line  -- str
    under -- str; a one-character string that specifies the underline
             style

    RETURNS:
    str; the underlined version of line
    """
    return line + '\n' + under * len(line)

def display_prompt(line, prompt='>>> '):
    """Formats a given line as if it had been typed in an interactive
    interpreter.

    PARAMETERS:
    line   -- object; represents a line of Python code. If not a
              string, line will be converted using repr. Otherwise,
              expected to contain no newlines for aesthetic reasons
    prompt -- str; prompt symbol. If a space is desired between the
              symbol and input, prompt must contain the space itself
    RETURNS:
    str; the formatted version of line
    """
    if type(line) != str:
        line = repr(line)
    return prompt + line


class TimeoutError(Exception):
    _message = 'Evaluation timed out!'

    def __init__(self, timeout):
        super().__init__(self)
        self.timeout = timeout

TIMEOUT = 10
def timed(fn, args=(), kargs={}, timeout=TIMEOUT):
    """Evaluates expr in the given frame.

    PARAMETERS:
    expr  -- str; Python expression to be evaluated
    frame -- dict; environment in which expr should be evaluated
    """
    from threading import Thread
    class ReturningThread(Thread):
        """Creates a daemon Thread with a result variable."""
        def __init__(self):
            Thread.__init__(self)
            self.daemon = True
            self.result = None
            self.error = None
        def run(self):
            try:
                self.result = fn(*args, **kargs)
            except Exception as e:
                e._message = traceback.format_exc(limit=2)
                self.error = e
    submission = ReturningThread()
    submission.start()
    submission.join(timeout)
    if submission.is_alive():
        raise TimeoutError(timeout)
    if submission.error is not None:
        raise submission.error
    return submission.result

#####################
# Testing Mechanism #
#####################

PS1 = '>>> '
PS2 = '... '

def get_name(test):
    return test['name'] if type(test['name'])==str else test['name'][0]

def run(test, global_frame=None, interactive=False):
    """Runs all test suites for this class.

    PARAMETERS:
    test         -- dict; test cases for a single question
    global_frame -- dict; bindings for the global frame
    interactive  -- bool; True if interactive mode is enabled

    DESCRIPTION:
    Test suites should be correspond to the key 'suites' in test.
    If no such key exists, run as if zero suites are
    defined. Use the first value corresponding to the key 'name' in
    test as the name of the test.

    RETURNS:
    bool; True if all suites passed.
    """
    name = get_name(test)
    print(underline('Test ' + name))
    if global_frame is None:
        global_frame = {}
    if 'note' in test:
        print('\n'.join(process_input(test['note'])))
    if 'suites' not in test:
        test['suites'] = []
    if 'cache' in test:
        test['cache'](global_frame)

    preamble = ''
    if 'preamble' in test and 'all' in test['preamble']:
        preamble += test['preamble']['all']
    postamble = ''
    if 'postamble' in test and 'all' in test['postamble']:
        postamble += test['postamble']['all']

    passed = 0
    for counter, suite in enumerate(test['suites']):
        # Preamble and Postamble
        new_preamble = preamble
        if 'preamble' in test:
            new_preamble += test['preamble'].get(counter, '')
        new_postamble = postamble
        if 'postamble' in test:
            new_postamble += test['postamble'].get(counter, '')
        new_postamble = compile('\n'.join(process_input(new_postamble)),
                  '{} suite {} postamble'.format(name, counter),
                  'exec')
        # Run tests
        try:
            run_suite(new_preamble, suite, global_frame.copy(),
                      '{} suite {}'.format(name, counter))
        except TestException as e:
            exec(new_postamble, global_frame.copy())
            failed = handle_failure(e, counter + 1,
                                       global_frame.copy(),
                                       interactive)
            assert failed, 'Autograder error'
            break
        else:
            passed += 1
        finally:
            exec(new_postamble, global_frame.copy())
    total_cases = 0
    for suite in test['suites']:
        total_cases += len(suite)
    if passed == len(test['suites']):
        print('All unlocked tests passed!')
    if test['total_cases'] and total_cases < test['total_cases']:
        print('Note: {} still has {} locked cases.'.format(
                name,
                test['total_cases'] - total_cases))
    print()
    return passed == len(test['suites'])

def run_suite(preamble, suite, global_frame, label):
    """Runs tests for a single suite.

    PARAMETERS:
    preamble     -- str; the preamble that should be run before
                    every test
    suite        -- list; each element is a test case, represented
                    as a 2-tuple or 3-tuple
    global_frame -- dict; global frame

    DESCRIPTION:
    Each test case in the parameter suite is represented as a
    2-tuple or a 3-tuple:

        (input, outputs)
        (input, outputs, explanation)

    where:
    input       -- str; a (possibly multiline) string of Python
                   source code
    outputs     -- iterable or string; if string, outputs is the
                   sole expected output. If iterable, each element
                   in outputs should correspond to an input slot
                   in input (delimited by '$ ').
    explanation -- (optional) str; an explanation for the test

    For each test, a new frame is created and houses all bindings
    made by the test. The preamble will run first (if it exists)
    before the test input.

    Expected output and actual output is tested on shallow equality
    (==). If a test fails, a TestException will be raised that
    contains information about the test.

    RAISES:
    TestException; contains information about the test that failed.
    """
    new_preamble = compile('\n'.join(process_input(preamble)),
                       '{} preamble'.format(label), 'exec')
    exec(new_preamble, global_frame)

    for test, outputs, *explanation in suite:
        frame = global_frame.copy()
        if type(outputs) == str:
            outputs = (outputs,)
        out_iter = iter(outputs)

        lines = process_input(test)
        current, prompts = '', 0
        for i, line in enumerate(lines):
            if line.startswith('$ ') or \
                    (i == len(lines) - 1 and prompts == 0):
                prompts += 1
                try:
                    exec(current, frame)
                except:
                    raise TestException(test, outputs, explanation,
                                        preamble)
                current = ''
                output = next(out_iter)
                expect = eval(output, frame)
                try:
                    actual = timed(eval, (line.lstrip('$ '), frame))
                except:
                    raise TestException(test, outputs, explanation,
                                        preamble)

                if expect != actual:
                    if explanation:
                        explanation = explanation[0]
                    else:
                        explanation = ''
                    raise TestException(test, outputs, explanation,
                                        preamble)
            else:
                current += line + '\n'
        if prompts == 0:
            output = next(out_iter)
            expect = eval(output, frame)
            actual = eval(line.lstrip('$ '), frame)
            if expect != actual:
                if explanation:
                    explanation = explanation[0]
                else:
                    explanation = ''
                raise TestException(test, outputs, explanation,
                                    preamble)

def handle_failure(error, suite, global_frame, interactive):
    """Handles a test failure.

    PARAMETERS:
    error        -- TestException; contains information about the
                    failed test
    suite        -- int; suite number (for informational purposes)
    global_frame -- dict; global frame
    interactive  -- bool; True if interactive mode is enabled

    DESCRIPTION:
    Expected output and actual output are checked with shallow
    equality (==).

    RETURNS:
    bool; True if error actually occurs, which should always be
    the case -- handle_failure should only be called if a test
    fails.
    """
    print(underline('Test case failed:'.format(suite), under='-'))
    console = InteractiveConsole(locals=global_frame)
    incomplete = False
    for line in process_input(error.preamble):
        if not incomplete and not line:
            incomplete = False
            continue
        prompt = PS2 if incomplete else PS1
        print(display_prompt(line, prompt))
        incomplete = console.push(line)

    incomplete = False
    outputs = iter(error.outputs)
    lines = process_input(error.test_src)
    prompts = 0
    for i, line in enumerate(lines):
        if line.startswith('$ ') or \
                (i == len(lines) - 1 and prompts == 0):
            line = line.lstrip('$ ')
            prompt = PS2 if incomplete else PS1
            print(display_prompt(line, prompt))

            expect = eval(next(outputs), global_frame.copy())
            try:
                actual = timed(eval, (line, global_frame.copy()))
            except RuntimeError:
                print('# Error: maximum recursion depth exceeded')
                if interactive:
                    interact(console)
                print()
                return True
            except TimeoutError as e:
                print('# Error: evaluation exceeded {} seconds'.format(e.timeout))
                return True
            except Exception as e:
                console.push(line)
                print('# Error: expected', repr(expect), 'got',
                      e.__class__.__name__)
                if interactive:
                    interact(console)
                print()
                return True

            print(display_prompt(actual, prompt=''))
            if expect != actual:
                print('# Error: expected', repr(expect), 'got', repr(actual))
                if interactive:
                    interact(console)
                print()
                return True
            incomplete = False
        else:
            if not incomplete and not line:
                incomplete = False
                continue
            prompt = PS2 if incomplete else PS1
            print(display_prompt(line, prompt))
            incomplete = console.push(line)
    print()
    return False

def interact(console):
    """Starts an interactive console."""
    console.interact('# Interactive console\n'
                     '# Type exit() to quit')

def process_input(src):
    """Splits a (possibly multiline) string of Python input into
    a list, adjusting for common indents.

    PARAMETERS:
    src -- str; (possibly) multiline string of Python input

    DESCRIPTION:
    Indentation adjustment is determined by the first nonempty
    line. The characters of indentation for that line will be
    removed from the front of each subsequent line.

    RETURNS:
    list of strings; lines of Python input
    """
    src = src.lstrip('\n').rstrip()
    match = re.match('\s+', src)
    if match:
        length = len(match.group(0))
    else:
        length = 0
    return [line[length:] for line in src.split('\n')]

##########################
# Command Line Interface #
##########################

def run_preamble(preamble, frame):
    """Displays the specified preamble."""
    console = InteractiveConsole(frame)
    incomplete = False
    for line in process_input(preamble):
        if not incomplete and not line:
            incomplete = False
            continue
        prompt = PS2 if incomplete else PS1
        print(display_prompt(line, prompt))
        incomplete = console.push(line)

def get_test(tests, question):
    """Retrieves a test for the specified question in the given list
    of tests.

    PARAMETERS:
    tests    -- list of dicts; list of tests
    quesiton -- str; name of test

    RETURNS:
    dict; the test corresponding to question. If no such test is found,
    return None
    """
    for test in tests:
        if 'name' not in test:
            continue
        names = test['name']
        if type(names) == str:
            names = (names,)
        if question in names:
            return test

def unlock(question, locked_tests, unlocked_tests):
    """Unlocks a question, given locked_tests and unlocked_tests.

    PARAMETERS:
    question       -- str; the name of the test
    locked_tests   -- module; contains a list of locked tests
    unlocked_tests -- module; contains a list of unlocked tests

    DESCRIPTION:
    This function incrementally unlocks all cases in a specified
    question. Students must answer in the order that test cases are
    written. Once a test case is unlocked, it will remain unlocked.

    Persistant state is stored by rewriting the contents of
    locked_tests.py and unlocked_tests.py. Students should NOT manually
    change these files.
    """
    hash_key = locked_tests['hash_key']
    imports = unlocked_tests['project_info']['imports']

    locked = get_test(locked_tests['tests'], question)
    unlocked = get_test(unlocked_tests['tests'], question)
    name = get_name(locked)

    prompt = '?'
    print(underline('Unlocking tests for {}'.format(name)))
    print('At each "{}", type in what you would expect the output to be if you had implemented {}'.format(prompt, name))
    print('Type exit() to quit')
    print()

    global_frame = {}
    for line in imports:
        exec(line, global_frame)

    has_preamble = 'preamble' in locked
    if has_preamble and 'all' in locked['preamble']:
        run_preamble(locked['preamble']['all'], global_frame)

    def hash_fn(x):
        return hmac.new(hash_key.encode('utf-8'),
                        x.encode('utf-8')).digest()

    if 'suites' not in locked:
        return
    for suite_num, suite in enumerate(locked['suites']):
        assert suite_num <= len(unlocked['suites']), 'Incorrect number of suites'
        if not suite:
            continue
        if has_preamble and suite_num in locked['preamble']:
            run_preamble(locked['preamble'][suite_num],
                         global_frame.copy())
            unlocked.setdefault('preamble', {})[suite_num] = locked['preamble'][suite_num]
        while suite:
            case = suite[0]
            lines = [x.strip() for x in case[0].split('\n')
                     if x.strip() != '']
            answers = []
            outputs = case[1]
            if type(outputs) not in (list, tuple):
                outputs = [outputs]
            output_num = 0
            for line in lines:
                if len(lines) > 1 and not line.startswith('$'):
                    print(">>> " + line)
                    continue
                line = line.lstrip('$ ')
                print(">>> " + line)
                correct = False
                while not correct:
                    try:
                        student_input = input(prompt + ' ')
                    except (KeyboardInterrupt, EOFError):
                        try:
                            print('\nExiting unlocker...')
                        # When you use Ctrl+C in Windows, it throws
                        # two exceptions, so you need to catch both of
                        # them.... aka Windows is terrible.
                        except (KeyboardInterrupt, EOFError):
                            pass
                        return
                    if student_input in ('exit()', 'quit()'):
                        print('\nExiting unlocker...')
                        return
                    correct = hash_fn(student_input) == outputs[output_num]
                    if not correct:
                        print("Not quite...try again!")
                answers.append(student_input)
                output_num += 1
            case[1] = answers
            if len(unlocked['suites']) == suite_num:
                unlocked['suites'].append([case])
            else:
                unlocked['suites'][-1].append(case)
            suite.pop(0)

            print("Congratulations, you have unlocked this case!")

    print("You have unlocked all of the tests for this question!")

def check_for_updates(remote, version):
    print('You are running version', version, 'of the autograder')
    try:
        url = os.path.join(remote, 'autograder.py')
        data = timed(urllib.request.urlopen, (url,), timeout=2)
        contents = data.read().decode('utf-8')
    except (urllib.error.URLError, urllib.error.HTTPError):
        print("Couldn't check remote autograder")
        return False
    except TimeoutError:
        print("Checking for updates timed out.")
        return False
    remote_version = re.search("__version__\s*=\s*'(.*)'",
                               contents)
    if remote_version and remote_version.group(1) != version:
        print('Version', remote_version.group(1), 'is available.')
        prompt = input('Do you want to automatically download these files? [y/n]: ')
        if 'y' in prompt.lower():
            with open('autograder.py', 'w') as new:
                new.write(contents)
                print('... autograder.py updated!')
            exit(0)
        else:
            print('You can download the new autograder from the following link:')
            print()
            print('\t' + os.path.join(remote, 'autograder.py'))
            print()
            return True
    return False

def run_all_tests():
    """Runs a command line interface for the autograder."""
    parser = argparse.ArgumentParser(description='CS61A autograder')
    parser.add_argument('-u', '--unlock', type=str, 
                        help='Unlocks the specified question')
    parser.add_argument('-q', '--question', type=str,
                        help='Run tests for the specified question')
    parser.add_argument('-a', '--all', action='store_true',
                        help='Runs all tests, regardless of failures')
    parser.add_argument('-i', '--interactive', action='store_true',
                        help='Enables interactive mode upon failure')
    args = parser.parse_args()

    with open('unlocked_tests.pkl', 'rb') as f:
        unlocked_tests = pickle.load(f)
    new = check_for_updates(unlocked_tests['project_info']['remote'],
                            __version__)
    if new:
        exit(0)
    print()

    if args.unlock:
        with open('locked_tests.pkl', 'rb') as f:
            locked_tests = pickle.load(f)
        unlock(args.unlock, locked_tests, unlocked_tests)

        with open('locked_tests.pkl', 'wb') as f:
            pickle.dump(locked_tests, f, pickle.HIGHEST_PROTOCOL)
        with open('unlocked_tests.pkl', 'wb') as f:
            pickle.dump(unlocked_tests, f, pickle.HIGHEST_PROTOCOL)
    else:
        if args.question:
            tests = get_test(unlocked_tests['tests'], args.question)
            if not tests:
                print('Test {} does not exist'.format(args.question))
                exit(1)
            tests = [tests]
        else:
            tests = unlocked_tests['tests']

        global_frame = {}
        for line in unlocked_tests['project_info']['imports']:
            exec(line, global_frame)
        for test in tests:
            passed = run(test, global_frame, args.interactive)
            if not args.all and not passed:
                return
        print(underline('Note:', under='-'))
        print("""Remember that the tests in this autograder are not exhaustive, so try your own tests in the interpreter!""")


if __name__ == '__main__':
    run_all_tests()