Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use make_regexp_translator method in autotest

Best Python code snippet using autotest_python

scheduler_lib.py

Source:scheduler_lib.py

1#pylint: disable-msg=C01112# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.3# Use of this source code is governed by a BSD-style license that can be4# found in the LICENSE file.5"""Scheduler helper libraries.6"""7import logging8import os9import common10from autotest_lib.client.common_lib import global_config11from autotest_lib.client.common_lib import logging_config12from autotest_lib.client.common_lib import logging_manager13from autotest_lib.client.common_lib import utils14from autotest_lib.database import database_connection15from autotest_lib.frontend import setup_django_environment16from autotest_lib.frontend.afe import readonly_connection17from autotest_lib.server import utils as server_utils18DB_CONFIG_SECTION = 'AUTOTEST_WEB'19# Translations necessary for scheduler queries to work with SQLite.20# Though this is only used for testing it is included in this module to avoid21# circular imports.22_re_translator = database_connection.TranslatingDatabase.make_regexp_translator23_DB_TRANSLATORS = (24        _re_translator(r'NOW\(\)', 'time("now")'),25        _re_translator(r'LAST_INSERT_ID\(\)', 'LAST_INSERT_ROWID()'),26        # older SQLite doesn't support group_concat, so just don't bother until27        # it arises in an important query28        _re_translator(r'GROUP_CONCAT\((.*?)\)', r'\1'),29        _re_translator(r'TRUNCATE TABLE', 'DELETE FROM'),30        _re_translator(r'ISNULL\(([a-z,_]+)\)',31                       r'ifnull(nullif(\1, NULL), \1) DESC'),32)33class SchedulerError(Exception):34    """Raised by the scheduler when an inconsistent state occurs."""35class ConnectionManager(object):36    """Manager for the django database connections.37    The connection is used through scheduler_models and monitor_db.38    """39    __metaclass__ = server_utils.Singleton40    def __init__(self, readonly=True, autocommit=True):41        """Set global django database options for correct connection handling.42        @param readonly: Globally disable readonly connections.43        @param autocommit: Initialize django autocommit options.44        """45        self.db_connection = None46        # bypass the readonly connection47        readonly_connection.set_globally_disabled(readonly)48        if autocommit:49            # ensure Django connection is in autocommit50            setup_django_environment.enable_autocommit()51    @classmethod52    def open_connection(cls):53        """Open a new database connection.54        @return: An instance of the newly opened connection.55        """56        db = database_connection.DatabaseConnection(DB_CONFIG_SECTION)57        db.connect(db_type='django')58        return db59    def get_connection(self):60        """Get a connection.61        @return: A database connection.62        """63        if self.db_connection is None:64            self.db_connection = self.open_connection()65        return self.db_connection66    def disconnect(self):67        """Close the database connection."""68        try:69            self.db_connection.disconnect()70        except Exception as e:71            logging.debug('Could not close the db connection. %s', e)72    def __del__(self):73        self.disconnect()74class SchedulerLoggingConfig(logging_config.LoggingConfig):75    """Configure timestamped logging for a scheduler."""76    GLOBAL_LEVEL = logging.INFO77    @classmethod78    def get_log_name(cls, timestamped_logfile_prefix):79        """Get the name of a logfile.80        @param timestamped_logfile_prefix: The prefix to apply to the81            a timestamped log. Eg: 'scheduler' will create a logfile named82            scheduler.log.2014-05-12-17.24.02.83        @return: The timestamped log name.84        """85        return cls.get_timestamped_log_name(timestamped_logfile_prefix)86    def configure_logging(self, log_dir=None, logfile_name=None,87                          timestamped_logfile_prefix='scheduler'):88        """Configure logging to a specified logfile.89        @param log_dir: The directory to log into.90        @param logfile_name: The name of the log file.91        @timestamped_logfile_prefix: The prefix to apply to the name of92            the logfile, if a log file name isn't specified.93        """94        super(SchedulerLoggingConfig, self).configure_logging(use_console=True)95        if log_dir is None:96            log_dir = self.get_server_log_dir()97        if not logfile_name:98            logfile_name = self.get_log_name(timestamped_logfile_prefix)99        self.add_file_handler(logfile_name, logging.DEBUG, log_dir=log_dir)100        symlink_path = os.path.join(101                log_dir, '%s.latest' % timestamped_logfile_prefix)102        try:103            os.unlink(symlink_path)104        except OSError:105            pass106        os.symlink(os.path.join(log_dir, logfile_name), symlink_path)107def setup_logging(log_dir, log_name, timestamped_logfile_prefix='scheduler'):108    """Setup logging to a given log directory and log file.109    @param log_dir: The directory to log into.110    @param log_name: Name of the log file.111    @param timestamped_logfile_prefix: The prefix to apply to the logfile.112    """113    logging_manager.configure_logging(114            SchedulerLoggingConfig(), log_dir=log_dir, logfile_name=log_name,115            timestamped_logfile_prefix=timestamped_logfile_prefix)116def check_production_settings(scheduler_options):117    """Check the scheduler option's production settings.118    @param scheduler_options: Settings for scheduler.119    @raises SchedulerError: If a loclhost scheduler is started with120       production settings.121    """122    db_server = global_config.global_config.get_config_value('AUTOTEST_WEB',123                                                             'host')124    if (not scheduler_options.production and125        not utils.is_localhost(db_server)):126        raise SchedulerError('Scheduler is not running in production mode, you '127                             'should not set database to hosts other than '128                             'localhost. It\'s currently set to %s.\nAdd option'129                             ' --production if you want to skip this check.' %...

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.