How to use _copy_exceptions method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...6_GLOBAL_CONFIG_NAMES = {7 'username' : 'user',8 'db_name' : 'database',9}10def _copy_exceptions(source, destination):11 for exception_name in _DB_EXCEPTIONS:12 setattr(destination, exception_name, getattr(source, exception_name))13class _GenericBackend(object):14 def __init__(self, database_module):15 self._database_module = database_module16 self._connection = None17 self._cursor = None18 self.rowcount = None19 _copy_exceptions(database_module, self)20 def connect(self, host=None, username=None, password=None, db_name=None):21 """22 This is assumed to enable autocommit.23 """24 raise NotImplementedError25 def disconnect(self):26 if self._connection:27 self._connection.close()28 self._connection = None29 self._cursor = None30 def execute(self, query, parameters=None):31 if parameters is None:32 parameters = ()33 self._cursor.execute(query, parameters)34 self.rowcount = self._cursor.rowcount35 return self._cursor.fetchall()36class _MySqlBackend(_GenericBackend):37 def __init__(self):38 import MySQLdb39 super(_MySqlBackend, self).__init__(MySQLdb)40 @staticmethod41 def convert_boolean(boolean, conversion_dict):42 'Convert booleans to integer strings'43 return str(int(boolean))44 def connect(self, host=None, username=None, password=None, db_name=None):45 import MySQLdb.converters46 convert_dict = MySQLdb.converters.conversions47 convert_dict.setdefault(bool, self.convert_boolean)48 self._connection = self._database_module.connect(49 host=host, user=username, passwd=password, db=db_name,50 conv=convert_dict)51 self._connection.autocommit(True)52 self._cursor = self._connection.cursor()53class _SqliteBackend(_GenericBackend):54 def __init__(self):55 from pysqlite2 import dbapi256 super(_SqliteBackend, self).__init__(dbapi2)57 self._last_insert_id_re = re.compile(r'\sLAST_INSERT_ID\(\)',58 re.IGNORECASE)59 def connect(self, host=None, username=None, password=None, db_name=None):60 self._connection = self._database_module.connect(db_name)61 self._connection.isolation_level = None # enable autocommit62 self._cursor = self._connection.cursor()63 def execute(self, query, parameters=None):64 # pysqlite2 uses paramstyle=qmark65 # TODO: make this more sophisticated if necessary66 query = query.replace('%s', '?')67 # pysqlite2 can't handle parameters=None (it throws a nonsense68 # exception)69 if parameters is None:70 parameters = ()71 # sqlite3 doesn't support MySQL's LAST_INSERT_ID(). Instead it has72 # something similar called LAST_INSERT_ROWID() that will do enough of73 # what we want (for our non-concurrent unittest use case).74 query = self._last_insert_id_re.sub(' LAST_INSERT_ROWID()', query)75 return super(_SqliteBackend, self).execute(query, parameters)76class _DjangoBackend(_GenericBackend):77 def __init__(self):78 from django.db import backend, connection, transaction79 super(_DjangoBackend, self).__init__(backend.Database)80 self._django_connection = connection81 self._django_transaction = transaction82 def connect(self, host=None, username=None, password=None, db_name=None):83 self._connection = self._django_connection84 self._cursor = self._connection.cursor()85 def execute(self, query, parameters=None):86 try:87 return super(_DjangoBackend, self).execute(query,88 parameters=parameters)89 finally:90 self._django_transaction.commit_unless_managed()91_BACKEND_MAP = {92 'mysql': _MySqlBackend,93 'sqlite': _SqliteBackend,94 'django': _DjangoBackend,95}96class DatabaseConnection(object):97 """98 Generic wrapper for a database connection. Supports both mysql and sqlite99 backends.100 Public attributes:101 * reconnect_enabled: if True, when an OperationalError occurs the class will102 try to reconnect to the database automatically.103 * reconnect_delay_sec: seconds to wait before reconnecting104 * max_reconnect_attempts: maximum number of time to try reconnecting before105 giving up. Setting to RECONNECT_FOREVER removes the limit.106 * rowcount - will hold cursor.rowcount after each call to execute().107 * global_config_section - the section in which to find DB information. this108 should be passed to the constructor, not set later, and may be None, in109 which case information must be passed to connect().110 * debug - if set True, all queries will be printed before being executed111 """112 _DATABASE_ATTRIBUTES = ('db_type', 'host', 'username', 'password',113 'db_name')114 def __init__(self, global_config_section=None, debug=False):115 self.global_config_section = global_config_section116 self._backend = None117 self.rowcount = None118 self.debug = debug119 # reconnect defaults120 self.reconnect_enabled = True121 self.reconnect_delay_sec = 20122 self.max_reconnect_attempts = 10123 self._read_options()124 def _get_option(self, name, provided_value):125 if provided_value is not None:126 return provided_value127 if self.global_config_section:128 global_config_name = _GLOBAL_CONFIG_NAMES.get(name, name)129 return global_config.global_config.get_config_value(130 self.global_config_section, global_config_name)131 return getattr(self, name, None)132 def _read_options(self, db_type=None, host=None, username=None,133 password=None, db_name=None):134 self.db_type = self._get_option('db_type', db_type)135 = self._get_option('host', host)136 self.username = self._get_option('username', username)137 self.password = self._get_option('password', password)138 self.db_name = self._get_option('db_name', db_name)139 def _get_backend(self, db_type):140 if db_type not in _BACKEND_MAP:141 raise ValueError('Invalid database type: %s, should be one of %s' %142 (db_type, ', '.join(_BACKEND_MAP.keys())))143 backend_class = _BACKEND_MAP[db_type]144 return backend_class()145 def _reached_max_attempts(self, num_attempts):146 return (self.max_reconnect_attempts is not RECONNECT_FOREVER and147 num_attempts > self.max_reconnect_attempts)148 def _is_reconnect_enabled(self, supplied_param):149 if supplied_param is not None:150 return supplied_param151 return self.reconnect_enabled152 def _connect_backend(self, try_reconnecting=None):153 num_attempts = 0154 while True:155 try:156 self._backend.connect(, username=self.username,157 password=self.password,158 db_name=self.db_name)159 return160 except self._backend.OperationalError:161 num_attempts += 1162 if not self._is_reconnect_enabled(try_reconnecting):163 raise164 if self._reached_max_attempts(num_attempts):165 raise166 traceback.print_exc()167 print ("Can't connect to database; reconnecting in %s sec" %168 self.reconnect_delay_sec)169 time.sleep(self.reconnect_delay_sec)170 self.disconnect()171 def connect(self, db_type=None, host=None, username=None, password=None,172 db_name=None, try_reconnecting=None):173 """174 Parameters passed to this function will override defaults from global175 config. try_reconnecting, if passed, will override176 self.reconnect_enabled.177 """178 self.disconnect()179 self._read_options(db_type, host, username, password, db_name)180 self._backend = self._get_backend(self.db_type)181 _copy_exceptions(self._backend, self)182 self._connect_backend(try_reconnecting)183 def disconnect(self):184 if self._backend:185 self._backend.disconnect()186 def execute(self, query, parameters=None, try_reconnecting=None):187 """188 Execute a query and return cursor.fetchall(). try_reconnecting, if189 passed, will override self.reconnect_enabled.190 """191 if self.debug:192 print 'Executing %s, %s' % (query, parameters)193 # _connect_backend() contains a retry loop, so don't loop here194 try:195 results = self._backend.execute(query, parameters)...

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:


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

Run autotest 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?