Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use log_console_output method in tempest

Best Python code snippet using tempest_python

uboot_expect.py

Source:uboot_expect.py

1#!/usr/bin/env python2# ==========================================================================3#4# Copyright (C) 2014 RidgeRun, LLC (http://www.ridgerun.com)5#6# Author: Jose Pablo Carballo <jose.carballo@ridgerun.com>7#8# This program is free software; you can redistribute it and/or modify9# it under the terms of the GNU General Public License version 2 as10# published by the Free Software Foundation.11#12# Communication with Uboot using pexpect to support the installer.13#14# ==========================================================================15import time16import re17import pexpect18import openfd.utils as utils19from openfd.utils.hexutils import to_hex20# ==========================================================================21# Constants22# ==========================================================================23CTRL_C = '\x03'24# Serial settings25DEFAULT_PORT = '/dev/ttyS0'26DEFAULT_BAUDRATE = 11520027DEFAULT_READ_TIMEOUT = 2 # seconds28# Uboot communication timeouts (seconds)29DEFAULT_UBOOT_TIMEOUT = 530# ==========================================================================31# Public Classes32# ==========================================================================33class UbootTimeoutException(Exception):34    """Uboot timeouts give an exception."""35class UbootExpect(object):36    """37    Class that abstracts the communication with uboot using a console.38    Based on pexpect.39    """40    41    def __init__(self, dryrun=False):42        self._l = utils.logger.get_global_logger()43        self._e = utils.executer.get_global_executer()44        self._open_cmd = None45        self._l_console = None46        self._prompt = ''47        self._log_prefix = '  Uboot'48        self._child = None49        self._dryrun = dryrun50        self._e.dryrun = dryrun51    @classmethod52    def comm_error_msg(cls, cmd):53        """54        Standard error message to report a failure communicating with uboot55        in the given port.56             Default: "  Uboot"57        58        :param cmd: The command used to communicate with u-boot.59        :return: A string with the standard message.60        """61        62        return ("Failed to handshake with uboot using '%s'.\n"63               "Be sure uboot is active and you have terminal "64               "programs like minicom or termnet closed." % cmd)65    66    def __set_dryrun(self, dryrun):67        self._dryrun = dryrun68        self._e.dryrun = dryrun69    70    def __get_dryrun(self):71        return self._dryrun72    73    dryrun = property(__get_dryrun, __set_dryrun,74                     doc="""Enable dryrun mode. System and uboot commands will75                     be logged, but not executed.""")76    def __set_log_prefix(self, prefix):77        self._log_prefix = prefix78    79    def __get_log_prefix(self):80        return self._log_prefix81    82    log_prefix = property(__get_log_prefix, __set_log_prefix,83             doc="""String to prefix log messages for commands.84             Default: "  Uboot" """)85    def __set_console_logger(self, logger):86        self._l_console = logger87    88    def __get_console_logger(self):89        return self._l_console90    91    console_logger = property(__get_console_logger, __set_console_logger,92             doc=""":class:`Logger` instance to log the console's output.93             Output will be logged on DEBUG level.""")94    def _check_is_alive(self):95        if not self._dryrun and not self._child.isalive():96            self._l.error('No child program.')97            return False98        else:99            return True100    def open_comm(self, cmd, timeout=DEFAULT_READ_TIMEOUT):101        """102        Opens the communication with the console where uboot is active.103        It's a good practice to call :func:`sync` after opening the port.104        105        :param cmd: Command to spawn106        :param timeout: Set a read timeout value107        :return: Returns true on success; false otherwise.108        :exception ExceptionPexpect: On error while spawning the child program.109        """110        111        self._l.debug("Spawning child program '%s'" % cmd)112        self._child = pexpect.spawn(cmd)113        self._child.timeout = DEFAULT_READ_TIMEOUT114        self._open_cmd = cmd115        return True116        117    def close_comm(self):118        """119        Closes the communication with the Serial port immediately.120        """121        122        if self._child:123            self._child.close(force=True)124            self._child = None125    126    def sync(self):127        """128        Synchronizes with uboot. If successful, uboot's prompt will be ready 129        to receive commands after this call.130        131        :return: Returns true on success; false otherwise.132        """133        134        self._l.debug("Synchronizing with uboot")135        136        if self._check_is_alive() is False: return False137        138        # Use an echo command to sync139        err_msg = UbootExpect.comm_error_msg(self._open_cmd)140        try:141            self.cmd('echo sync', prompt_timeout=False)142        except UbootTimeoutException:143            self._l.error(err_msg)144            return False145        146        if not self._dryrun:147            try:148                self._child.expect('sync', timeout=1)149            except (pexpect.TIMEOUT, pexpect.EOF):150                self._l.error(err_msg)151                return False152        153        # Identify the prompt in the following line154        155            try:156                self._child.expect('(?P<prompt>.*)[ |$|#]')157            except pexpect.EOF:158                self._l.error(err_msg)159                return False160            except pexpect.TIMEOUT:161                self._l.error("Couldn't identify the uboot prompt.")162                return False163            self._prompt = self._child.match.group('prompt').strip()164            self._l.debug('Uboot prompt: %s' % self._prompt)165            166        return True167    168    def cmd(self, cmd, echo_timeout=DEFAULT_UBOOT_TIMEOUT,169                  prompt_timeout=DEFAULT_UBOOT_TIMEOUT):170        """171        Sends a command to uboot.172        173        :param cmd: Command.174        :param echo_timeout: Timeout to wait for the command to be echoed. Set175            to None to a_l_serialvoid waiting for the echo.176        :type echo_timeout: integer or none177        :param prompt_timeout: Timeout to wait for the prompt after sending178            the command. Set to None to avoid waiting for the prompt.179        :type prompt_timeout: integer or none180        :exception UbootTimeoutException: When a timeout is reached.181        """182        183        if cmd == CTRL_C:184            self._l.info("%s <= '<ctrl_c>'" % self._log_prefix)185        else:186            self._l.info("%s <= '%s'" % (self._log_prefix, cmd.strip()))187        188        if not self._dryrun:189        190            self._child.send('%s\n' % cmd)191            time.sleep(0.1)192            193            # Wait for the echo194            if echo_timeout:195                try:196                    self._child.expect(cmd.strip(), timeout=echo_timeout)197                except (pexpect.TIMEOUT, pexpect.EOF):198                    msg = ("Uboot didn't echo the '%s' command, maybe it "199                        "froze. " % cmd.strip())200                    raise UbootTimeoutException(msg)201            202            # Wait for the prompt203            if self._prompt and prompt_timeout:204                try:205                    self._child.expect(self._prompt, timeout=prompt_timeout)206                except (pexpect.TIMEOUT, pexpect.EOF):207                    msg = ("Didn't get the uboot prompt back after "208                               "executing the '%s' command." % cmd.strip())209                    raise UbootTimeoutException(msg)210    def expect(self, response, timeout=DEFAULT_UBOOT_TIMEOUT,211               log_console_output=False):212        """213        Expects a response from Uboot for no more than timeout seconds.214        The lines read from the console will be stripped before being215        compared with response.216        217        :param response: A string to expect in the console.218        :param timeout: Timeout in seconds to wait for the response.219        :param log_console_output: Logs (debug level INFO) the output from220            the console while expecting the response.221        :return: Returns a tuple with two items. The first item is true if the222            response was found; false otherwise. The second item is the223            complete line where the response was found, or the last line read224            if the response wasn't found and the timeout reached. The line is225            returned stripped.226        """227        228        if self._check_is_alive() is False: return False, ''229        if self._dryrun: return True, ''230        231        found = False232        line = ''233        start_time = time.time()234        235        while not found and (time.time() - start_time) < timeout:236            try:237                line = self._child.readline().strip(' \r\n')238                if self._l_console:239                    msg = "%s => '%s'" % (self._log_prefix, line)240                    if log_console_output:241                        self._l_console.info(msg)242                    else:243                        self._l_console.debug(msg)244            except pexpect.EOF as e:245                self._l.error(e)246                return False, ''247            except pexpect.TIMEOUT:248                pass249            if response in line:250                found = True251            252        return found, line253    def save_env(self):254        """255        Saves the uboot environment to persistent storage.256        257        :exception UbootTimeoutException: When a timeout is reached.258        """259        260        self.cmd('saveenv')261    def set_env(self, variable, value):262        """263        Sets an uboot env variable.264        265        :exception UbootTimeoutException: When a timeout is reached.266        """267        268        value = value.strip()269        if ' ' in value or ';' in value:270            self.cmd("setenv %s '%s'" % (variable, value), echo_timeout=None)271        else:272            if value.startswith('0x') and value.endswith('L'):273                self.cmd("setenv %s %s" % (variable, to_hex(value)),274                                            echo_timeout=None)275            else:276                self.cmd("setenv %s %s" % (variable, value), echo_timeout=None)277    278    def get_env(self, variable):279        """280        Obtains a string with the value of the uboot env variable if found;281        an empty string otherwise.282        283        :exception UbootTimeoutException: When a timeout is reached.284        """285        286        value=''287        self.cmd('printenv %s' % variable, prompt_timeout=None)288        found, line = self.expect('%s=' % variable,289                                    timeout=DEFAULT_READ_TIMEOUT)290        if found:291            m = re.match('.*?=(?P<value>.*)', line)292            if m:293                value = m.group('value').strip()294        return value295    def cancel_cmd(self):296        """297        Cancels the command being executed by uboot (equivalent to CTRL+C).298        299        :exception UbootTimeoutException: When a timeout is reached.300        """301        ...

display_controller.py

Source:display_controller.py

...131        log.exception("Error fetching notification status")132        response = {"success": False, "error": str(ex), "err_type": type(ex).__name__}133    finally:134        return jsonify(response)135def log_console_output(min_timestamp: int = None):136    try:137        console_log = driver.get_log("browser")138        for entry in console_log:139            if entry["source"] == "console-api":140                level = logging.getLevelName(entry["level"])141                entry_timestamp = entry["timestamp"]142                if min_timestamp is None or entry_timestamp >= min_timestamp:143                    log.log(level, f"CONSOLE: {entry['message']}")144    except Exception as ex:145        log.exception("Console Log Error")146    finally:147        t = Timer(interval=30, function=log_console_output, args=(time.time(),))148        t.start()149def main():150    try:151        exit_code = 0152        subprocess.run("sudo systemctl restart hyperion", shell=True)153        global driver154        load_dotenv(dotenv_path=constants.env_dir / "datanode.env")155        options = Options()156        options.add_argument("--kiosk")157        options.add_argument("--noerrdialogs")158        options.add_argument("--disable-infobars")159        options.add_argument("--incognito")160        options.add_experimental_option("excludeSwitches", ["enable-automation"])161        options.add_experimental_option("useAutomationExtension", False)162        desired_capabilities = DesiredCapabilities.CHROME163        desired_capabilities["goog:loggingPrefs"] = {"browser": "ALL"}164        driver = webdriver.Chrome(options=options, desired_capabilities=desired_capabilities)165        global wait166        wait = WebDriverWait(driver, 20)167        loaded = False168        tries = 0169        while not loaded and tries < 10:170            try:171                log.info("Connecting to pihome webserver.")172                driver.get(constants.piframe_url)173                log_console_output()174                loaded = True175            except Exception as ex:176                if tries < 4:177                    log.error(178                        f"Could not connect to pihome webserver. Retrying ({tries+1} tries)..."179                    )180                    time.sleep(10)181                else:182                    log.exception(f"Site unreachable")183            finally:184                tries += 1185        global notify_shown186        notify_shown = False187        pyautogui.FAILSAFE = False...

parent_class.py

Source:parent_class.py

1""" Splunk parent class containing generic functions for the children inheritance.2__classes__:3    MissingConfigError4    SplunkParentClass5@author: dcsteve246__python_version__ = Py2/Py37__os__ =  All8__updated__ = '2021-09-05'9"""10from py_splunk.globals import FORWARDER_SPLUNK_COMMAND, FORWARDER_LOGGER_NAME, \11    FORWARDER_PYTHON_LOGGER_LOCATION, SPLUNK_COMMAND, SPLUNK_LOGGER_NAME, \12    SPLUNK_PYTHON_LOGGER_LOCATION, SPLUNK_USER, UNIVERSAL_FORWARDER13from py_utilities.configuration_operations import get_config_object, DEFAULT_APP_FILE_LOCATION, \14    MissingConfigError, MissingSettingError, read_config15from py_utilities.dynamic_logger import create_logger, log_message16from py_utilities.remote_operations import remote_operations17# TODO: Tie in authentication or better use cases than only those mentioned in the18#    restart_splunk_service19# TODO: Should be a clean/secure way of password storing in memory vs being able to see in20#    debug/etc. Look into. I know SQLAlchemy manages to do this so might be worth digging into how21#    there.22class SplunkParentClass(object):23    """ Parent class for Splunk operations classes. This class holds all the common functions,24    attributes, etc. for all Splunk operations classes.25    Functions:26        read_connector_config27        restart_splunk_service28    Attributes:29        host: Str. The host which we connect to for the specific actions we are conducting. Can be30            IP or hostname.31        port: Int. The port used for the connection of the host.32        username: Str. The username to connect to the Splunk services. Not all children require33            this. Defaults to None.34        password: Str. The password for the passed username. Not all children require this.35            Defaults to None.36        logger_name: Str. The name of the logger this class uses to log to.37    """38    def __init__(self,39                 host,40                 port,41                 username=None,42                 password=None,43                 read_connector_config=False,44                 log_level='info',45                 log_console_output=False):46        """ Initializes the class assigning args as attributes.47        Args:48            Optional:49                read_connector_config: Bool. If true will fill in missing values with the values in50                    the connector config.51                log_level: Str. The classes will log to their loggers at this level. Defaults to52                    'info'.53                log_console_output: Bool. If True, the classes will output their logs to the54                    console.55        """56        self.host = host57        self.port = port58        self.username = username59        self.password = password60        if UNIVERSAL_FORWARDER in self.host:61            self.logger_name = FORWARDER_LOGGER_NAME62            create_logger(FORWARDER_PYTHON_LOGGER_LOCATION, log_level, self.logger_name,63                          console_output=log_console_output)64        else:65            self.logger_name = SPLUNK_LOGGER_NAME66            create_logger(SPLUNK_PYTHON_LOGGER_LOCATION, log_level, self.logger_name,67                          console_output=log_console_output)68        if read_connector_config:69            if not self.host and not self.port and not self.username and not self.password:70                try:71                    self.read_connector_config()72                except EnvironmentError:73                    log_message('warn', 'One or more values of host, port, username, or password'74                                ' not passed and no app connector file detected at %s. Values have'75                                ' not been set.' % DEFAULT_APP_FILE_LOCATION, self.logger_name)76                except MissingSettingError:77                    log_message('warn', 'Missing the Splunk INI setting in the %s file' %78                                DEFAULT_APP_FILE_LOCATION, self.logger_name)79    def restart_splunk_service(self, host=None, port=None, splunk_user=SPLUNK_USER):80        """ Restarts the splunk service. Can be on a remote server instance.81        NOTE: This only works if you are able to sudo into the splunk service account without82        password prompts (edit the sudo config as needed). If you can run splunk commands, just use83        yourself. Improper permissions result in a single line return annotating to run it from the84        correct user (neither generates an error and doesn't restart splunk)85        Args:86            Optional:87                host: Str. The host we are restarting the service on. Defaults to self.host.88                port: Int. The port used to SSH into the host. Defaults to self.port.89                splunk_user: The user able to run the splunk restart command. Defaults to the global90                    setting.91        """92        host = self.host if not host else host93        port = self.port if not port else port94        splunk_command = FORWARDER_SPLUNK_COMMAND if UNIVERSAL_FORWARDER in host else SPLUNK_COMMAND95        command = ['sudo -u %s %s restart' % (splunk_user, splunk_command)]96        remote_operations(host, command, port, catch_errors=False, tty_session=True)97    def read_connector_config(self, config_path=DEFAULT_APP_FILE_LOCATION, overwrite=False):98        """ Reads in the application connector configuration to find your Splunk settings and99        assigns them to the corresponding attributes. By default this will not overwrite attributes100        already set, but you can flip a flag to do so.101        Args:102            Optional:103                config_path: Str. The absolute path to the application connector configuration.104                    Defaults to the default home path location.105                overwrite: Bool. If set will overwrite the attribute with the read in configuration106                    values regardless if the attribute values are already set or not. Defaults to107                    False.108        Raises:109            MissingConfigError: The splunk connector config settings were not detected.110        """111        config_object_list = read_config(config_path)112        splunk_connector = get_config_object(config_object_list, 'splunk')113        if not splunk_connector:114            raise MissingConfigError('No splunk connector settings detected in the file.',115                                     self.logger_name)116        else:117            if overwrite or not self.host:118                self.host = splunk_connector.host119            if overwrite or not self.port:120                self.port = splunk_connector.port121            if overwrite or not self.username:122                self.username = splunk_connector.username123            if overwrite or not self.password:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.