xref: /aosp_15_r20/external/openthread/tools/harness-automation/autothreadharness/harness_case.py (revision cfb92d1480a9e65faed56933e9c12405f45898b4)

#!/usr/bin/env python
#
# Copyright (c) 2016, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
#    notice, this list of conditions and the following disclaimer in the
#    documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
#    names of its contributors may be used to endorse or promote products
#    derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#

import ConfigParser
import json
import logging
import os
import subprocess
import re
import time
import unittest

from selenium import webdriver
from selenium.webdriver import ActionChains
from selenium.webdriver.support.ui import Select
from selenium.common.exceptions import UnexpectedAlertPresentException
from selenium.common.exceptions import NoSuchElementException
from functools import reduce

from autothreadharness import settings
from autothreadharness.exceptions import FailError, FatalError, GoldenDeviceNotEnoughError
from autothreadharness.harness_controller import HarnessController
from autothreadharness.helpers import HistoryHelper
from autothreadharness.open_thread_controller import OpenThreadController
from autothreadharness.pdu_controller_factory import PduControllerFactory
from autothreadharness.rf_shield_controller import get_rf_shield_controller

logger = logging.getLogger(__name__)

THREAD_CHANNEL_MAX = 26
"""Maximum channel number of thread protocol"""

THREAD_CHANNEL_MIN = 11
"""Minimum channel number of thread protocol"""

DEFAULT_TIMEOUT = 2700
"""Timeout for each test case in seconds"""


def wait_until(what, times=-1):
    """Wait until `what` return True

    Args:
        what (Callable[bool]): Call `wait()` again and again until it returns True
        times (int): Maximum times of trials before giving up

    Returns:
        True if success, False if times threshold reached

    """
    while times:
        logger.info('Waiting times left %d', times)
        try:
            if what() is True:
                return True
        except BaseException:
            logger.exception('Wait failed')
        else:
            logger.warning('Trial[%d] failed', times)
        times -= 1
        time.sleep(1)

    return False


class HarnessCase(unittest.TestCase):
    """This is the case class of all automation test cases.

    All test case classes MUST define properties `role`, `case` and `golden_devices_required`
    """

    channel = settings.THREAD_CHANNEL
    """int: Thread channel.

    Thread channel ranges from 11 to 26.
    """

    ROLE_LEADER = 1
    ROLE_ROUTER = 2
    ROLE_SED = 4
    ROLE_BORDER = 8
    ROLE_REED = 16
    ROLE_ED = 32
    ROLE_COMMISSIONER = 64
    ROLE_JOINER = 128
    ROLE_FED = 512
    ROLE_MED = 1024

    role = None
    """int: role id.

    1
        Leader
    2
        Router
    4
        Sleepy end device
    16
        Router eligible end device
    32
        End device
    64
        Commissioner
    128
        Joiner
    512
        Full end device
    1024
        Minimal end device
    """

    case = None
    """str: Case id, e.g. '6 5 1'.
    """

    golden_devices_required = 0
    """int: Golden devices needed to finish the test
    """

    child_timeout = settings.THREAD_CHILD_TIMEOUT
    """int: Child timeout in seconds
    """

    sed_polling_interval = settings.THREAD_SED_POLLING_INTERVAL
    """int: SED polling interval in seconds
    """

    auto_dut = settings.AUTO_DUT
    """bool: whether use harness auto dut feature"""

    timeout = hasattr(settings, 'TIMEOUT') and settings.TIMEOUT or DEFAULT_TIMEOUT
    """number: timeout in seconds to stop running this test case"""

    started = 0
    """number: test case started timestamp"""

    case_need_shield = False
    """bool: whether needs RF-box"""

    device_order = []
    """list: device drag order in TestHarness TestBed page"""

    def __init__(self, *args, **kwargs):
        self.dut = None
        self._browser = None
        self._hc = None
        self.result_dir = '%s\\%s' % (settings.OUTPUT_PATH, self.__class__.__name__)
        self.history = HistoryHelper()
        self.add_all_devices = False
        self.new_th = False

        harness_info = ConfigParser.ConfigParser()
        harness_info.read('%s\\info.ini' % settings.HARNESS_HOME)
        if harness_info.has_option('Thread_Harness_Info', 'Version') and harness_info.has_option(
                'Thread_Harness_Info', 'Mode'):
            harness_version = harness_info.get('Thread_Harness_Info', 'Version').rsplit(' ', 1)[1]
            harness_mode = harness_info.get('Thread_Harness_Info', 'Mode')

            if harness_mode == 'External' and harness_version > '1.4.0':
                self.new_th = True

            if harness_mode == 'Internal' and harness_version > '49.4':
                self.new_th = True

        super(HarnessCase, self).__init__(*args, **kwargs)

    def _init_devices(self):
        """Reboot all usb devices.

        Note:
            If PDU_CONTROLLER_TYPE is not valid, usb devices is not rebooted.
        """
        if not settings.PDU_CONTROLLER_TYPE:
            if settings.AUTO_DUT:
                return

            for device in settings.GOLDEN_DEVICES:
                port, _ = device
                try:
                    with OpenThreadController(port) as otc:
                        logger.info('Resetting %s', port)
                        otc.reset()
                except BaseException:
                    logger.exception('Failed to reset device %s', port)
                    self.history.mark_bad_golden_device(device)

            return

        tries = 3
        pdu_factory = PduControllerFactory()

        while True:
            try:
                pdu = pdu_factory.create_pdu_controller(settings.PDU_CONTROLLER_TYPE)
                pdu.open(**settings.PDU_CONTROLLER_OPEN_PARAMS)
            except EOFError:
                logger.warning('Failed to connect to telnet')
                tries = tries - 1
                if tries:
                    time.sleep(10)
                    continue
                else:
                    logger.error('Fatal error: cannot connect to apc')
                    raise
            else:
                pdu.reboot(**settings.PDU_CONTROLLER_REBOOT_PARAMS)
                pdu.close()
                break

        time.sleep(len(settings.GOLDEN_DEVICES))

    def _init_harness(self):
        """Restart harness backend service.

        Please start the harness controller before running the cases, otherwise, nothing happens
        """
        self._hc = HarnessController(self.result_dir)
        self._hc.stop()
        time.sleep(1)
        self._hc.start()
        time.sleep(2)

        harness_config = ConfigParser.ConfigParser()
        harness_config.read('%s\\Config\\Configuration.ini' % settings.HARNESS_HOME)
        if harness_config.has_option('THREAD_HARNESS_CONFIG', 'BrowserAutoNavigate') and harness_config.getboolean(
                'THREAD_HARNESS_CONFIG', 'BrowserAutoNavigate'):
            logger.error('BrowserAutoNavigate in Configuration.ini should be False')
            raise FailError('BrowserAutoNavigate in Configuration.ini should be False')
        if settings.MIXED_DEVICE_TYPE:
            if harness_config.has_option('THREAD_HARNESS_CONFIG',
                                         'EnableDeviceSelection') and not harness_config.getboolean(
                                             'THREAD_HARNESS_CONFIG', 'EnableDeviceSelection'):
                logger.error('EnableDeviceSelection in Configuration.ini should be True')
                raise FailError('EnableDeviceSelection in Configuration.ini should be True')

    def _destroy_harness(self):
        """Stop harness backend service

        Stop harness service.
        """
        self._hc.stop()
        time.sleep(2)

    def _init_dut(self):
        """Initialize the DUT.

        DUT will be restarted. and openthread will started.
        """
        if self.auto_dut:
            self.dut = None
            return

        dut_port = settings.DUT_DEVICE[0]
        dut = OpenThreadController(dut_port)
        self.dut = dut

    def _destroy_dut(self):
        self.dut = None

    def _init_browser(self):
        """Open harness web page.

        Open a quiet chrome which:
        1. disables extensions,
        2. ignore certificate errors and
        3. always allow notifications.
        """
        try:
            chrome_options = webdriver.ChromeOptions()
            chrome_options.add_argument('--disable-extensions')
            chrome_options.add_argument('--disable-infobars')
            chrome_options.add_argument('--ignore-certificate-errors')
            chrome_options.add_experimental_option('prefs',
                                                   {'profile.managed_default_content_settings.notifications': 1})

            browser = webdriver.Chrome(chrome_options=chrome_options)
            browser.set_page_load_timeout(20)
            browser.implicitly_wait(1)
            browser.get(settings.HARNESS_URL)
            browser.maximize_window()
            self._browser = browser
            if not wait_until(lambda: 'Thread' in browser.title, 30):
                self.assertIn('Thread', browser.title)
            return True
        except Exception as e:
            logger.info('Init chrome error: {0}'.format(type(e).__name__))
            return False

    def _destroy_browser(self):
        """Close the browser.
        """
        if self._browser:
            self._browser.close()
        self._browser = None

    def _init_rf_shield(self):
        if getattr(settings, 'SHIELD_CONTROLLER_TYPE', None) and getattr(settings, 'SHIELD_CONTROLLER_PARAMS', None):
            self.rf_shield = get_rf_shield_controller(shield_type=settings.SHIELD_CONTROLLER_TYPE,
                                                      params=settings.SHIELD_CONTROLLER_PARAMS)
        else:
            self.rf_shield = None

    def _destroy_rf_shield(self):
        self.rf_shield = None

    def setUp(self):
        """Prepare to run test case.

        Start harness service, init golden devices, reset DUT and open browser.
        """
        if self.__class__ is HarnessCase:
            return

        logger.info('Setting up')
        # clear files

        logger.info('Deleting all .pcapng')
        os.system('del /q "%s\\Captures\\*.pcapng"' % settings.HARNESS_HOME)
        logger.info('Empty files in Logs')
        os.system('del /q "%s\\Logs\\*.*"' % settings.HARNESS_HOME)

        # using temp files to fix excel downloading fail
        if self.new_th:
            logger.info('Empty files in Reports')
            os.system('del /q "%s\\Reports\\*.*"' % settings.HARNESS_HOME)
        else:
            logger.info('Empty files in temps')
            os.system('del /q "%s\\Thread_Harness\\temp\\*.*"' % settings.HARNESS_HOME)

        # create directory
        os.system('mkdir %s' % self.result_dir)
        self._init_harness()
        self._init_devices()
        self._init_dut()
        self._init_rf_shield()

    def tearDown(self):
        """Clean up after each case.

        Stop harness service, close browser and close DUT.
        """
        if self.__class__ is HarnessCase:
            return

        logger.info('Tearing down')
        self._destroy_harness()
        self._destroy_browser()
        self._destroy_dut()
        self._destroy_rf_shield()

    def _setup_page(self):
        """Do sniffer settings and general settings
        """
        if not self.started:
            self.started = time.time()

        # Detect Sniffer
        try:
            dialog = self._browser.find_element_by_id('capture-Setup-modal')
        except BaseException:
            logger.exception('Failed to get dialog.')
        else:
            if dialog and dialog.get_attribute('aria-hidden') == 'false':
                times = 100
                while times:
                    status = dialog.find_element_by_class_name('status-notify').text
                    if 'Searching' in status:
                        logger.info('Still detecting..')
                    elif 'Not' in status:
                        logger.warning('Sniffer device not verified!')
                        button = dialog.find_element_by_id('snifferAutoDetectBtn')
                        button.click()
                    elif 'Verified' in status:
                        logger.info('Verified!')
                        button = dialog.find_element_by_id('saveCaptureSettings')
                        button.click()
                        break
                    else:
                        logger.warning('Unexpected sniffer verification status')

                    times = times - 1
                    time.sleep(1)

                if not times:
                    raise Exception('Unable to detect sniffer device')

        time.sleep(1)

        try:
            skip_button = self._browser.find_element_by_id('SkipPrepareDevice')
            if skip_button.is_enabled():
                skip_button.click()
                time.sleep(1)
        except BaseException:
            logger.info('Still detecting sniffers')

        try:
            next_button = self._browser.find_element_by_id('nextButton')
        except BaseException:
            logger.exception('Failed to finish setup')
            return

        if not next_button.is_enabled():
            logger.info('Harness is still not ready')
            return

        # General Setup
        try:
            if self.child_timeout or self.sed_polling_interval:
                logger.info('finding general Setup button')
                button = self._browser.find_element_by_id('general-Setup')
                button.click()
                time.sleep(2)

                dialog = self._browser.find_element_by_id('general-Setup-modal')
                if dialog.get_attribute('aria-hidden') != 'false':
                    raise Exception('Missing General Setup dialog')

                field = dialog.find_element_by_id('inp_general_child_update_wait_time')
                field.clear()
                if self.child_timeout:
                    field.send_keys(str(self.child_timeout))

                field = dialog.find_element_by_id('inp_general_sed_polling_rate')
                field.clear()
                if self.sed_polling_interval:
                    field.send_keys(str(self.sed_polling_interval))

                button = dialog.find_element_by_id('saveGeneralSettings')
                button.click()
                time.sleep(1)

        except BaseException:
            logger.info('general setup exception')
            logger.exception('Failed to do general setup')
            return

        # Finish this page
        next_button.click()
        time.sleep(1)

    def _connect_devices(self):
        connect_all = self._browser.find_element_by_link_text('Connect All')
        connect_all.click()

    def _add_device(self, port, device_type_id):
        browser = self._browser
        test_bed = browser.find_element_by_id('test-bed')
        device = browser.find_element_by_id(device_type_id)
        # drag
        action_chains = ActionChains(browser)
        action_chains.click_and_hold(device)
        action_chains.move_to_element(test_bed).perform()
        time.sleep(1)

        # drop
        drop_hw = browser.find_element_by_class_name('drop-hw')
        action_chains = ActionChains(browser)
        action_chains.move_to_element(drop_hw)
        action_chains.release(drop_hw).perform()

        time.sleep(0.5)
        selected_hw = browser.find_element_by_class_name('selected-hw')
        form_inputs = selected_hw.find_elements_by_tag_name('input')
        form_port = form_inputs[0]
        form_port.clear()
        form_port.send_keys(port)

    def _test_bed(self):
        """Set up the test bed.

        Connect number of golden devices required by each case.
        """
        browser = self._browser
        test_bed = browser.find_element_by_id('test-bed')
        time.sleep(3)
        selected_hw_set = test_bed.find_elements_by_class_name('selected-hw')
        selected_hw_num = len(selected_hw_set)

        while selected_hw_num:
            remove_button = selected_hw_set[selected_hw_num - 1].find_element_by_class_name('removeSelectedDevice')
            remove_button.click()
            selected_hw_num = selected_hw_num - 1

        devices = [
            device for device in settings.GOLDEN_DEVICES if not self.history.is_bad_golden_device(device[0]) and
            not (settings.DUT_DEVICE and device[0] == settings.DUT_DEVICE[0])
        ]
        logger.info('Available golden devices: %s', json.dumps(devices, indent=2))

        shield_devices = [
            shield_device for shield_device in settings.SHIELD_GOLDEN_DEVICES
            if not self.history.is_bad_golden_device(shield_device[0]) and
            not (settings.DUT2_DEVICE and shield_device[0] == settings.DUT2_DEVICE[0])
        ]
        logger.info('Available shield golden devices: %s', json.dumps(shield_devices, indent=2))
        golden_devices_required = self.golden_devices_required

        dut_device = ()
        if settings.DUT_DEVICE:
            dut_device = settings.DUT_DEVICE
        """check if test case needs to use RF-shield box and its device order in Testbed page
        Two parameters case_need_shield & device_order should be set in the case script
        according to the requires: https://openthread.io/certification/test-cases#rf_shielding
        Example:
         In case script leader_9_2_9.py:
          case_need_shield = True
          device_order = [('Router_2', False), ('Commissioner', True), ('Router_1', False), ('DUT', True)]
         On the TestBed page of the Test Harness, the device sort order for Leader_9_2_9
           should be like:
             Router_2
             Commissioner
             Router_1
             DUT
           The ('Commissioner', True) and ('DUT', True) indicate Commissioner device and DUT2 device should
           be in the RF-box and choose from SHIELD_GOLDEN_DEVICES and DUT2_DEVICE. Otherwise ('DUT', False) means
           DUT device is not in RF-box and use DUT_DEVICE. The other roles devices with False should be selected
           from GOLDEN_DEVICES.

         In case script med_6_3_2.py:
         case_need_shield = True
         device_order = [] # or not defined
         means no device drag order. DUT2_DEVICE should be applied as DUT and the other golden devices
         are from GOLDEN_DEVICES.
        """
        if self.case_need_shield:
            if not settings.DUT2_DEVICE:
                logger.info('Must set DUT2_DEVICE')
                raise FailError('DUT2_DEVICE must be set in settings.py')
            if isinstance(self.device_order, list) and self.device_order:
                logger.info('case %s devices ordered by %s ', self.case, self.device_order)
            else:
                logger.info('case %s uses %s as DUT', self.case, settings.DUT2_DEVICE)

        # for test bed with multi-vendor devices
        if settings.MIXED_DEVICE_TYPE:
            topo_file = settings.HARNESS_HOME + "\\Thread_Harness\\TestScripts\\TopologyConfig.txt"
            try:
                f_topo = open(topo_file, 'r')
            except IOError:
                logger.info('%s can NOT be found', topo_file)
                raise GoldenDeviceNotEnoughError()
            topo_mixed_devices = []
            try:
                while True:
                    topo_line = f_topo.readline().strip()
                    if re.match(r'#.*', topo_line):
                        continue
                    match_line = re.match(r'(.*)-(.*)', topo_line, re.M | re.I)
                    if not match_line:
                        continue
                    case_id = match_line.group(1)

                    if re.sub(r'\.', ' ', case_id) == self.case:
                        logger.info('Get line by case %s: %s', case_id, topo_line)
                        topo_device_list = re.split(',', match_line.group(2))
                        for i in range(len(topo_device_list)):
                            topo_device = re.split(':', topo_device_list[i])
                            topo_mixed_devices.append(tuple(topo_device))
                        break
                    else:
                        continue
            except Exception as e:
                logger.info('Get devices from topology config file error: %s', e)
                raise GoldenDeviceNotEnoughError()
            logger.info('Golden devices in topology config file for case %s: %s', case_id, topo_mixed_devices)
            f_topo.close()
            golden_device_candidates = []
            missing_golden_devices = topo_mixed_devices[:]

            # mapping topology config devices with golden devices by device order
            if self.case_need_shield and self.device_order:
                matched_dut = False
                for device_order_item in self.device_order:
                    matched = False
                    for mixed_device_item in topo_mixed_devices:
                        # mapping device in device_order which needs to be shielded
                        if device_order_item[1]:
                            if 'DUT' in device_order_item[0]:
                                golden_device_candidates.append(settings.DUT2_DEVICE)
                                dut_device = settings.DUT2_DEVICE
                                matched_dut = True
                                matched = True
                                break
                            for device_item in shield_devices:
                                if (device_order_item[0] == mixed_device_item[0] and
                                        mixed_device_item[1] == device_item[1]):
                                    golden_device_candidates.append(device_item)
                                    shield_devices.remove(device_item)
                                    matched = True
                                    break
                        # mapping device in device_order which does not need to be shielded
                        else:
                            if 'DUT' in device_order_item[0]:
                                golden_device_candidates.append(settings.DUT_DEVICE)
                                matched_dut = True
                                matched = True
                                break
                            for device_item in devices:
                                if (device_order_item[0] == mixed_device_item[0] and
                                        mixed_device_item[1] == device_item[1]):
                                    golden_device_candidates.append(device_item)
                                    devices.remove(device_item)
                                    matched = True
                                    break
                    if not matched:
                        logger.info('Golden device not enough in : no %s', device_order_item)
                        raise GoldenDeviceNotEnoughError()
                if not matched_dut:
                    raise FailError('Failed to find DUT in device_order')
                devices = golden_device_candidates
                self.add_all_devices = True
            else:
                for mixed_device_item in topo_mixed_devices:
                    for device_item in devices:
                        if mixed_device_item[1] == device_item[1]:
                            golden_device_candidates.append(device_item)
                            devices.remove(device_item)
                            missing_golden_devices.remove(mixed_device_item)
                            break
                logger.info('Golden devices in topology config file mapped in settings : %s', golden_device_candidates)
                if len(topo_mixed_devices) != len(golden_device_candidates):
                    device_dict = dict()
                    for missing_device in missing_golden_devices:
                        if missing_device[1] in device_dict:
                            device_dict[missing_device[1]] += 1
                        else:
                            device_dict[missing_device[1]] = 1
                    logger.info('Missing Devices: %s', device_dict)
                    raise GoldenDeviceNotEnoughError()
                else:
                    devices = golden_device_candidates
                    golden_devices_required = len(devices)
                    logger.info('All case-needed golden devices: %s', json.dumps(devices, indent=2))
        # for test bed with single vendor devices
        else:
            golden_device_candidates = []
            if self.case_need_shield and self.device_order:
                matched_dut = False
                for device_order_item in self.device_order:
                    matched = False
                    # choose device which needs to be shielded
                    if device_order_item[1]:
                        if 'DUT' in device_order_item[0]:
                            golden_device_candidates.append(settings.DUT2_DEVICE)
                            dut_device = settings.DUT2_DEVICE
                            matched_dut = True
                            matched = True
                        else:
                            for device_item in shield_devices:
                                golden_device_candidates.append(device_item)
                                shield_devices.remove(device_item)
                                matched = True
                                break
                    # choose device which does not need to be shielded
                    else:
                        if 'DUT' in device_order_item[0]:
                            golden_device_candidates.append(settings.DUT_DEVICE)
                            matched_dut = True
                            matched = True
                        else:
                            for device_item in devices:
                                golden_device_candidates.append(device_item)
                                devices.remove(device_item)
                                matched = True
                                break
                    if not matched:
                        logger.info('Golden device not enough in : no %s', device_order_item)
                        raise GoldenDeviceNotEnoughError()
                if not matched_dut:
                    raise FailError('Failed to find DUT in device_order')
                devices = golden_device_candidates
                self.add_all_devices = True

        if self.auto_dut and not settings.DUT_DEVICE:
            if settings.MIXED_DEVICE_TYPE:
                logger.info('Must set DUT_DEVICE')
                raise FailError('DUT_DEVICE must be set for mixed testbed')
            golden_devices_required += 1

        if len(devices) < golden_devices_required:
            raise GoldenDeviceNotEnoughError()

        # add golden devices
        number_of_devices_to_add = len(devices) if self.add_all_devices else golden_devices_required
        for i in range(number_of_devices_to_add):
            self._add_device(*devices.pop())

        # add DUT
        if self.case_need_shield:
            if not self.device_order:
                self._add_device(*settings.DUT2_DEVICE)
        else:
            if settings.DUT_DEVICE:
                self._add_device(*settings.DUT_DEVICE)

        # enable AUTO DUT
        if self.auto_dut:
            checkbox_auto_dut = browser.find_element_by_id('EnableAutoDutSelection')
            if not checkbox_auto_dut.is_selected():
                checkbox_auto_dut.click()
                time.sleep(1)

            if settings.DUT_DEVICE:
                radio_auto_dut = browser.find_element_by_class_name('AutoDUT_RadBtns')
                if not radio_auto_dut.is_selected() and not self.device_order:
                    radio_auto_dut.click()

                if self.device_order:
                    selected_hw_set = test_bed.find_elements_by_class_name('selected-hw')
                    for selected_hw in selected_hw_set:
                        form_inputs = selected_hw.find_elements_by_tag_name('input')
                        form_port = form_inputs[0]
                        port = form_port.get_attribute('value').encode('utf8')
                        if port == dut_device[0]:
                            radio_auto_dut = selected_hw.find_element_by_class_name('AutoDUT_RadBtns')
                            if not radio_auto_dut.is_selected():
                                radio_auto_dut.click()

        while True:
            try:
                self._connect_devices()
                button_next = browser.find_element_by_id('nextBtn')
                if not wait_until(
                        lambda: 'disabled' not in button_next.get_attribute('class'),
                        times=(30 + 4 * number_of_devices_to_add),
                ):
                    bad_ones = []
                    selected_hw_set = test_bed.find_elements_by_class_name('selected-hw')
                    for selected_hw in selected_hw_set:
                        form_inputs = selected_hw.find_elements_by_tag_name('input')
                        form_port = form_inputs[0]
                        if form_port.is_enabled():
                            bad_ones.append(selected_hw)

                    for selected_hw in bad_ones:
                        form_inputs = selected_hw.find_elements_by_tag_name('input')
                        form_port = form_inputs[0]
                        port = form_port.get_attribute('value').encode('utf8')
                        if port == dut_device[0]:
                            if settings.PDU_CONTROLLER_TYPE is None:
                                # connection error cannot recover without power
                                # cycling
                                raise FatalError('Failed to connect to DUT')
                            else:
                                raise FailError('Failed to connect to DUT')

                        if settings.PDU_CONTROLLER_TYPE is None:
                            # port cannot recover without power cycling
                            self.history.mark_bad_golden_device(port)

                        # remove the bad one
                        selected_hw.find_element_by_class_name('removeSelectedDevice').click()
                        time.sleep(0.1)

                        if len(devices):
                            self._add_device(*devices.pop())
                        else:
                            devices = None

                    if devices is None:
                        logger.warning('Golden devices not enough')
                        raise GoldenDeviceNotEnoughError()
                    else:
                        logger.info('Try again with new golden devices')
                        continue

                if self.auto_dut and not settings.DUT_DEVICE:
                    radio_auto_dut = browser.find_element_by_class_name('AutoDUT_RadBtns')
                    if not radio_auto_dut.is_selected():
                        radio_auto_dut.click()

                    time.sleep(5)

                button_next.click()
                if not wait_until(lambda: self._browser.current_url.endswith('TestExecution.html'), 20):
                    raise Exception('Failed to load TestExecution page')
            except FailError:
                raise
            except BaseException:
                logger.exception('Unexpected error')
            else:
                break

    def _select_case(self, role, case):
        """Select the test case.
        """
        # select the case
        elem = Select(self._browser.find_element_by_id('select-dut'))
        elem.select_by_value(str(role))
        time.sleep(1)

        checkbox = None
        wait_until(lambda: self._browser.find_elements_by_css_selector('.tree-node .tree-title') and True)
        elems = self._browser.find_elements_by_css_selector('.tree-node .tree-title')
        finder = re.compile(r'.*\b' + case + r'\b')
        finder_dotted = re.compile(r'.*\b' + case.replace(' ', r'\.') + r'\b')
        for elem in elems:
            # elem.txt might be null when the required reference devices could
            # not be met (either due to the quantity or the type) for specific test.
            # perform() will throw exceptions if elem.text is null since Chrome
            # and chromedriver 80. If elem is not shown in current case list
            # window, move_to_element() will not work either.
            if not elem.text:
                continue
            # execute a javascript to scroll the window to the elem
            self._browser.execute_script('arguments[0].scrollIntoView();', elem)
            action_chains = ActionChains(self._browser)
            action_chains.move_to_element(elem)
            action_chains.perform()
            logger.debug(elem.text)
            if finder.match(elem.text) or finder_dotted.match(elem.text):
                parent = elem.find_element_by_xpath('..')
                checkbox = parent.find_element_by_class_name('tree-checkbox')
                break

        if not checkbox:
            time.sleep(5)
            raise Exception('Failed to find the case')

        self._browser.execute_script("$('.overview').css('left', '0')")
        checkbox.click()
        time.sleep(1)

        elem = self._browser.find_element_by_id('runTest')
        elem.click()
        if not wait_until(lambda: self._browser.find_element_by_id('stopTest') and True, 10):
            raise Exception('Failed to start test case')

    def _collect_result(self):
        """Collect test result.

        Copy PDF and pcap file to result directory
        """

        if self.new_th:
            os.system('copy "%s\\Reports\\*.*" "%s"' % (settings.HARNESS_HOME, self.result_dir))
        else:
            os.system('copy "%s\\Thread_Harness\\temp\\*.*" "%s"' % (settings.HARNESS_HOME, self.result_dir))

        os.system('copy "%s\\Captures\\*.pcapng" %s\\' % (settings.HARNESS_HOME, self.result_dir))

    def _wait_dialog(self):
        """Wait for dialogs and handle them until done.
        """
        logger.debug('waiting for dialog')
        done = False
        error = False

        logger.info('self timeout %d', self.timeout)
        while not done and self.timeout:
            try:
                dialog = self._browser.find_element_by_id('RemoteConfirm')
            except BaseException:
                logger.exception('Failed to get dialog.')
            else:
                if dialog and dialog.get_attribute('aria-hidden') == 'false':
                    title = dialog.find_element_by_class_name('modal-title').text
                    time.sleep(1)
                    logger.info('Handling dialog[%s]', title)

                    try:
                        done = self._handle_dialog(dialog, title)
                    except BaseException:
                        logger.exception('Error handling dialog: %s', title)
                        error = True

                    if done is None:
                        raise FailError('Unexpected dialog occurred')

                    dialog.find_element_by_id('ConfirmOk').click()

            time.sleep(1)

            try:
                stop_button = self._browser.find_element_by_id('stopTest')
                if done:
                    stop_button.click()
                    # wait for stop procedure end
                    time.sleep(10)
            except NoSuchElementException:
                logger.info('Test stopped')
                time.sleep(5)
                done = True

            self.timeout -= 1

            # check if already ended capture
            if self.timeout % 10 == 0:
                lines = self._hc.tail()
                if 'SUCCESS: The process "dumpcap.exe" with PID ' in lines:
                    logger.info('Tshark should be ended now, lets wait at most 30 seconds.')
                    if not wait_until(lambda: 'tshark.exe' not in subprocess.check_output('tasklist'), 30):
                        res = subprocess.check_output('taskkill /t /f /im tshark.exe',
                                                      stderr=subprocess.STDOUT,
                                                      shell=True)
                        logger.info(res)

        # Wait until case really stopped
        wait_until(lambda: self._browser.find_element_by_id('runTest') and True, 30)

        if error:
            raise FailError('Fail for previous exceptions')

    def _handle_dialog(self, dialog, title):
        """Handle a dialog.

        Returns:
            bool True if no more dialogs expected,
                 False if more dialogs needed, and
                 None if not handled
        """
        done = self.on_dialog(dialog, title)
        if isinstance(done, bool):
            return done

        if title.startswith('Start DUT'):
            body = dialog.find_element_by_id('cnfrmMsg').text
            if 'Sleepy End Device' in body:
                self.dut.mode = 's'
                self.dut.child_timeout = self.child_timeout
            elif 'End Device' in body:
                self.dut.mode = 'rn'
                self.dut.child_timeout = self.child_timeout
            else:
                self.dut.mode = 'rdn'

            if 'at channel' in body:
                self.channel = int(body.split(':')[1])

            self.dut.channel = self.channel
            self.dut.panid = settings.THREAD_PANID
            self.dut.networkname = settings.THREAD_NETWORKNAME
            self.dut.extpanid = settings.THREAD_EXTPANID
            self.dut.start()

        elif title.startswith('MAC Address Required') or title.startswith('DUT Random Extended MAC Address Required'):
            mac = self.dut.mac
            inp = dialog.find_element_by_id('cnfrmInpText')
            inp.clear()
            inp.send_keys('0x%s' % mac)

        elif title.startswith('LL64 Address'):
            ll64 = None
            for addr in self.dut.addrs:
                addr = addr.lower()
                if addr.startswith('fe80') and not re.match('.+ff:fe00:[0-9a-f]{0,4}$', addr):
                    ll64 = addr
                    break

            if not ll64:
                raise FailError('No link local address found')

            logger.info('Link local address is %s', ll64)
            inp = dialog.find_element_by_id('cnfrmInpText')
            inp.clear()
            inp.send_keys(ll64)

        elif title.startswith('Enter Channel'):
            self.dut.channel = self.channel
            inp = dialog.find_element_by_id('cnfrmInpText')
            inp.clear()
            inp.send_keys(str(self.dut.channel))

        elif title.startswith('User Action Needed'):
            body = dialog.find_element_by_id('cnfrmMsg').text
            if body.startswith('Power Down the DUT'):
                self.dut.stop()
            return True

        elif title.startswith('Short Address'):
            short_addr = '0x%s' % self.dut.short_addr
            inp = dialog.find_element_by_id('cnfrmInpText')
            inp.clear()
            inp.send_keys(short_addr)

        elif title.startswith('ML64 Address'):
            ml64 = None
            for addr in self.dut.addrs:
                if addr.startswith('fd') and not re.match('.+ff:fe00:[0-9a-f]{0,4}$', addr):
                    ml64 = addr
                    break

            if not ml64:
                raise Exception('No mesh local address found')

            logger.info('Mesh local address is %s', ml64)
            inp = dialog.find_element_by_id('cnfrmInpText')
            inp.clear()
            inp.send_keys(ml64)

        elif title.startswith('Shield Devices') or title.startswith('Shield DUT'):
            time.sleep(2)
            if self.rf_shield:
                logger.info('Shielding devices')
                with self.rf_shield:
                    self.rf_shield.shield()
            elif self.dut and settings.SHIELD_SIMULATION:
                self.dut.channel = (self.channel == THREAD_CHANNEL_MAX and THREAD_CHANNEL_MIN) or (self.channel + 1)
            else:
                input('Shield DUT and press enter to continue..')

        elif title.startswith('Unshield Devices') or title.startswith('Bring DUT back to network'):
            time.sleep(5)
            if self.rf_shield:
                logger.info('Unshielding devices')
                with self.rf_shield:
                    self.rf_shield.unshield()
            elif self.dut and settings.SHIELD_SIMULATION:
                self.dut.channel = self.channel
            else:
                input('Bring DUT and press enter to continue..')

        elif title.startswith('Configure Prefix on DUT'):
            body = dialog.find_element_by_id('cnfrmMsg').text
            body = body.split(': ')[1]
            params = reduce(
                lambda params, param: params.update(((param[0].strip(' '), param[1]),)) or params,
                [it.split('=') for it in body.split(', ')],
                {},
            )
            prefix = params['P_Prefix'].strip('\0\r\n\t ')
            flags = []
            if params.get('P_slaac_preferred', 0) == '1':
                flags.append('p')
            flags.append('ao')
            if params.get('P_stable', 0) == '1':
                flags.append('s')
            if params.get('P_default', 0) == '1':
                flags.append('r')
            prf = 'high'
            self.dut.add_prefix(prefix, ''.join(flags), prf)

        return False

    def test(self):
        """This method will only start test case in child class"""
        if self.__class__ is HarnessCase:
            logger.warning('Skip this harness itself')
            return

        logger.info('Testing role[%d] case[%s]', self.role, self.case)

        init_browser_times = 5
        while True:
            if self._init_browser():
                break
            elif init_browser_times > 0:
                init_browser_times -= 1
                self._destroy_browser()
            else:
                raise SystemExit()

        try:
            # prepare test case
            while True:
                url = self._browser.current_url
                if url.endswith('SetupPage.html'):
                    self._setup_page()
                elif url.endswith('TestBed.html'):
                    self._test_bed()
                elif url.endswith('TestExecution.html'):
                    logger.info('Ready to handle dialogs')
                    break
                time.sleep(2)
        except UnexpectedAlertPresentException:
            logger.exception('Failed to connect to harness server')
            raise SystemExit()
        except FatalError:
            logger.exception('Test stopped for fatal error')
            raise SystemExit()
        except FailError:
            logger.exception('Test failed')
            raise
        except BaseException:
            logger.exception('Something wrong')

        self._select_case(self.role, self.case)

        logger.info('start to wait test process end')
        self._wait_dialog()

        try:
            self._collect_result()
        except BaseException:
            logger.exception('Failed to collect results')
            raise

        # get case result
        status = self._browser.find_element_by_class_name('title-test').get_attribute('innerText')
        logger.info(status)
        success = 'Pass' in status
        self.assertTrue(success)