How to use _get_env method in avocado

Best Python code snippet using avocado_python

logger.py

Source:logger.py Github

copy

Full Screen

...9def get_logger():10 return __log11def new_logger(name, properties: dict = None):12 return _DelayedConfigurationLogger(name, properties=properties)13def _get_env(key, default=None):14 return os.environ.get(key, default)15def _get_properties(properties: dict, key, default=None):16 keys = key.split('.')17 p = properties18 for k in keys:19 if k in p:20 p = p[k]21 else:22 return default23 return p24def _log_handle_env(log_: logging.Logger, path, default_state, handler_type=logging.StreamHandler, *args, **kwargs):25 if eval(_get_env(path + '.open', default_state)):26 handle = handler_type(*args, **kwargs)27 handle.setLevel(eval('logging.' + _get_env(path + '.level', 'INFO').upper()))28 handle.setFormatter(29 logging.Formatter(_get_env(path + '.format', '%(asctime)s - %(name)s - %(levelname)s - %(message)s')))30 log_.addHandler(handle)31def _log_handle_properties(log_: logging.Logger, properties: dict, path, default_state,32 handler_type=logging.StreamHandler, *args, **kwargs):33 if eval(_get_properties(properties, path + '.open', default_state)):34 handle = handler_type(*args, **kwargs)35 handle.setLevel(eval('logging.' + _get_properties(properties, path + '.level', 'INFO').upper()))36 handle.setFormatter(37 logging.Formatter(38 _get_properties(properties, path + '.format', '%(asctime)s - %(name)s - %(levelname)s - %(message)s')))39 log_.addHandler(handle)40def _delay_config(func):41 def wrapper(*args, **kwargs):42 args[0].init_config()43 return func(*args, **kwargs)44 return wrapper45class _DelayedConfigurationLogger(Logger):46 """47 延迟初始化配置的日志对象, 仅用于当前全局日志对象的延迟装载48 """49 def __init__(self, name: str, level=logging.INFO, properties=None):50 super().__init__("_DelayedConfigurationLogger." + name, level)51 self.__delegate = logging.getLogger(name)52 self.__init = False53 self.__properties = properties54 @_delay_config55 def setLevel(self, level):56 self.__delegate.setLevel(level)57 @_delay_config58 def debug(self, msg, *args, **kwargs):59 self.__delegate.debug(msg, *args, **kwargs)60 @_delay_config61 def info(self, msg, *args, **kwargs):62 self.__delegate.info(msg, *args, **kwargs)63 @_delay_config64 def warning(self, msg, *args, **kwargs):65 self.__delegate.warning(msg, *args, **kwargs)66 @_delay_config67 def error(self, msg, *args, **kwargs):68 self.__delegate.error(msg, *args, **kwargs)69 @_delay_config70 def exception(self, msg, *args, exc_info=True, **kwargs):71 self.__delegate.exception(msg, *args, exc_info=exc_info, **kwargs)72 @_delay_config73 def critical(self, msg, *args, **kwargs):74 self.__delegate.critical(msg, *args, **kwargs)75 fatal = critical76 @_delay_config77 def log(self, level, msg, *args, **kwargs):78 self.__delegate.log(level, msg, *args, **kwargs)79 @_delay_config80 def findCaller(self, stack_info=False):81 return self.__delegate.findCaller(stack_info=stack_info)82 @_delay_config83 def makeRecord(self, name, level, fn, lno, msg, args, exc_info,84 func=None, extra=None, sinfo=None):85 return self.__delegate.makeRecord(name, level, fn, lno, msg, args, exc_info, func, extra, sinfo)86 @_delay_config87 def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False):88 """89 Not used90 """91 pass92 @_delay_config93 def handle(self, record):94 self.__delegate.handle(record)95 @_delay_config96 def addHandler(self, hdlr):97 self.__delegate.addHandler(hdlr)98 @_delay_config99 def removeHandler(self, hdlr):100 self.__delegate.removeHandler(hdlr)101 @_delay_config102 def hasHandlers(self):103 return self.__delegate.hasHandlers()104 @_delay_config105 def callHandlers(self, record):106 self.__delegate.callHandlers(record)107 @_delay_config108 def getEffectiveLevel(self):109 return self.__delegate.getEffectiveLevel()110 @_delay_config111 def isEnabledFor(self, level):112 return self.__delegate.isEnabledFor(level)113 @_delay_config114 def getChild(self, suffix):115 return self.__delegate.getChild(suffix)116 @_delay_config117 def handlers_(self):118 return self.__delegate.handlers119 @_delay_config120 def root_(self):121 return self.__delegate.root122 @_delay_config123 def filters_(self):124 return self.__delegate.filters125 @_delay_config126 def disabled_(self):127 return self.__delegate.disabled128 @_delay_config129 def manager_(self):130 return self.__delegate.manager131 @_delay_config132 def parent_(self):133 return self.__delegate.parent134 @_delay_config135 def propagate_(self):136 return self.__delegate.propagate137 @_delay_config138 def level_(self):139 return self.__delegate.level140 def init_config(self):141 """142 从环境变量中初始化日志配置143 """144 if not self.__init:145 if not self.__properties:146 self.__delegate.setLevel(eval('logging.' + _get_env('logging.level', 'INFO').upper()))147 _log_handle_env(self.__delegate, _STREAM_HANDLE, 'True')148 _log_handle_env(self.__delegate, _FILE_HANDLE, 'False', logging.FileHandler,149 _get_env(_FILE_HANDLE + '.filename', './global.log'),150 _get_env(_FILE_HANDLE + '.mode', 'a'),151 _get_env(_FILE_HANDLE + '.encoding', 'utf-8'),152 eval(_get_env(_FILE_HANDLE + '.delay', 'False')))153 _log_handle_env(self.__delegate, _ROTATING_FILE_HANDLE, 'False', logging.handlers.RotatingFileHandler,154 _get_env(_ROTATING_FILE_HANDLE + '.filename', './rotating.log'),155 _get_env(_ROTATING_FILE_HANDLE + '.mode', 'a'),156 int(_get_env(_ROTATING_FILE_HANDLE + '.maxBytes', 0)),157 int(_get_env(_ROTATING_FILE_HANDLE + '.backupCount', 0)),158 _get_env(_ROTATING_FILE_HANDLE + '.encoding', 'utf-8'),159 eval(_get_env(_ROTATING_FILE_HANDLE + '.delay', 'False')))160 _log_handle_env(self.__delegate, _TIME_ROTATING_FILE_HANDLE, 'False',161 logging.handlers.TimedRotatingFileHandler,162 _get_env(_TIME_ROTATING_FILE_HANDLE + '.filename', './rotating.log'),163 _get_env(_TIME_ROTATING_FILE_HANDLE + '.when', 'h'),164 int(_get_env(_TIME_ROTATING_FILE_HANDLE + '.interval', 1)),165 int(_get_env(_TIME_ROTATING_FILE_HANDLE + '.backupCount', 0)),166 _get_env(_TIME_ROTATING_FILE_HANDLE + '.encoding', 'utf-8'),167 eval(_get_env(_TIME_ROTATING_FILE_HANDLE + '.delay', 'False')),168 eval(_get_env(_TIME_ROTATING_FILE_HANDLE + '.utc', 'False')),169 _get_env(_TIME_ROTATING_FILE_HANDLE + '.atTime')170 )171 else:172 self.__delegate.setLevel(173 eval('logging.' + _get_properties(self.__properties, 'logging.level', 'INFO').upper()))174 _log_handle_properties(self.__delegate, self.__properties, _STREAM_HANDLE, 'True')175 _log_handle_properties(self.__delegate, self.__properties, _FILE_HANDLE, 'False',176 logging.FileHandler,177 _get_properties(self.__properties, _FILE_HANDLE + '.filename',178 './custom.log'),179 _get_properties(self.__properties, _FILE_HANDLE + '.mode', 'a'),180 _get_properties(self.__properties, _FILE_HANDLE + '.encoding', 'utf-8'),181 eval(_get_properties(self.__properties, _FILE_HANDLE + '.delay', 'False')))182 _log_handle_properties(self.__delegate, self.__properties, _ROTATING_FILE_HANDLE, 'False',183 logging.handlers.RotatingFileHandler,...

Full Screen

Full Screen

change.py

Source:change.py Github

copy

Full Screen

...13 """14 self.client = None15 self.waiter = None16 self.rr_skeleton = dict()17 def _get_env(self, variable, exit=True):18 """19 Try to fetch a variable from the environment.20 Per default the method will raise an exception if the variable isn't present.21 This behaviour can be switched off via the exit flag.22 """23 value = os.environ.get(variable)24 if not value and exit:25 raise NameError("Cannot find environment variable: " + str(variable))26 return value27 def _connect(self):28 """29 Creates a new client object which wraps the connection to AWS.30 """31 if not self.client:32 self.client = boto3.client(33 "route53",34 aws_access_key_id=self._get_env("INPUT_AWS_ACCESS_KEY_ID"),35 aws_secret_access_key=self._get_env("INPUT_AWS_SECRET_ACCESS_KEY")36 )37 self.waiter = self.client.get_waiter("resource_record_sets_changed")38 def _set_comment(self):39 """40 Appends an additional comment field to the record set.41 """42 comment = self._get_env("INPUT_AWS_ROUTE53_RR_COMMENT", False)43 if comment:44 self.rr_skeleton["Comment"] = comment45 def _set_base_changes(self):46 """47 Creates the base skeleton required for creating a new record set.48 """49 self.rr_skeleton["Changes"] = [{50 "Action": self._get_env("INPUT_AWS_ROUTE53_RR_ACTION"),51 "ResourceRecordSet": {52 "Name": self._get_env("INPUT_AWS_ROUTE53_RR_NAME"),53 "Type": self._get_env("INPUT_AWS_ROUTE53_RR_TYPE"),54 "TTL": int(self._get_env("INPUT_AWS_ROUTE53_RR_TTL")),55 "ResourceRecords": [{56 "Value": self._get_env("INPUT_AWS_ROUTE53_RR_VALUE")57 }]58 }59 }]60 def _build_record_set(self):61 """62 Builds up the skeleton used for modulating the record set.63 """64 self._set_comment()65 self._set_base_changes()66 return self.rr_skeleton67 def _change_record_set(self, record_set):68 """69 Requests the required change at AWS.70 """71 return self.client.change_resource_record_sets(72 HostedZoneId=self._get_env("INPUT_AWS_ROUTE53_HOSTED_ZONE_ID"),73 ChangeBatch=record_set74 )75 def _wait(self, request_id):76 """77 Waits until the requested operations is finished.78 """79 wait = self._get_env("INPUT_AWS_ROUTE53_WAIT", False)80 if wait and util.strtobool(wait):81 self.waiter.wait(82 Id=request_id,83 WaiterConfig={84 "Delay": 10,85 "MaxAttempts": 5086 }87 )88 def _obtain_request_id(self, result):89 """90 Grabs and returns the id of the given request.91 """92 return result["ChangeInfo"]["Id"]93 def _obtain_marshalled_result(self, result):...

Full Screen

Full Screen

config.py

Source:config.py Github

copy

Full Screen

1import os2def _get_env(env: str, default: str):3 if env in os.environ:4 return os.environ[env]5 return default6# Cow Transfer settings7COW_USERNAME = _get_env('COW_USERNAME', '')8COW_PASSWORD = _get_env('COW_PASSWORD', '')9# Database settings10# sqlite / mysql11DATABASE_TYPE = _get_env('DATABASE_TYPE', 'sqlite')12# SQLite settings13SQLITE_NAME = _get_env('SQLITE_NAME', 'cow.db')14# MySQL settings15MYSQL_DB = _get_env('MYSQL_DB', 'cow')16MYSQL_HOST = _get_env('MYSQL_HOST', '127.0.0.1')17MYSQL_PORT = int(_get_env('MYSQL_PORT', 3306))18MYSQL_USER = _get_env('MYSQL_USER', 'root')...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful