From 794efa5dcccf8e7b6e33e1f12f9a25fbda4b4740 Mon Sep 17 00:00:00 2001 From: "gasperfele@fri1.uni-lj.si" Date: Mon, 24 Nov 2014 15:25:26 +0000 Subject: The first working version of test_task.py git-svn-id: https://svn.lusy.fri.uni-lj.si/kpov-public-svn/kpov-public@9 5cf9fbd1-b2bc-434c-b4b7-e852f4f63414 --- kpov_judge/tasks/set_motd/task.py | 6 +- kpov_judge/test_task.py | 192 +++++++++++++++++++++++++++----------- 2 files changed, 138 insertions(+), 60 deletions(-) diff --git a/kpov_judge/tasks/set_motd/task.py b/kpov_judge/tasks/set_motd/task.py index 5367577..f329773 100644 --- a/kpov_judge/tasks/set_motd/task.py +++ b/kpov_judge/tasks/set_motd/task.py @@ -50,7 +50,7 @@ computers = { networks = { 'net1': {'public': False}, 'test-net': {'public': True} } params_meta = { - 'peer_ip': {'descriptions': {'si': 'IP računalnika'}, 'w': True, 'public':True, 'type': 'IP', 'generated': False}, + 'peer_ip': {'descriptions': {'si': u'IP računalnika'}, 'w': True, 'public':True, 'type': 'IP', 'generated': False}, 'peer_user': {'descriptions': {'si': 'ime uporabnika'}, 'w': False, 'public': True, 'type': 'username', 'generated': True}, 'peer_passwd': {'descriptions': {'si': 'geslo uporabnika'}, 'w': True, 'public': True, 'type': 'alnumstr', 'generated': False}, 'niz': {'w': False, 'public': True, 'type': 'short_text', 'generated': True}, @@ -63,7 +63,7 @@ def task(peer_ip, peer_user, peer_passwd, niz): results = dict() s = pxssh.pxssh() s.login (peer_ip, peer_user, peer_passwd, original_prompt=r'{0}@.*:\~\$'.format(peer_user),auto_prompt_reset=False) - temp = s.before # print everything before the prompt. + temp = s.before # print everything before the prompt. results['motd'] = temp.replace("#","") #zamenja vse pojavitve "#" s praznim stringom #results['motd'] = s.before # print everything before the prompt. prvotna verzija return results @@ -73,7 +73,7 @@ def gen_params(user_id, params_meta): def task_check(results, params): niz = params['niz'] - niz=niz.replace("#","") + niz=niz.replace("#","") score = 0 if (results['motd'].find(niz) > -1): score += 10 diff --git a/kpov_judge/test_task.py b/kpov_judge/test_task.py index 449be35..c35aabd 100755 --- a/kpov_judge/test_task.py +++ b/kpov_judge/test_task.py @@ -5,12 +5,13 @@ import yaml import json import urllib import urllib2 -import dialog +from dialog import Dialog import os import inspect import sys from bson.son import SON import kpov_random_helpers +import argparse # SERVER_URL = "http://localhost/kpov_judge/" #SERVER_URL = "http://kpov.fri.uni-lj.si/kpov_judge/" @@ -19,14 +20,16 @@ import kpov_random_helpers #SUBMIT_URL = SERVER_URL + '{task_id}/results.json' TASK_URL = "file://./" -PARAMS_URL = None PARAMS_FILE = os.path.expanduser("~/.kpov_params.yaml") DEFAULT_LANGUAGE = 'si' """get the parameters for a task either from the user or from a file.""" -def get_params_dialog(params, param_name_list, meta): - lang = params['language'] +def get_params_dialog(params, meta, param_name_list=None, dialog=None, language = None): + if param_name_list is None: + param_name_list = meta.keys() + if language is None: + language = params.get('language', DEFAULT_LANGUAGE) for name in param_name_list: if name not in meta: pass @@ -34,29 +37,38 @@ def get_params_dialog(params, param_name_list, meta): while not got_param: m = meta.get(name, {}) try: - description = m['descriptions'][lang] + description = m['descriptions'][language] except KeyError: - description = '' - ret, s = dialog.inputbox(m.get('descriptions', name), init=params.get(name, '')) - if ret == 0 and m.get('w', False): - params[name] = s + description = name + if params.get(name, None) is None: + init = '' + else: + init = params[name] + ret, s = dialog.inputbox(description, init=init) + if ret == 0: + if m.get('w', True): + params[name] = s + elif name not in params: + params[name] = None got_param = (ret == 0) and name in params return params -def get_params_web(params, param_name_list, meta, url=PARAMS_URL): - req = urllib2.Request(url.format(**saved_params)) - response = urllib2.urlopen(req) - params = json.load(response) - return params - -def get_params_args(params, param_names_list, meta): - pass +def add_meta_to_argparser(argparser, meta, defaults = {}): + language = defaults.get('language', DEFAULT_LANGUAGE) + for k, v in meta.iteritems(): + try: + desc = v['descriptions'][language].encode("utf-8") + except: + desc = k + print desc, type(desc) + argparser.add_argument('--'+k, nargs=1, help=desc, type=unicode, default=defaults.get(k, None)) + return argparser -def auth_open(url, username, password): +def http_auth(url, username, password): password_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm() # Add the username and password. # If we knew the realm, we could use it instead of None. - password_mgr.add_password(None, SERVER_URL, username, password) + password_mgr.add_password(None, url, username, password) handler = urllib2.HTTPBasicAuthHandler(password_mgr) # create "opener" (OpenerDirector instance) opener = urllib2.build_opener(handler) @@ -74,57 +86,123 @@ def load_task(stream): exec(t) # get a list of arguments for task(...) # args_list = inspect.getargs(task.func_code)[0] - return task, task_check, params_meta + return task, task_check, params_meta, gen_params +def locate_task(params, argparser, dialog): + # first the URL where all tasks are stored + url_meta = { + 'task_url': {'descriptions': {'si': 'URL z nalogami', 'en': 'Root URL for all tasks'}, } + } + if 'task_url' not in params: + params['task_url'] = TASK_URL + argparser = add_meta_to_argparser(argparser, meta = url_meta, defaults = params) + args = argparser.parse_args() + params['task_url'] = args.task_url + if dialog is not None: + params = get_params_dialog(params, url_meta, dialog = dialog) + # then the student's ID (and password if neccessarry) + fetch_params_meta = { + 'username':{'descriptions': {'si': 'Uporabniško ime', 'en': 'Username'}, }, + } + if params['task_url'] is not None and params['task_url'].startswith('http'): + fetch_params_meta['password'] = {'descriptions': {'si': 'Geslo', 'en': 'Password'}, }, + # and finally, the name of the task + fetch_params_meta['task_name'] = {'descriptions': {'si': 'Ime naloge', 'en': 'Task name'}, } + argparser = add_meta_to_argparser(argparser, meta = fetch_params_meta, defaults = params) + args = argparser.parse_args() + if dialog is not None: + params = get_params_dialog(params, fetch_params_meta, dialog=dialog) + return argparser, params -if __name__ == '__main__': - # get default parameters +def load_params(filename): try: - with open(PARAMS_FILE) as f: + with open(filename) as f: saved_params = yaml.load(f) except: saved_params = dict() + if saved_params is None: + saved_params = dict() + return saved_params + +if __name__ == '__main__': # get the parameters needed to get to the task, such as the URLs, the name of the task and optionally an ID from the student - import argparse - task_argparser = argparse.ArgumentParser(description='Get the task parameters', conflict_handler='resolve', add_help=False) - task_argparser.add_argument('-h', '--help', action='store_true') - task_argparser.add_argument('-q', '--quiet', action='store_true', help='disable dialog') - task_argparser.add_argument('-pf','--params_file', nargs='?', + # start with the the parameters needed for the dialog gui to work + basic_argparser = argparse.ArgumentParser(description='Get the task parameters', conflict_handler='resolve', add_help=False) + basic_argparser.add_argument('-h', '--help', action='store_true') + basic_argparser.add_argument('-q', '--quiet', action='store_true', help='disable dialog', default=False) + basic_argparser.add_argument('-g', '--generate_params', action='store_true', help='Generate initial values the task parameters', default=False) + basic_argparser.add_argument('-pf','--params_file', nargs='?', help='a local file containing saved param values', default=PARAMS_FILE) - basic_args = task_argparser.parse_args() - print basic_args - task_argparser.add_argument('-pu', '--params_url', nargs=1, - help='the URL for task parameters', default=PARAMS_URL) - task_argparser.add_argument('-tu', '--task_url', nargs='?', - help='the root URL for all tasks', default=TASK_URL) - task_argparser.add_argument('-l', '--language', nargs=1, - help='the language used', default=DEFAULT_LANGUAGE) - task_argparser.add_argument('task_name', help='task name') - args = task_argparser.parse_args() - print args - lang = args.language - print args.task_dir, args.task_name - task_fname = os.path.join(args.task_dir, args.task_name, 'task.py') - # get task source + basic_args = basic_argparser.parse_args() + # get default parameters including language + params = load_params(basic_args.params_file) + basic_argparser.add_argument('-l', '--language', nargs=1, + help='the language used', default=params.get('language', DEFAULT_LANGUAGE)) + basic_args = basic_argparser.parse_args() + params['language'] = basic_args.language + if not basic_args.quiet: + dialog = Dialog(dialog="dialog") + else: + dialog = None + # continue with the parameters needed to get the task + argparser, params = locate_task(params, basic_argparser, dialog=dialog) + # TODO: if the task name is missing or invalid, try to get a list of tasks + # get task source and generate params if neccessarry try: - with open(task_fname) as source: - task, task_check, params_meta = load_task(source) - except: - if args.help: - task_argparser.print_help() + if params['task_url'].startswith('http'): + http_auth(params['task_url'], params['username'], params['password']) + req = urllib2.Request("{task_url}/{task_name}/task.py".format(**params)) + source = urllib2.urlopen(req) + task, task_check, task_params_meta, gen_params = load_task(source) + except Exception, e: + print e + if basic_args.help: + argparser.print_help() else: - task_argparser.print_usage() + argparser.print_usage() + with open(basic_args.params_file, 'w') as f: + yaml.dump(params, f) exit(1) # get task parameters - params_argparser = argparse.ArgumentParser(parents=[task_argparser], conflict_handler='resolve', add_help=True) - for pm_name, pm in params_meta.iteritems(): - params_argparser.add_argument('--' + pm_name, nargs = 1, help=pm['descriptions'][lang]) - if args.help: - params_argparser.print_help() - args = params_argparser.parse_args() + params['task_params'] = params.get('task_params', dict()) + params['task_params'][params['task_name']] = params['task_params'].get(params['task_name'], dict()) + task_params = params['task_params'][params['task_name']] + if basic_args.generate_params: + task_params.update(gen_params(params['username'], task_params_meta)) + if params['task_url'].startswith('http'): + req = urllib2.Request("{task_url}/{task_name}/params.json".format(**params)) + web_task_params = json.load(urllib2.urlopen(req)) + task_params.update(web_task_params) + params_argparser = argparse.ArgumentParser(parents=[argparser], conflict_handler='resolve', add_help=True) + params_argparser = add_meta_to_argparser(params_argparser, task_params_meta, defaults=task_params) + args = vars(params_argparser.parse_args()) + for k in task_params_meta: + if k in args and args[k] is not None: + task_params[k] = args[k] + if dialog is not None: + print task_params + task_params = get_params_dialog(task_params, task_params_meta, dialog=dialog, language = params['language']) + if basic_args.help: + # params_argparser.print_help() + print params_argparser.format_help() # run task.task() - s = task(**public_params) + public_params = dict() + for k in inspect.getargs(task.func_code)[0]: + public_params[k] = task_params[k] + params['task_params'][params['task_name']] = task_params + try: + s = task(**public_params) # run task.task_check() - result = task_check(s, params) + except Exception, e: + print("Error running task(...):") + print(e) + try: + result = task_check(s, task_params) + except Exception, e: + print("Error running task_check(results, params):") + print(e) # print results + # save parameters for the next run + with open(basic_args.params_file, 'w') as f: + yaml.dump(params, f) -- cgit v1.2.1