Add support for sequences of fully automated tests.

src/scripts/mod_for_factory_scripts/factory_ui

 #!/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
@@ skipping 17 matching lines @@
 import os
 import pango
 import subprocess
 import sys
 import time
 
 
 def XXX_log(s):
     print >> sys.stderr, '--- XXX : ' + s
 
+_ACTIVE = 'ACTIVE'
+_PASSED = 'PASS'
+_FAILED = 'FAIL'
+_UNTESTED = 'UNTESTED'
 
 _LABEL_COLORS = {
-    'active': gtk.gdk.color_parse('light goldenrod'),
-    'passed': gtk.gdk.color_parse('pale green'),
-    'failed': gtk.gdk.color_parse('tomato'),
-    'untested': gtk.gdk.color_parse('dark slate grey')}
+    _ACTIVE: gtk.gdk.color_parse('light goldenrod'),
+    _PASSED: gtk.gdk.color_parse('pale green'),
+    _FAILED: gtk.gdk.color_parse('tomato'),
+    _UNTESTED: gtk.gdk.color_parse('dark slate grey')}
 
 _LABEL_EN_SIZE = (160, 35)
 _LABEL_EN_FONT = pango.FontDescription('courier new extra-condensed 16')
 _LABEL_ZW_SIZE = (70, 35)
 _LABEL_ZW_FONT = pango.FontDescription('normal 12')
 _LABEL_T_SIZE = (30, 35)
 _LABEL_T_FONT = pango.FontDescription('courier new italic 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')
 _SEP_COLOR = gtk.gdk.color_parse('grey50')
 _BLACK = gtk.gdk.color_parse('black')
 _LIGHT_GREEN = gtk.gdk.color_parse('light green')
-
+_OTHER_LABEL_FONT = pango.FontDescription('courier new condensed 20')
 
 class console_proc:
     '''Display a progress log.  Implemented by launching an borderless
     xterm at a strategic location, and running tail against the log.'''
 
     def __init__(self, allocation, log_file_path):
         xterm_coords = '135x14+%d+%d' % (allocation.x, allocation.y)
         XXX_log('xterm_coords = %s' % xterm_coords)
         xterm_cmd = ('xterm --geometry %s -bw 0 -e ' % xterm_coords +
                      'tail -f %s' % log_file_path)
@@ skipping 13 matching lines @@
     gtk.main_quit()
     return True
 
 def control_recv():
     return eval(sys.stdin.readline().rstrip())
 
 def control_send(x):
     print repr(x)
     sys.stdout.flush()
 
+def control_send_target_test_update(test):
+    XXX_log('ui send_target_test_update %s.%s_%s' %
+            (test.formal_name, test.tag_prefix, test.count))
+    control_send((test.formal_name, test.tag_prefix, test.count))
+
 
 # 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):
     XXX_log('base ui key event (%s)' % event.keyval)
     return True
 
 
-def update_label_status(test, status):
-    if status != 'untested':
-        test.label_box.modify_fg(gtk.STATE_NORMAL, _BLACK)
-        for label in test.label_list:
-            label.modify_fg(gtk.STATE_NORMAL, _BLACK)
-    test.label_box.modify_bg(gtk.STATE_NORMAL, _LABEL_COLORS[status])
-    test.label_box.queue_draw()
+class test_label_box(gtk.EventBox):
+
+    def __init__(self, test):
+        gtk.EventBox.__init__(self)
+        label_en = gtk.Label(test.label_en)
+        label_en.set_size_request(*_LABEL_EN_SIZE)
+        label_en.modify_font(_LABEL_EN_FONT)
+        label_en.set_alignment(0.8, 0.5)
+        label_en.modify_fg(gtk.STATE_NORMAL, _LABEL_UNTESTED_FG)
+        label_zw = gtk.Label(test.label_zw)
+        label_zw.set_size_request(*_LABEL_ZW_SIZE)
+        label_zw.modify_font(_LABEL_ZW_FONT)
+        label_zw.set_alignment(0.2, 0.5)
+        label_zw.modify_fg(gtk.STATE_NORMAL, _LABEL_UNTESTED_FG)
+        label_t = gtk.Label('C-' + test.trigger)
+        label_t.set_size_request(*_LABEL_T_SIZE)
+        label_t.modify_font(_LABEL_T_FONT)
+        label_t.set_alignment(0.5, 0.5)
+        label_t.modify_fg(gtk.STATE_NORMAL, _BLACK)
+        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_status(self, status):
+        if status != _UNTESTED:
+            self.modify_fg(gtk.STATE_NORMAL, _BLACK)
+            for label in self.label_list:
+                label.modify_fg(gtk.STATE_NORMAL, _BLACK)
+        self.modify_bg(gtk.STATE_NORMAL, _LABEL_COLORS[status])
+        self.queue_draw()
+
+
+class subtest_label_box(gtk.EventBox):
+
+    def __init__(self, test):
+        gtk.EventBox.__init__(self)
+        self.modify_bg(gtk.STATE_NORMAL, _BLACK)
+        label_status = gtk.Label(_UNTESTED)
+        label_status.set_size_request(*_LABEL_STATUS_SIZE)
+        label_status.set_alignment(0, 0.5)
+        label_status.modify_font(_LABEL_STATUS_FONT)
+        label_status.modify_fg(gtk.STATE_NORMAL, _LABEL_UNTESTED_FG)
+        label_en = gtk.Label(test.label_en)
+        label_en.set_alignment(1, 0.5)
+        label_en.modify_font(_LABEL_EN_FONT)
+        label_en.modify_fg(gtk.STATE_NORMAL, _LIGHT_GREEN)
+        label_zw = gtk.Label(test.label_zw)
+        label_zw.set_alignment(1, 0.5)
+        label_zw.modify_font(_LABEL_ZW_FONT)
+        label_zw.modify_fg(gtk.STATE_NORMAL, _LIGHT_GREEN)
+        label_sep = gtk.Label(' : ')
+        label_sep.set_alignment(0.5, 0.5)
+        label_sep.modify_font(_LABEL_EN_FONT)
+        label_sep.modify_fg(gtk.STATE_NORMAL, _LIGHT_GREEN)
+        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_status(self, status):
+        if status != _UNTESTED:
+            self.label_status.set_text(status)
+            self.label_status.modify_fg(gtk.STATE_NORMAL, _LABEL_COLORS[status])
+        self.queue_draw()
+
+
+class status_map():
+
+    def __init__(self):
+        self.status_dict = {}
+
+    def index(self, formal_name, tag_prefix):
+        return '%s.%s' % (formal_name, tag_prefix)
+
+    def lookup(self, formal_name, tag_prefix):
+        return self.status_dict.setdefault(
+            self.index(formal_name, tag_prefix),
+            (_UNTESTED, 0))
+
+    def update(self, formal_name, tag_prefix, status, count):
+        _, existing_count = self.lookup(formal_name, tag_prefix)
+        if count > existing_count:
+            index = self.index(formal_name, tag_prefix)
+            self.status_dict[index] = (status, count)
+
+    def get_subtest_status(self, test):
+        map(self.set_test_status, test.automated_seq)
+        sub_status_set = set(st.status for st in test.automated_seq)
+        min_count = min([st.count for st in test.automated_seq])
+        max_count = max([st.count for st in test.automated_seq])
+        if len(sub_status_set) == 1:
+            return (sub_status_set.pop(), max_count)
+        if test.count > min_count:
+            return (_ACTIVE, max_count)
+        return (_FAILED, max_count)
+
+    def set_test_status(self, test):
+        status, count = (
+            test.automated_seq
+            and self.get_subtest_status(test)
+            or self.lookup(test.formal_name, test.tag_prefix))
+        status = test.count > count and _ACTIVE or status
+        max_count = max(test.count, count)
+        if test.status != status or test.count != max_count:
+            XXX_log('status change for %s : %s/%s -> %s/%s' %
+                    (self.index(test.formal_name, test.tag_prefix),
+                     test.count, test.status, max_count, status))
+            test.status = status
+            test.count = max_count
+            test.label_box.update_status(status)
 
 
 def refresh_test_status(status_file_path, test_list):
-    result_dict = {}
+    smap = status_map()
     with open(status_file_path) as file:
         for line in file:
             columns = line.split('\t')
             if len(columns) >= 8 and not columns[0] and not columns[1]:
-                result_state = columns[2]
-                full_name = columns[3]
-                result_dict[full_name] = result_state
-    for test in test_list:
-        full_name = '%s.%d' % (test.formal_name, test.count)
-        result_state = result_dict.get(full_name, None)
-        if result_state is None:
-            status = 'untested'
-        elif result_state == 'GOOD':
-            status = 'passed'
-        else:
-            status = 'failed'
-        if test.status != status:
-            XXX_log('status change for %s : %s -> %s' %
-                    (test.label_en, test.status, status))
-        test.status = status
-        update_label_status(test, status)
+                status = columns[2] == 'GOOD' and _PASSED or _FAILED
+                formal_name, _, tag = columns[3].rpartition('.')
+                tag_prefix, _, count = tag.rpartition('_')
+                count = int(count)
+                smap.update(formal_name, tag_prefix, status, count)
+    map(smap.set_test_status, test_list)
 
 
-def select_active_test(test_list, remaining_tests_queue,
-                       test_counters, trigger):
-    active_test = None
-    if trigger is not None:
-        trigger_dict = dict((test.trigger, test) for test in test_list)
-        active_test = trigger_dict.get(trigger, None)
-        if active_test in remaining_tests_queue:
-            remaining_tests_queue.remove(active_test)
-    if active_test is None:
-        active_test = remaining_tests_queue.pop()
-    count = test_counters[active_test.formal_name]
-    count += 1
-    active_test.count = count
-    test_counters[active_test.formal_name] = count
-    update_label_status(active_test, 'active')
-    XXX_log('select_active_test %s.%d' %
-            (active_test.formal_name, active_test.count))
-    return (active_test.label_en, active_test.count)
-
-
-def make_test_label(test):
-    label_en = gtk.Label(test.label_en)
-    label_en.set_size_request(*_LABEL_EN_SIZE)
-    label_en.modify_font(_LABEL_EN_FONT)
-    label_en.set_alignment(0.8, 0.5)
-    label_en.modify_fg(gtk.STATE_NORMAL, _LABEL_UNTESTED_FG)
-    label_zw = gtk.Label(test.label_zw)
-    label_zw.set_size_request(*_LABEL_ZW_SIZE)
-    label_zw.modify_font(_LABEL_ZW_FONT)
-    label_zw.set_alignment(0.2, 0.5)
-    label_zw.modify_fg(gtk.STATE_NORMAL, _LABEL_UNTESTED_FG)
-    label_t = gtk.Label('C-' + test.trigger)
-    label_t.set_size_request(*_LABEL_T_SIZE)
-    label_t.modify_font(_LABEL_T_FONT)
-    label_t.set_alignment(0.5, 0.5)
-    label_t.modify_fg(gtk.STATE_NORMAL, _BLACK)
-    hbox = gtk.HBox()
-    hbox.pack_start(label_en, False, False)
-    hbox.pack_start(label_zw, False, False)
-    hbox.pack_start(label_t, False, False)
-    label_box = gtk.EventBox()
-    label_box.add(hbox)
-    test.label_box = label_box
-    test.label_list = [label_en, label_zw]
-    return label_box
+def set_active_test(test):
+    test.count += 1
+    test.label_box.update_status(_ACTIVE)
+    control_send_target_test_update(test)
 
 
 def make_hsep(width=1):
     frame = gtk.EventBox()
     frame.set_size_request(-1, width)
     frame.modify_bg(gtk.STATE_NORMAL, _SEP_COLOR)
     return frame
 
 
 def make_vsep(width=1):
     frame = gtk.EventBox()
     frame.set_size_request(width, -1)
     frame.modify_bg(gtk.STATE_NORMAL, _SEP_COLOR)
     return frame
 
 
-def make_test_widget_box():
+def make_notest_label():
     label = gtk.Label('no active test')
-    font = pango.FontDescription('courier new condensed 20')
-    label.modify_font(font)
+    label.modify_font(_OTHER_LABEL_FONT)
     label.set_alignment(0.5, 0.5)
     label.modify_fg(gtk.STATE_NORMAL, _LIGHT_GREEN)
     box = gtk.EventBox()
     box.modify_bg(gtk.STATE_NORMAL, _BLACK)
     box.add(label)
-    align = gtk.Alignment(xalign=0.5, yalign=0.5)
-    align.set_size_request(-1, -1)
-    align.add(box)
-    return align
+    return box
+
+
+def make_automated_seq_widget(as_test):
+    vbox = gtk.VBox()
+    vbox.set_spacing(0)
+    map(lambda st: vbox.pack_start(st.label_box, False, False),
+        as_test.automated_seq)
+    return vbox
 
 
 def main():
     window = gtk.Window(gtk.WINDOW_TOPLEVEL)
     window.connect('destroy', lambda _: gtk.main_quit())
     window.modify_bg(gtk.STATE_NORMAL, _BLACK)
 
     screen = window.get_screen()
     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, _BLACK)
 
-    test_widget_box = make_test_widget_box()
+    notest_label = make_notest_label()
+
+    test_widget_box = gtk.Alignment(xalign=0.5, yalign=0.5)
+    test_widget_box.set_size_request(-1, -1)
+    test_widget_box.add(notest_label)
 
     lhs_box = gtk.VBox()
     lhs_box.pack_end(console_box, False, False)
     lhs_box.pack_start(test_widget_box)
     lhs_box.pack_start(make_hsep(3), False, False)
 
     base_box = gtk.HBox()
     base_box.pack_end(rhs_box, False, False)
     base_box.pack_end(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)
 
     # On startup, get general configuration data from the autotest
     # control program, specifically the list of tests to run (in
     # order) and some filenames.
     XXX_log('pulling control info')
     test_list = control_recv()
     status_file_path = control_recv()
     log_file_path = control_recv()
 
     for test in test_list:
         test.status = None
-        label = make_test_label(test)
-        label_trough.pack_start(label, False, False)
+        test.count = 0
+        test.tag_prefix = test.trigger
+        test.label_box = test_label_box(test)
+        for subtest in test.automated_seq:
+            subtest.status = None
+            subtest.count = 0
+            subtest.tag_prefix = test.formal_name
+            subtest.label_box = subtest_label_box(subtest)
+        label_trough.pack_start(test.label_box, False, False)
         label_trough.pack_start(make_hsep(), False, False)
 
     window.add(base_box)
     window.show_all()
 
     test_widget_allocation = test_widget_box.get_allocation()
     test_widget_size = (test_widget_allocation.width,
                         test_widget_allocation.height)
     XXX_log('test_widget_size = %s' % repr(test_widget_size))
     control_send(test_widget_size)
 
-    # Use a common datastructure for counters to allow multiple tests
-    # to share the same formal name.
-    test_counters = dict((test.formal_name, 0) for test in test_list)
-    for test in test_list:
-        test.count = 0
+    trigger_dict = dict((test.trigger, test) for test in test_list)
 
     refresh_test_status(status_file_path, test_list)
     remaining_tests_queue = [x for x in reversed(test_list)
-                             if test.status != 'passed']
+                             if x.status == _UNTESTED]
+    XXX_log('remaining_tests_queue = %s' %
+            repr([x.label_en for x in remaining_tests_queue]))
+
+    active_test = None
 
     gobject.io_add_watch(sys.stdin, gobject.IO_IN, stdin_callback)
 
     console = console_proc(console_box.get_allocation(), log_file_path)
 
     XXX_log('finished ui setup')
 
     # Test selection is driven either by triggers or by the
     # remaining_tests_queue.  If a trigger was seen, explicitly run
     # the corresponding test.  Otherwise choose the next test from the
     # queue.  Tests are removed from the queue as they are run,
     # regarless of the outcome.  Tests that are interrupted by trigger
     # are treated as having failed.
     #
     # Iterations in the main loop here are driven by data availability
     # on stdin, which is used to communicate with the autotest control
     # program.  On each step through the loop, a trigger is received
     # (possibly None) to indicate how the next test should be selected.
 
     while remaining_tests_queue:
-        trigger = control_recv()
-        XXX_log('ui received trigger (%s)' % trigger)
-        active_test_name, count = select_active_test(
-            test_list, remaining_tests_queue,
-            test_counters, trigger)
-        control_send((active_test_name, count))
-        gtk.main()
-        refresh_test_status(status_file_path, test_list)
+        command, arg = control_recv()
+        XXX_log('ui received command %s(%s)' % (command, arg))
+        if command == 'switch_to':
+            active_test = trigger_dict.get(arg, None)
+            if active_test in remaining_tests_queue:
+                remaining_tests_queue.remove(active_test)
+            set_active_test(active_test)
+        elif command == 'next_test':
+            active_test = remaining_tests_queue.pop()
+            set_active_test(active_test)
+        else:
+            XXX_log('ui command unknown, exiting...')
+            break
+        if active_test.automated_seq:
+            XXX_log('ui starting automated_seq')
+            subtest_queue = [x for x in reversed(active_test.automated_seq)]
+            test_widget_box.remove(notest_label)
+            as_widget = make_automated_seq_widget(active_test)
+            test_widget_box.add(as_widget)
+            window.show_all()
+            command = None
+            while command != 'quit_automated_seq':
+                active_subtest = subtest_queue.pop()
+                active_subtest.label_box.update_status(_ACTIVE)
+                gtk.main()
+                command = control_recv()
+                XXX_log('ui automated_seq step (%s)' % command)
+                refresh_test_status(status_file_path, test_list)
+                test_widget_box.queue_draw()
+            test_widget_box.remove(as_widget)
+            test_widget_box.add(notest_label)
+            window.show_all()
+            XXX_log('ui exiting automated_seq')
+        else:
+            gtk.main()
+            refresh_test_status(status_file_path, test_list)
 
+    # Tell the control process we are done.
     control_send((None, 0))
 
     XXX_log('exiting ui')
 
 if __name__ == '__main__':
 
     # In global scope, get the test_data class description from the
     # control program -- this allows a convenient single point of
     # definition for this class.
     test_data_class_def = control_recv()
     exec(test_data_class_def)
 
     main()