How to use mobly - 10 common examples

args.test_bed)
    # Parse test specifiers if exist.
    tests = None
    if args.tests:
        tests = args.tests
    # Execute the test class with configs.
    ok = True
    for config in test_configs:
        runner = TestRunner(log_dir=config.log_path,
                            testbed_name=config.testbed_name)
        with runner.mobly_logger():
            runner.add_test_class(config, test_class, tests)
            try:
                runner.run()
                ok = runner.results.is_all_pass and ok
            except signals.TestAbortAll:
                pass
            except:
                logging.exception('Exception when executing %s.',
                                  config.testbed_name)
                ok = False
    if not ok:
        sys.exit(1)

def output_path(self):
        utils.create_dir(self._output_dir_path)
        return self._output_dir_path

def run(self):
        """Executes tests.

        This will instantiate controller and test classes, execute tests, and
        print a summary.

        Raises:
            Error: if no tests have previously been added to this runner using
                add_test_class(...).
        """
        if not self._test_run_infos:
            raise Error('No tests to execute.')

        # Ensure the log path exists. Necessary if `run` is used outside of the
        # `mobly_logger` context.
        utils.create_dir(self._root_output_path)

        summary_writer = records.TestSummaryWriter(
            os.path.join(self._root_output_path, records.OUTPUT_FILE_SUMMARY))
        try:
            for test_run_info in self._test_run_infos:
                # Set up the test-specific config
                test_config = test_run_info.config.copy()
                test_config.log_path = self._root_output_path
                test_config.summary_writer = summary_writer
                test_config.test_class_name_suffix = test_run_info.test_class_name_suffix
                try:
                    self._run_test_class(config=test_config,
                                         test_class=test_run_info.test_class,
                                         tests=test_run_info.tests)
                except signals.TestAbortAll as e:
                    logging.warning(

def _update_log_path(self):
        """Updates the logging values with the current timestamp."""
        self._start_time = logger.get_log_file_timestamp()
        self._root_output_path = os.path.join(self._log_dir,
                                              self._testbed_name,
                                              self._start_time)

def mobly_logger(self, alias='latest'):
        """Starts and stops a logging context for a Mobly test run.

        Args:
          alias: optional string, the name of the latest log alias directory to
              create. If a falsy value is specified, then the directory will not
              be created.

        Yields:
            The host file path where the logs for the test run are stored.
        """
        self._update_log_path()
        logger.setup_test_logger(self._root_output_path,
                                 self._testbed_name,
                                 alias=alias)
        try:
            yield self._root_output_path
        finally:
            logger.kill_test_logger(logging.getLogger())

if begin_time is None:
            begin_time = mobly_logger.get_log_file_timestamp()

        new_br = True
        try:
            stdout = self.adb.shell('bugreportz -v').decode('utf-8')
            # This check is necessary for builds before N, where adb shell's ret
            # code and stderr are not propagated properly.
            if 'not found' in stdout:
                new_br = False
        except adb.AdbError:
            new_br = False

        if destination is None:
            destination = os.path.join(self.log_path, 'BugReports')
        br_path = utils.abs_path(destination)
        utils.create_dir(br_path)
        filename = self.generate_filename(prefix, str(begin_time), 'txt')
        if new_br:
            filename = filename.replace('.txt', '.zip')
        full_out_path = os.path.join(br_path, filename)
        # in case device restarted, wait for adb interface to return
        self.wait_for_boot_completion()
        self.log.debug('Start taking bugreport.')
        if new_br:
            out = self.adb.shell('bugreportz', timeout=timeout).decode('utf-8')
            if not out.startswith('OK'):
                raise DeviceError(self, 'Failed to take bugreport: %s' % out)
            br_out_path = out.split(':')[1].strip()
            self.adb.pull([br_out_path, full_out_path])
        else:
            # shell=True as this command redirects the stdout to a local file

begin_time = mobly_logger.get_log_file_timestamp()

        new_br = True
        try:
            stdout = self.adb.shell('bugreportz -v').decode('utf-8')
            # This check is necessary for builds before N, where adb shell's ret
            # code and stderr are not propagated properly.
            if 'not found' in stdout:
                new_br = False
        except adb.AdbError:
            new_br = False

        if destination is None:
            destination = os.path.join(self.log_path, 'BugReports')
        br_path = utils.abs_path(destination)
        utils.create_dir(br_path)
        filename = self.generate_filename(prefix, str(begin_time), 'txt')
        if new_br:
            filename = filename.replace('.txt', '.zip')
        full_out_path = os.path.join(br_path, filename)
        # in case device restarted, wait for adb interface to return
        self.wait_for_boot_completion()
        self.log.debug('Start taking bugreport.')
        if new_br:
            out = self.adb.shell('bugreportz', timeout=timeout).decode('utf-8')
            if not out.startswith('OK'):
                raise DeviceError(self, 'Failed to take bugreport: %s' % out)
            br_out_path = out.split(':')[1].strip()
            self.adb.pull([br_out_path, full_out_path])
        else:
            # shell=True as this command redirects the stdout to a local file
            # using shell redirection.

if any(conditions):
            raise MonsoonError(err_msg)
        hz_str = lines[4].split()[2]
        hz = int(hz_str[:-2])
        voltage_str = lines[2].split()[1]
        voltage = int(voltage_str[:-1])
        lines = lines[6:]
        t = []
        v = []
        for l in lines:
            try:
                timestamp, value = l.split(' ')
                t.append(int(timestamp))
                v.append(float(value))
            except ValueError:
                raise MonsoonError(err_msg)
        return MonsoonData(v, t, hz, voltage)

None if no config existed for this controller and it was not a
            required controller.

        Raises:
            ControllerError:
                * The controller module has already been registered.
                * The actual number of objects instantiated is less than the
                * `min_number`.
                * `required` is True and no corresponding config can be found.
                * Any other error occurred in the registration process.
        """
        verify_controller_module(module)
        # Use the module's name as the ref name
        module_ref_name = module.__name__.split('.')[-1]
        if module_ref_name in self._controller_objects:
            raise signals.ControllerError(
                'Controller module %s has already been registered. It cannot '
                'be registered again.' % module_ref_name)
        # Create controller objects.
        module_config_name = module.MOBLY_CONTROLLER_CONFIG_NAME
        if module_config_name not in self.controller_configs:
            if required:
                raise signals.ControllerError(
                    'No corresponding config found for %s' %
                    module_config_name)
            logging.warning(
                'No corresponding config found for optional controller %s',
                module_config_name)
            return None
        try:
            # Make a deep copy of the config to pass to the controller module,
            # in case the controller module modifies the config internally.

def list_adb_devices_by_usb_id():
    """List the usb id of all android devices connected to the computer that
    are detected by adb.

    Returns:
        A list of strings that are android device usb ids. Empty if there's
        none.
    """
    out = adb.AdbProxy().devices(['-l'])
    clean_lines = new_str(out, 'utf-8').strip().split('\n')
    results = []
    for line in clean_lines:
        tokens = line.strip().split()
        if len(tokens) > 2 and tokens[1] == 'device':
            results.append(tokens[2])
    return results