| #!/usr/bin/python |
| # |
| # Copyright (c) 2010 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| |
| # DESCRIPTION : |
| # |
| # This UI is intended to be used by the factory autotest suite to |
| # provide factory operators feedback on test status and control over |
| # execution order. |
| # |
| # In short, the UI is composed of a 'console' panel on the bottom of |
| # the screen which displays the autotest log, and there is also a |
| # 'test list' panel on the right hand side of the screen. The |
| # majority of the screen is dedicated to tests, which are executed in |
| # seperate processes, but instructed to display their own UIs in this |
| # dedicated area whenever possible. Tests in the test list are |
| # executed in order by default, but can be activated on demand via |
| # associated keyboard shortcuts. As tests are run, their status is |
| # color-indicated to the operator -- greyed out means untested, yellow |
| # means active, green passed and red failed. |
| |
| |
| import gobject |
| import gtk |
| import imp |
| import os |
| import pango |
| import select |
| import signal |
| import subprocess |
| import sys |
| import time |
| |
| import common |
| import factory |
| import factory_state |
| import factory_ui_lib as ful |
| import utils |
| |
| from gtk import gdk |
| from Xlib import X |
| from Xlib.display import Display as X_Display |
| |
| |
| # These definitions are expose these classes directly into this |
| # namespace, so that the test_list that is sent from the control file |
| # can have cleaner syntax. These are done in this fashion, as opposed |
| # to "from factory import <class>" to work-around Python namespace |
| # wackiness -- the from syntax does not work, creating new classes. |
| OperatorTest = factory.OperatorTest |
| InformationScreen = factory.InformationScreen |
| AutomatedSequence = factory.AutomatedSequence |
| AutomatedSubTest = factory.AutomatedSubTest |
| AutomatedRebootSubTest = factory.AutomatedRebootSubTest |
| |
| |
| _LABEL_EN_SIZE = (170, 35) |
| _LABEL_ZW_SIZE = (70, 35) |
| _LABEL_EN_FONT = pango.FontDescription('courier new extra-condensed 16') |
| _LABEL_ZW_FONT = pango.FontDescription('normal 12') |
| _LABEL_T_SIZE = (40, 35) |
| _LABEL_T_FONT = pango.FontDescription('arial ultra-condensed 10') |
| _LABEL_UNTESTED_FG = gtk.gdk.color_parse('grey40') |
| _LABEL_TROUGH_COLOR = gtk.gdk.color_parse('grey20') |
| _LABEL_STATUS_SIZE = (140, 30) |
| _LABEL_STATUS_FONT = pango.FontDescription( |
| 'courier new bold extra-condensed 16') |
| _OTHER_LABEL_FONT = pango.FontDescription('courier new condensed 20') |
| |
| _ST_LABEL_EN_SIZE = (250, 35) |
| _ST_LABEL_ZW_SIZE = (150, 35) |
| |
| class Console: |
| '''Display a progress log. Implemented by launching an borderless |
| xterm at a strategic location, and running tail against the log.''' |
| |
| def __init__(self, allocation): |
| xterm_coords = '145x13+%d+%d' % (allocation.x, allocation.y) |
| factory.log('xterm_coords = %s' % xterm_coords) |
| xterm_opts = ('-bg black -fg lightgray -bw 0 -g %s' % xterm_coords) |
| xterm_cmd = (('urxvt %s -e bash -c ' % xterm_opts).split() + |
| ['tail -f "%s"' % factory.CONSOLE_LOG_PATH]) |
| factory.log('xterm_cmd = %s' % xterm_cmd) |
| self._proc = subprocess.Popen(xterm_cmd) |
| |
| def __del__(self): |
| factory.log('console_proc __del__') |
| self._proc.kill() |
| |
| |
| # Capture keyboard events here for debugging -- under normal |
| # circumstances, all keyboard events should be captured by executing |
| # tests, and hence this should not be called. |
| |
| def handle_key_release_event(_, event): |
| factory.log('base ui key event (%s)' % event.keyval) |
| return True |
| |
| |
| class TestLabelBox(gtk.EventBox): |
| |
| def __init__(self, test, show_shortcut=False): |
| gtk.EventBox.__init__(self) |
| self.modify_bg(gtk.STATE_NORMAL, ful.LABEL_COLORS[ful.UNTESTED]) |
| |
| label_en = ful.make_label(test.label_en, size=_LABEL_EN_SIZE, |
| font=_LABEL_EN_FONT, alignment=(0.5, 0.5), |
| fg=_LABEL_UNTESTED_FG) |
| label_zw = ful.make_label(test.label_zw, size=_LABEL_ZW_SIZE, |
| font=_LABEL_ZW_FONT, alignment=(0.5, 0.5), |
| fg=_LABEL_UNTESTED_FG) |
| label_t = ful.make_label('C-' + test.kbd_shortcut.upper(), |
| size=_LABEL_T_SIZE, |
| font=_LABEL_T_FONT, alignment=(0.5, 0.5), |
| fg=ful.BLACK) |
| |
| # build a better label_en with shortcuts |
| index_hotkey = test.label_en.upper().find(test.kbd_shortcut.upper()) |
| if show_shortcut and index_hotkey >= 0: |
| attrs = label_en.get_attributes() or pango.AttrList() |
| attrs.insert(pango.AttrUnderline( |
| pango.UNDERLINE_LOW, index_hotkey, index_hotkey + 1)) |
| attrs.insert(pango.AttrWeight( |
| pango.WEIGHT_BOLD, index_hotkey, index_hotkey + 1)) |
| label_en.set_attributes(attrs) |
| |
| hbox = gtk.HBox() |
| hbox.pack_start(label_en, False, False) |
| hbox.pack_start(label_zw, False, False) |
| hbox.pack_start(label_t, False, False) |
| self.add(hbox) |
| self.label_list = [label_en, label_zw] |
| |
| def update(self, status): |
| label_fg = status == ful.UNTESTED and _LABEL_UNTESTED_FG or ful.BLACK |
| for label in self.label_list: |
| label.modify_fg(gtk.STATE_NORMAL, label_fg) |
| self.modify_bg(gtk.STATE_NORMAL, ful.LABEL_COLORS[status]) |
| self.queue_draw() |
| |
| |
| class SubTestLabelBox(gtk.EventBox): |
| |
| def __init__(self, test): |
| gtk.EventBox.__init__(self) |
| self.modify_bg(gtk.STATE_NORMAL, ful.BLACK) |
| label_status = ful.make_label(ful.UNTESTED, size=_LABEL_STATUS_SIZE, |
| alignment=(0, 0.5), |
| font=_LABEL_STATUS_FONT, |
| fg=_LABEL_UNTESTED_FG) |
| label_sep = ful.make_label(' : ', alignment=(0.5, 0.5), |
| font=_LABEL_EN_FONT) |
| label_en = ful.make_label(test.label_en, size=_ST_LABEL_EN_SIZE, |
| alignment=(0.5, 0.5), |
| font=_LABEL_EN_FONT) |
| label_zw = ful.make_label(test.label_zw, size=_ST_LABEL_ZW_SIZE, |
| alignment=(0.5, 0.5), font=_LABEL_ZW_FONT) |
| hbox = gtk.HBox() |
| hbox.pack_end(label_status, False, False) |
| hbox.pack_end(label_sep, False, False) |
| hbox.pack_end(label_zw, False, False) |
| hbox.pack_end(label_en, False, False) |
| self.add(hbox) |
| self.label_status = label_status |
| |
| def update(self, status): |
| self.label_status.set_text(status) |
| self.label_status.modify_fg(gtk.STATE_NORMAL, ful.LABEL_COLORS[status]) |
| self.queue_draw() |
| |
| |
| class UiState(): |
| |
| def __init__(self, test_widget_box, automated_sequence_set): |
| |
| def make_empty_test_label_widget(): |
| label_box = gtk.EventBox() |
| label_box.modify_bg(gtk.STATE_NORMAL, ful.BLACK) |
| label = ful.make_label('no active test', font=_OTHER_LABEL_FONT, |
| alignment=(0.5, 0.5)) |
| label_box.add(label) |
| return label_box |
| |
| def make_automated_sequence_label_widget(subtest_list): |
| vbox = gtk.VBox() |
| vbox.set_spacing(0) |
| for subtest in subtest_list: |
| label_box = SubTestLabelBox(subtest) |
| self.set_label_box(subtest, label_box) |
| vbox.pack_start(label_box, False, False) |
| return vbox |
| |
| self._label_box_map = {} |
| self._test_widget_box = test_widget_box |
| self._empty_test_widget = make_empty_test_label_widget() |
| self._active_test_widget = self._empty_test_widget |
| self.active_test = None |
| |
| self._test_widget_box.add(self._empty_test_widget) |
| |
| self._automated_seq_widget_map = dict( |
| (test, make_automated_sequence_label_widget(test.subtest_list)) |
| for test in automated_sequence_set) |
| |
| def status_change_callback(self, test, status): |
| label_box = self._label_box_map.get(test) |
| if label_box: |
| label_box.update(status) |
| |
| def set_label_box(self, test, label_box): |
| self._label_box_map[test] = label_box |
| |
| def set_active_test(self, test_db, test): |
| '''Control what kind of widget is shown in the testing area of |
| the screen. For normal operator tests, this is just a label |
| saying there is no active test. The expectation is that the |
| operator test itself has a window obscuring this message. For |
| automated sequences, since there is no other window, the no |
| active test message is replaced with an updated list of |
| subtest status.''' |
| if test is None or test == self.active_test: |
| return |
| factory.log('UI active test is %s' % test_db.get_unique_id_str(test)) |
| self.active_test = test |
| self._test_widget_box.remove(self._active_test_widget) |
| active_widget = (isinstance(test, factory.AutomatedSequence) |
| and self._automated_seq_widget_map[test] |
| or self._empty_test_widget) |
| self._test_widget_box.add(active_widget) |
| self._active_test_widget = active_widget |
| self._test_widget_box.show_all() |
| |
| |
| def state_change_handler(fd, cb_condition, test_db, test_state, ui_state): |
| # Currently the 'state change' event (sent via fd) is fired when each single |
| # test changed its state, but the state_change_handler will change all tests |
| # at once. To get best performance, we wait N=merge_events_duration seconds |
| # to collect and accumulate all changes occured in a short period. |
| events = 0 |
| max_merged_events = 10 |
| merge_events_duration = 0.2 |
| while (events < max_merged_events and |
| len(select.select((fd,), (), (), merge_events_duration)[0]) > 0): |
| fd.read(1) |
| events = events + 1 |
| |
| try: |
| factory.log('factory_ui: refresh status (merged %s events)' % events) |
| for name, status in test_state.get_all_tests().items(): |
| test = test_db.get_test_by_unique_id_str(name) |
| ui_state.status_change_callback(test, status) |
| if (status == factory.ACTIVE and |
| (not isinstance(test, AutomatedSubTest))): |
| ui_state.set_active_test(test_db, test) |
| except Exception, e: |
| factory.log('state_change_handler got exception: %s', repr(e)) |
| return True |
| |
| def kill_active_test(control_pid): |
| # Currently we kill the active test directly. |
| # If we want to use the old way - notify main control to kill it, |
| # then send USR1 to main control instead: |
| # os.kill(control_pid, signal.SIGUSR1) |
| active_test_data = factory.get_shared_data('active_test_data') |
| if active_test_data is None: |
| return |
| factory.log('KILLING active test %s' % repr(active_test_data)) |
| try: |
| utils.nuke_pid(active_test_data[1]) |
| # To remove the dependency of utils, use kill: |
| # os.kill(active_test_data[1], signal.SIGTERM) |
| except OSError, e: |
| factory.log('Failed to kill test: ' + repr(e)) |
| |
| def grab_shortcut_keys(kbd_shortcut_set, control_pid): |
| disp = X_Display() |
| root = disp.screen().root |
| |
| # We want to receive KeyPress events |
| root.change_attributes(event_mask = X.KeyPressMask) |
| |
| keycode_map = {} |
| for shortcut in kbd_shortcut_set: |
| keysym = gdk.keyval_from_name(shortcut) |
| keycode = disp.keysym_to_keycode(keysym) |
| keycode_map[keycode] = shortcut |
| root.grab_key(keycode, X.ControlMask, 1, |
| X.GrabModeAsync, X.GrabModeAsync) |
| |
| # This flushes the XGrabKey calls to the server. |
| for x in range(0, root.display.pending_events()): |
| root.display.next_event() |
| |
| def handle_xevent(src, cond, xhandle=root.display, |
| keycode_map=keycode_map, |
| control_pid=control_pid): |
| for i in range(0, xhandle.pending_events()): |
| xevent = xhandle.next_event() |
| if xevent.type == X.KeyPress: |
| keycode = xevent.detail |
| factory.set_shared_data('activated_kbd_shortcut', |
| keycode_map[keycode]) |
| kill_active_test(control_pid) |
| return True |
| |
| gobject.io_add_watch(root.display, gobject.IO_IN, handle_xevent) |
| |
| |
| def main(test_list, status_file_path, control_pid): |
| '''This process is launched by the autotest suite_Factory control |
| process, which should be identified by the <control pid> cmdline |
| argument. When operators press keyboard shortcuts, the shortcut |
| value is sent by set_shared_data() and a SIGUSR1 is sent to |
| the control program.''' |
| |
| test_db = factory.TestDatabase(test_list) |
| test_state = factory_state.get_instance() |
| |
| window = gtk.Window(gtk.WINDOW_TOPLEVEL) |
| window.connect('destroy', lambda _: gtk.main_quit()) |
| window.modify_bg(gtk.STATE_NORMAL, ful.BLACK) |
| |
| screen = window.get_screen() |
| if (screen is None): |
| factory.log('ERROR: communication with the X server is not working, ' + |
| 'could not find a working screen. UI exiting.') |
| sys.exit(1) |
| |
| screen_size = (screen.get_width(), screen.get_height()) |
| window.set_size_request(*screen_size) |
| |
| label_trough = gtk.VBox() |
| label_trough.set_spacing(0) |
| |
| rhs_box = gtk.EventBox() |
| rhs_box.modify_bg(gtk.STATE_NORMAL, _LABEL_TROUGH_COLOR) |
| rhs_box.add(label_trough) |
| |
| console_box = gtk.EventBox() |
| console_box.set_size_request(-1, 180) |
| console_box.modify_bg(gtk.STATE_NORMAL, ful.BLACK) |
| |
| test_widget_box = gtk.Alignment(xalign=0.5, yalign=0.5) |
| test_widget_box.set_size_request(-1, -1) |
| |
| lhs_box = gtk.VBox() |
| lhs_box.pack_end(console_box, False, False) |
| lhs_box.pack_start(test_widget_box) |
| lhs_box.pack_start(ful.make_hsep(3), False, False) |
| |
| base_box = gtk.HBox() |
| base_box.pack_end(rhs_box, False, False) |
| base_box.pack_end(ful.make_vsep(3), False, False) |
| base_box.pack_start(lhs_box) |
| |
| window.connect('key-release-event', handle_key_release_event) |
| window.add_events(gtk.gdk.KEY_RELEASE_MASK) |
| |
| ui_state = UiState(test_widget_box, test_db.get_automated_sequences()) |
| |
| for test in test_list: |
| label_box = TestLabelBox(test, True) |
| ui_state.set_label_box(test, label_box) |
| label_trough.pack_start(label_box, False, False) |
| label_trough.pack_start(ful.make_hsep(), False, False) |
| |
| window.add(base_box) |
| window.show_all() |
| |
| kbd_shortcut_db = factory.KbdShortcutDatabase(test_list, test_db) |
| grab_shortcut_keys(kbd_shortcut_db.get_shortcut_keys(), control_pid) |
| |
| ful.hide_cursor(window.window) |
| |
| test_widget_allocation = test_widget_box.get_allocation() |
| test_widget_size = (test_widget_allocation.width, |
| test_widget_allocation.height) |
| factory.set_shared_data('test_widget_size', test_widget_size) |
| |
| console = Console(console_box.get_allocation()) |
| gobject.io_add_watch(sys.stdin, gobject.IO_IN, state_change_handler, |
| test_db, test_state, ui_state) |
| |
| # This is to notify suite_Factory/control that UI is ready. |
| sys.stdout.write('Factory UI Ready\n') |
| sys.stdout.flush() |
| |
| factory.log('factory_ui setup done, starting gtk.main()...') |
| gtk.main() |
| factory.log('factory_ui gtk.main() finished, exiting.') |
| |
| if __name__ == '__main__': |
| |
| if len(sys.argv) != 4: |
| print ('usage: %s <test list path> <status file path> <control pid>' % |
| sys.argv[0]) |
| test_list_path, status_file_path, control_pid_str = sys.argv[1:] |
| control_pid = int(control_pid_str) |
| |
| execfile(test_list_path) |
| |
| main(TEST_LIST, status_file_path, control_pid) |
