run_autograder

#!/usr/bin/env python3
import grade_util as gu
import bootstrap
import shutil
import os
import stat
import glob
import yaml
import json
import subprocess
import unittest
import logging
from typing import List

from gradescope_utils.autograder_utils.decorators import weight, visibility
from gradescope_utils.autograder_utils.json_test_runner import JSONTestRunner

logging.basicConfig()
LOGGER = logging.getLogger('autograder')
LOGGER.setLevel(logging.INFO)

ZERO_LEADERBOARD = [
    {
        'name': 'Score',
        'value': 0
    }
]

def ZERO_RESULT(msg):
    return {
        'score': 0.0,
        'stdout_visibility': 'hidden',
        'output': msg,
        'leaderboard': ZERO_LEADERBOARD
    }


def BAD_FORMAT(file):
    return ZERO_RESULT('Required file not submitted: \'{0}\'.'.format(file))


def SUBMISSIONS_EXCEEDED(limit):
    return ZERO_RESULT('Exceeded maximum number of submissions: {}'.format(limit))


def NUM_SUBMISSIONS_INFO(num, limit):
    return ZERO_RESULT('Submission {} out of {}'.format(num, limit))


#  class used in generating unittest TestCase's
class Test(type):

    def __new__(mcs, test, bases, attrs):
        attrs[test.__doc__] = test
        return super(Test, mcs).__new__(mcs, test.__doc__, bases, attrs)


def get_test_dirs() -> List[str]:
    '''
    Returns a list of directory paths, as strings, for each test contained in 'tests' directory
    '''
    return [dr[:-1] for dr in glob.glob(gu.Config.TEST_DIR + '/*/')]


def load_yaml(file: str):
    try:
        with open(file, 'r') as f:
            return yaml.safe_load(f) or {}
    except FileNotFoundError:
        return {}


def generate_test(dir_name):
    settings = load_yaml(os.path.join(gu.Config.TEST_DIR, dir_name, 'test.yml'))
    LOGGER.info('Generating test \'{}\''.format(settings.get('name', '')))

    def run_test():
        run_path = os.path.join(gu.Config.TEST_DIR, dir_name, 'run_test')
        os.chmod(run_path, os.stat(run_path).st_mode | stat.S_IEXEC)
        subprocess.check_call(['dos2unix', 'run_test'], cwd=os.path.join(gu.Config.TEST_DIR, dir_name),
                              stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        subprocess.check_output(['./run_test'], cwd=os.path.join(gu.Config.TEST_DIR, dir_name),
                                stderr=subprocess.STDOUT, timeout=settings.get('timeout', None))

    @weight(settings.get('weight', 1))
    @visibility(settings.get('visibility', 'visible'))
    def wrapper(self):
        show_output = settings.get('show_output', True)
        try:
            run_test()
        except subprocess.CalledProcessError as e:  # test script returned non-zero
            msg = '{}\n\n{}'.format(settings.get('message', ''),
                                    e.output.decode() if show_output else '')
            raise Exception(msg)
        except subprocess.TimeoutExpired as e:
            msg = '{}\n\n{}'.format('Test timed out', e.output.decode() if show_output else '')
            raise Exception(msg)

    wrapper.__doc__ = '{}'.format(settings.get('name', os.path.basename(dir_name)))
    return wrapper


def run_tests(pending_messages):
    suite = unittest.TestSuite()

    #  go through all directories in /tests
    for dr in get_test_dirs():
        LOGGER.info('Test directory discovered: {}'.format(dr))
        if not gu.file_exists(os.path.join(gu.Config.TEST_DIR, dr, 'run_test')):
            LOGGER.error('Test improperly configured: missing \'run_test\' executable')
            exit(-1)

        test_fn = generate_test(dr)
        t = Test(test_fn, (unittest.TestCase,), {})
        suite.addTest(t(test_fn.__doc__))

    with open(gu.Config.RESULTS_FILE, 'w+') as result_stream:
        JSONTestRunner(stdout_visibility='hidden', stream=result_stream).run(suite)

        LOGGER.info('Adding final score to leaderboard')
        result_stream.seek(0)  # reset file pointer
        data = json.load(result_stream)  # load contents to JSON
        final_score = data.get('score', 0.0)
        data['leaderboard'] = [{'name': 'Score', 'value': final_score}]

        LOGGER.info('Prepending additional messages to output')
        if 'tests' not in data:
            data['tests'] = []
        for msg in pending_messages:
            data['tests'].insert(0, msg)
        result_stream.truncate(0)
        result_stream.seek(0)

        json.dump(data, result_stream)  # rewrite JSON to file


#  Execution begins here
if __name__ == '__main__':
    bootstrap.init()

    pending_messages = []
    LOGGER.info('Loading config.yml file')
    config = load_yaml('config.yml')

    limit = config.get('limit_submissions', -1)
    LOGGER.info('Using submission limit: {}'.format(limit))
    if limit is not -1:
        num_submissions = gu.number_submissions() + 1
        if num_submissions > limit:
            LOGGER.info('Submission limit reached. Observed submissions: {}'.format(num_submissions))
            gu.write_result(**SUBMISSIONS_EXCEEDED(limit))
            exit(0)
        else:
            pending_messages.append((NUM_SUBMISSIONS_INFO(num_submissions, limit)))

    for file in config.get('required_files', []):
        LOGGER.info('Attempting to load file: {}'.format(file))
        #  check if required file is submitted
        if not gu.is_submitted(file):
            LOGGER.info('File not submitted: {}'.format(file))
            gu.write_result(**BAD_FORMAT(file))
            exit(0)
        #  copy submitted file to shared test directory
        LOGGER.info('Copying file to test directory: {}'.format(file))
        shutil.copy(os.path.join(gu.Config.SUBMITTED_SOURCE, file), gu.Config.TEST_DIR)

    run_tests(pending_messages)