How to use _get_database_connection method in autotest

Best Python code snippet using autotest_python

database_connection_unittest.py

Source:database_connection_unittest.py

...21        self.god.stub_function(time, 'sleep')22    def tearDown(self):23        global_config.global_config.reset_config_values()24        self.god.unstub_all()25    def _get_database_connection(self, config_section=_CONFIG_SECTION):26        if config_section == _CONFIG_SECTION:27            self._override_config()28        db = database_connection.DatabaseConnection(config_section)29        self._fake_backend = self.god.create_mock_class(30            database_connection._GenericBackend, 'fake_backend')31        for exception in database_connection._DB_EXCEPTIONS:32            setattr(self._fake_backend, exception, FakeDatabaseError)33        self._fake_backend.rowcount = 034        def get_fake_backend(db_type):35            self._db_type = db_type36            return self._fake_backend37        self.god.stub_with(db, '_get_backend', get_fake_backend)38        db.reconnect_delay_sec = _RECONNECT_DELAY39        return db40    def _override_config(self):41        c = global_config.global_config42        c.override_config_value(_CONFIG_SECTION, 'host', _HOST)43        c.override_config_value(_CONFIG_SECTION, 'user', _USER)44        c.override_config_value(_CONFIG_SECTION, 'password', _PASS)45        c.override_config_value(_CONFIG_SECTION, 'database', _DB_NAME)46        c.override_config_value(_CONFIG_SECTION, 'db_type', _DB_TYPE)47    def test_connect(self):48        db = self._get_database_connection(config_section=None)49        self._fake_backend.connect.expect_call(**_CONNECT_KWARGS)50        db.connect(db_type=_DB_TYPE, host=_HOST, username=_USER,51                   password=_PASS, db_name=_DB_NAME)52        self.assertEquals(self._db_type, _DB_TYPE)53        self.god.check_playback()54    def test_global_config(self):55        db = self._get_database_connection()56        self._fake_backend.connect.expect_call(**_CONNECT_KWARGS)57        db.connect()58        self.assertEquals(self._db_type, _DB_TYPE)59        self.god.check_playback()60    def _expect_reconnect(self, fail=False):61        self._fake_backend.disconnect.expect_call()62        call = self._fake_backend.connect.expect_call(**_CONNECT_KWARGS)63        if fail:64            call.and_raises(FakeDatabaseError())65    def _expect_fail_and_reconnect(self, num_reconnects, fail_last=False):66        self._fake_backend.connect.expect_call(**_CONNECT_KWARGS).and_raises(67            FakeDatabaseError())68        for i in xrange(num_reconnects):69            time.sleep.expect_call(_RECONNECT_DELAY)70            if i < num_reconnects - 1:71                self._expect_reconnect(fail=True)72            else:73                self._expect_reconnect(fail=fail_last)74    def test_connect_retry(self):75        db = self._get_database_connection()76        self._expect_fail_and_reconnect(1)77        db.connect()78        self.god.check_playback()79        self._fake_backend.disconnect.expect_call()80        self._expect_fail_and_reconnect(0)81        self.assertRaises(FakeDatabaseError, db.connect,82                          try_reconnecting=False)83        self.god.check_playback()84        db.reconnect_enabled = False85        self._fake_backend.disconnect.expect_call()86        self._expect_fail_and_reconnect(0)87        self.assertRaises(FakeDatabaseError, db.connect)88        self.god.check_playback()89    def test_max_reconnect(self):90        db = self._get_database_connection()91        db.max_reconnect_attempts = 592        self._expect_fail_and_reconnect(5, fail_last=True)93        self.assertRaises(FakeDatabaseError, db.connect)94        self.god.check_playback()95    def test_reconnect_forever(self):96        db = self._get_database_connection()97        db.max_reconnect_attempts = database_connection.RECONNECT_FOREVER98        self._expect_fail_and_reconnect(30)99        db.connect()100        self.god.check_playback()101    def _simple_connect(self, db):102        self._fake_backend.connect.expect_call(**_CONNECT_KWARGS)103        db.connect()104        self.god.check_playback()105    def test_disconnect(self):106        db = self._get_database_connection()107        self._simple_connect(db)108        self._fake_backend.disconnect.expect_call()109        db.disconnect()110        self.god.check_playback()111    def test_execute(self):112        db = self._get_database_connection()113        self._simple_connect(db)114        params = object()115        self._fake_backend.execute.expect_call('query', params)116        db.execute('query', params)117        self.god.check_playback()118    def test_execute_retry(self):119        db = self._get_database_connection()120        self._simple_connect(db)121        self._fake_backend.execute.expect_call('query', None).and_raises(122            FakeDatabaseError())123        self._expect_reconnect()124        self._fake_backend.execute.expect_call('query', None)125        db.execute('query')126        self.god.check_playback()127        self._fake_backend.execute.expect_call('query', None).and_raises(128            FakeDatabaseError())129        self.assertRaises(FakeDatabaseError, db.execute, 'query',130                          try_reconnecting=False)131if __name__ == '__main__':...

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.