| #!/usr/bin/python |
| |
| # Copyright (C) 2006 - 2007 Michal Piotrowski <michal.k.k.piotrowski@gmail.com> |
| # |
| # This program is free software; you can redistribute it and/or modify |
| # it under the terms of the GNU General Public License as published by |
| # the Free Software Foundation; either version 2 of the License, or |
| # (at your option) any later version. |
| |
| import os, sys, dircache, string, dialog, time |
| from menu_lib import * |
| |
| def handle_exit_code(d, code): |
| if code in (d.DIALOG_CANCEL, d.DIALOG_ESC): |
| if d.yesno("Do you want to exit Autotest Control Center?") == d.DIALOG_OK: |
| sys.exit(0) |
| return 0 |
| else: |
| return 1 |
| |
| def atcc_t_menu(test_type, t): |
| u = atcc_list_control_files(test_type, at_dir) |
| |
| (code, tag) = d.checklist(text = test_type + ":", choices = u, title = t) |
| |
| return tag |
| |
| def atcc_t_run(res, test_type): |
| if len(res) == 0: |
| return -1 |
| |
| atcc_setup_tmp_dirs_files(menu_dir) |
| |
| for i in res: |
| os.system(at_dir + '/bin/autotest ' + at_dir + '/' + test_type + '/' + i) |
| atcc_save_results1(i, at_dir, menu_dir) |
| |
| def atcc_t_p_run(res, test_type): |
| if len(res) == 0: |
| return -1 |
| |
| atcc_setup_tmp_dirs_files(menu_dir) |
| |
| file = (menu_dir + '/tmp/parallel') |
| f = open(file, 'w') |
| |
| for i in res: |
| z = i.split('/') |
| line = ("def " + z[0] + "():\n") |
| z = str(line) |
| f.write(z) |
| |
| file = (at_dir + '/' + test_type + '/' + i) |
| f2 = open(file, 'r') |
| k = f2.readlines() |
| |
| for i in k: |
| x = ("\t" + i + "\n") |
| z = str(x) |
| f.write(z) |
| |
| f2.close() |
| |
| f.write('job.parallel(') |
| |
| for i in range(len(res)): |
| z = res[i].split('/') |
| z = ('[' + z[0] + '],') |
| z = str(z) |
| f.write(z) |
| |
| f.write(')') |
| |
| f.close() |
| |
| os.system(at_dir + '/bin/autotest ' + menu_dir + '/tmp/parallel') |
| atcc_save_results1("", at_dir, menu_dir) |
| |
| def atcc_profilers_run(res1, res2): |
| if len(res1) == 0 or len(res2) == 0: |
| return -1 |
| |
| atcc_setup_tmp_dirs_files(menu_dir) |
| |
| file = (menu_dir + '/tmp/profilers') |
| f = open(file, 'w') |
| |
| f.write('for profiler in (') |
| |
| for i in range(len(res1)): |
| z = res1[i].split('/') |
| z = ('\'' + z[0] + '\', ') |
| z = str(z) |
| f.write(z) |
| |
| f.write('):\n') |
| |
| f.write('\ttry:\n') |
| f.write('\t\tprint "Testing profiler %s ..." % profiler\n') |
| f.write('\t\tjob.profilers.add(profiler)\n') |
| |
| for i in range(len(res2)): |
| z = atcc_control_file_read(res2[i], 'tests', at_dir) |
| n = res2[i].split('/') |
| k = ('\t\tjob.run_test(\'' + n[0] + '\',' + z + ')\n') |
| k = str(k) |
| f.write(k) |
| |
| f.write('\t\tjob.profilers.delete(profiler)\n') |
| f.write('\texcept:\n') |
| f.write('\t\tprint "Test of profiler %s failed" % profiler\n') |
| f.write('\t\traise\n') |
| |
| f.close() |
| |
| os.system(at_dir + '/bin/autotest ' + menu_dir + '/tmp/profilers') |
| atcc_save_results2(res1, res2, at_dir, menu_dir) |
| |
| def atcc_remote_get_ip(): |
| while 1: |
| (code, answer) = d.inputbox("IP address of remote host", init = "192.168.0.1") |
| |
| if code == d.DIALOG_CANCEL or code == d.DIALOG_ESC: |
| break |
| return answer |
| |
| def atcc_t_remote_run(ip, res, test_type): |
| if len(res) == 0: |
| return -1 |
| |
| atcc_setup_tmp_dirs_files(menu_dir) |
| |
| file = (menu_dir + '/tmp/remote_test') |
| f = open(file, 'w') |
| |
| st = str("remote_host = hosts.SSHHost(\"" + ip +"\")") |
| f.write(st + "\n") |
| st = str("at = autotest.Autotest()") |
| f.write(st + "\n") |
| |
| file2 = (menu_dir + '/tmp/remote_control') |
| f2 = open(file2, 'w') |
| |
| for i in res: |
| print i |
| file3 = (at_dir + '/' + test_type + '/' + i) |
| f3 = open(file3, 'r') |
| k = f3.readlines() |
| for i in k: |
| x = (i + "\n") |
| z = str(x) |
| f2.write(z) |
| f3.close() |
| |
| f2.close() |
| st = str("at.run(\"" + menu_dir + "/tmp/remote_control\",\"results\", remote_host)") |
| f.write(st + "\n") |
| f.close() |
| |
| os.system(srv_dir + '/autoserv ' + menu_dir + '/tmp/remote_test') |
| |
| def atcc_t_p_remote_run(ip, res, test_type): |
| if len(res) == 0: |
| return -1 |
| |
| atcc_setup_tmp_dirs_files(menu_dir) |
| |
| file = (menu_dir + '/tmp/remote_parallel_control') |
| f = open(file, 'w') |
| |
| for i in res: |
| z = i.split('/') |
| line = ("def " + z[0] + "():\n") |
| z = str(line) |
| f.write(z) |
| |
| file = (at_dir + '/' + test_type + '/' + i) |
| f2 = open(file, 'r') |
| k = f2.readlines() |
| |
| for i in k: |
| x = ("\t" + i + "\n") |
| z = str(x) |
| f.write(z) |
| |
| f2.close() |
| |
| f.write('job.parallel(') |
| |
| for i in range(len(res)): |
| z = res[i].split('/') |
| z = ('[' + z[0] + '],') |
| z = str(z) |
| f.write(z) |
| |
| f.write(')') |
| |
| f.close() |
| |
| file = (menu_dir + '/tmp/remote_parallel_test') |
| f = open(file, 'w') |
| |
| st = str("remote_host = hosts.SSHHost(\"" + ip +"\")") |
| f.write(st + "\n") |
| st = str("at = autotest.Autotest()") |
| f.write(st + "\n") |
| |
| st = str("at.run(\"" + menu_dir + "/tmp/remote_parallel_control\",\"results\", remote_host)") |
| f.write(st + "\n") |
| f.close() |
| |
| os.system(srv_dir + '/autoserv ' + menu_dir + '/tmp/remote_parallel_test') |
| |
| def atcc_profilers_remote_run(ip, res1, res2): |
| if len(res1) == 0 or len(res2) == 0: |
| return -1 |
| |
| atcc_setup_tmp_dirs_files(menu_dir) |
| |
| file = (menu_dir + '/tmp/remote_profilers_control') |
| f = open(file, 'w') |
| |
| f.write('for profiler in (') |
| |
| for i in range(len(res1)): |
| z = res1[i].split('/') |
| z = ('\'' + z[0] + '\', ') |
| z = str(z) |
| f.write(z) |
| |
| f.write('):\n') |
| |
| f.write('\ttry:\n') |
| f.write('\t\tprint "Testing profiler %s ..." % profiler\n') |
| f.write('\t\tjob.profilers.add(profiler)\n') |
| |
| for i in range(len(res2)): |
| z = atcc_control_file_read(res2[i], 'tests', at_dir) |
| n = res2[i].split('/') |
| k = ('\t\tjob.run_test(\'' + n[0] + '\',' + z + ')\n') |
| k = str(k) |
| f.write(k) |
| |
| f.write('\t\tjob.profilers.delete(profiler)\n') |
| f.write('\texcept:\n') |
| f.write('\t\tprint "Test of profiler %s failed" % profiler\n') |
| f.write('\t\traise\n') |
| |
| f.close() |
| |
| file = (menu_dir + '/tmp/remote_profilers') |
| f = open(file, 'w') |
| |
| st = str("remote_host = hosts.SSHHost(\"" + ip +"\")") |
| f.write(st + "\n") |
| st = str("at = autotest.Autotest()") |
| f.write(st + "\n") |
| |
| st = str("at.run(\"" + menu_dir + "/tmp/remote_profilers_control\",\"results\", remote_host)") |
| f.write(st + "\n") |
| f.close() |
| |
| os.system(srv_dir + '/autoserv ' + menu_dir + '/tmp/remote_profilers') |
| |
| def atcc_tests_results(t): |
| if os.path.exists(menu_dir + "/tmp/"): |
| dir_ls = dircache.listdir(menu_dir + "/tmp/") |
| else: |
| d.infobox(menu_dir + "/tmp/ doesn't exist") |
| time.sleep(5) |
| return -1 |
| |
| if len(dir_ls) == 0: |
| return -1 |
| |
| u = [] |
| |
| for i in dir_ls: |
| k = i, "" |
| u.append(k) |
| |
| while 1: |
| (code, tag) = d.menu("Results:", choices = u, title = t) |
| |
| if code == d.DIALOG_CANCEL or code == d.DIALOG_ESC: |
| break |
| else: |
| d.textbox(menu_dir + '/tmp/' + tag, width = -1) |
| |
| def atcc_config_show_help(tag, test_type): |
| tag = tag.split('/') |
| tag = tag[0] |
| if os.path.exists(at_dir + '/' + test_type + '/' + tag + '/help'): |
| d.textbox(at_dir + '/' + test_type + '/' + tag + '/help', width = -1) |
| else: |
| d.infobox(at_dir + '/' + test_type + '/' + tag + '/help' " doesn't exist") |
| time.sleep(5) |
| |
| def atcc_config_edit(tag, test_type, ed): |
| os.system(ed + " " + at_dir + '/' + test_type + '/' + tag) |
| |
| def atcc_config_help_or_edit(tag, test_type): |
| ed = "vim" |
| while 1: |
| (code, answer) = d.inputbox("Type 'help' to see documentation, or name of favourite text editor", init = ed) |
| |
| if code == d.DIALOG_CANCEL or code == d.DIALOG_ESC: |
| break |
| elif answer == "help": |
| atcc_config_show_help(tag, test_type) |
| continue |
| else: |
| ed = answer |
| atcc_config_edit(tag, test_type, ed) |
| break |
| |
| def atcc_config_choose_control_file(test_type, t): |
| u_tmp = atcc_list_control_files(test_type, at_dir) |
| u = [] |
| for i in u_tmp: |
| k = i[0], "" |
| u.append(k) |
| |
| while 1: |
| (code, tag) = d.menu(test_type + ":", choices = u, title = t) |
| |
| if code == d.DIALOG_CANCEL or code == d.DIALOG_ESC: |
| break |
| else: |
| atcc_config_help_or_edit(tag, test_type) |
| |
| def atcc_upgrade(): |
| os.system("svn checkout svn://test.kernel.org/autotest/trunk " + at_dir + "/../") |
| |
| def atcc_rsync_mirror(): |
| os.system(at_dir + "/../mirror/mirror") |
| |
| def atcc_main_menu(): |
| while 1: |
| (code, tag) = d.menu("Main menu", |
| choices = [("1", "Tests (local)"), |
| ("2", "Parallel tests (local)"), |
| ("3", "Profilers (local)"), |
| ("4", "Tests (remote)"), |
| ("5", "Parallel tests (remote)"), |
| ("6", "Profilers (remote)"), |
| ("7", "Tests' results"), |
| ("8", "Configure tests"), |
| ("9", "Upgrade Autotest"), |
| ("10", "Sync local kernel.org mirror")]) |
| if handle_exit_code(d, code): |
| break |
| return tag |
| |
| def main(): |
| while 1: |
| res = int(atcc_main_menu()) |
| if res == 1: |
| res = atcc_t_menu(test_type = 'tests', t = 'Tests selection menu') |
| atcc_t_run(res, test_type = 'tests') |
| elif res == 2: |
| res = atcc_t_menu(test_type = 'tests', t = 'Parallel tests selection menu') |
| atcc_t_p_run(res, test_type = 'tests') |
| elif res == 3: |
| res1 = atcc_t_menu(test_type = 'profilers', t = 'Profilers selection menu') |
| res2 = atcc_t_menu(test_type = 'tests', t = 'Tests selection menu') |
| atcc_profilers_run(res1, res2) |
| elif res == 4: |
| ip = atcc_remote_get_ip() |
| res = atcc_t_menu(test_type = 'tests', t = 'Tests selection menu') |
| atcc_t_remote_run(ip, res, test_type = 'tests') |
| elif res == 5: |
| ip = atcc_remote_get_ip() |
| res = atcc_t_menu(test_type = 'tests', t = 'Parallel tests selection menu') |
| atcc_t_p_remote_run(ip, res, test_type = 'tests') |
| elif res == 6: |
| ip = atcc_remote_get_ip() |
| res1 = atcc_t_menu(test_type = 'profilers', t = 'Profilers selection menu') |
| res2 = atcc_t_menu(test_type = 'tests', t = 'Tests selection menu') |
| atcc_profilers_remote_run(ip, res1, res2) |
| elif res == 7: |
| atcc_tests_results(t = 'Tests\' results menu') |
| elif res == 8: |
| atcc_config_choose_control_file(test_type = 'tests', t = 'Tests configuration menu') |
| elif res == 9: |
| atcc_upgrade() |
| elif res == 10: |
| atcc_rsync_mirror() |
| elif res == 0: |
| sys.exit(1) |
| |
| check_python_version() |
| |
| menu_dir = os.path.abspath(os.path.dirname(sys.argv[0])) |
| at_dir = os.path.dirname(menu_dir + "/../client/") |
| srv_dir = os.path.dirname(menu_dir + "/../server/") |
| |
| d = dialog.Dialog(dialog = "dialog") |
| d.add_persistent_args(["--backtitle", "Autotest Control Center v0.07"]) |
| |
| main() |