KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_shutdown method in Molotov

Best Python code snippet using molotov_python

test_reconnect.py

Source:test_reconnect.py

...182    def test_reconnect(self):183        conn = engine.connect()184        eq_(conn.execute(select([1])).scalar(), 1)185        assert not conn.closed186        engine.test_shutdown()187        try:188            conn.execute(select([1]))189            assert False190        except tsa.exc.DBAPIError, e:191            if not e.connection_invalidated:192                raise193        assert not conn.closed194        assert conn.invalidated195        assert conn.invalidated196        eq_(conn.execute(select([1])).scalar(), 1)197        assert not conn.invalidated198        # one more time199        engine.test_shutdown()200        try:201            conn.execute(select([1]))202            assert False203        except tsa.exc.DBAPIError, e:204            if not e.connection_invalidated:205                raise206        assert conn.invalidated207        eq_(conn.execute(select([1])).scalar(), 1)208        assert not conn.invalidated209        conn.close()210    def test_ensure_is_disconnect_gets_connection(self):211        def is_disconnect(e, conn, cursor):212            # connection is still present213            assert conn.connection is not None214            # the error usually occurs on connection.cursor(),215            # though MySQLdb we get a non-working cursor.216            # assert cursor is None217        engine.dialect.is_disconnect = is_disconnect218        conn = engine.connect()219        engine.test_shutdown()220        assert_raises(221            tsa.exc.DBAPIError,222            conn.execute, select([1])223        )224    def test_rollback_on_invalid_plain(self):225        conn = engine.connect()226        trans = conn.begin()227        conn.invalidate()228        trans.rollback()229    @testing.requires.two_phase_transactions230    def test_rollback_on_invalid_twophase(self):231        conn = engine.connect()232        trans = conn.begin_twophase()233        conn.invalidate()234        trans.rollback()235    @testing.requires.savepoints236    def test_rollback_on_invalid_savepoint(self):237        conn = engine.connect()238        trans = conn.begin()239        trans2 = conn.begin_nested()240        conn.invalidate()241        trans2.rollback()242    def test_invalidate_twice(self):243        conn = engine.connect()244        conn.invalidate()245        conn.invalidate()246    def test_explode_in_initializer(self):247        engine = engines.testing_engine()248        def broken_initialize(connection):249            connection.execute("select fake_stuff from _fake_table")250        engine.dialect.initialize = broken_initialize251        # raises a DBAPIError, not an AttributeError252        assert_raises(exc.DBAPIError, engine.connect)253        # dispose connections so we get a new one on254        # next go255        engine.dispose()256        p1 = engine.pool257        def is_disconnect(e, conn, cursor):258            return True259        engine.dialect.is_disconnect = is_disconnect260        # invalidate() also doesn't screw up261        assert_raises(exc.DBAPIError, engine.connect)262        # pool was recreated263        assert engine.pool is not p1264    @testing.fails_on('+informixdb',265                      "Wrong error thrown, fix in informixdb?")266    def test_null_pool(self):267        engine = \268            engines.reconnecting_engine(options=dict(poolclass=pool.NullPool))269        conn = engine.connect()270        eq_(conn.execute(select([1])).scalar(), 1)271        assert not conn.closed272        engine.test_shutdown()273        try:274            conn.execute(select([1]))275            assert False276        except tsa.exc.DBAPIError, e:277            if not e.connection_invalidated:278                raise279        assert not conn.closed280        assert conn.invalidated281        eq_(conn.execute(select([1])).scalar(), 1)282        assert not conn.invalidated283    @testing.fails_on('+informixdb',284                      "Wrong error thrown, fix in informixdb?")285    def test_close(self):286        conn = engine.connect()287        eq_(conn.execute(select([1])).scalar(), 1)288        assert not conn.closed289        engine.test_shutdown()290        try:291            conn.execute(select([1]))292            assert False293        except tsa.exc.DBAPIError, e:294            if not e.connection_invalidated:295                raise296        conn.close()297        conn = engine.connect()298        eq_(conn.execute(select([1])).scalar(), 1)299    @testing.fails_on('+informixdb',300                      "Wrong error thrown, fix in informixdb?")301    def test_with_transaction(self):302        conn = engine.connect()303        trans = conn.begin()304        eq_(conn.execute(select([1])).scalar(), 1)305        assert not conn.closed306        engine.test_shutdown()307        try:308            conn.execute(select([1]))309            assert False310        except tsa.exc.DBAPIError, e:311            if not e.connection_invalidated:312                raise313        assert not conn.closed314        assert conn.invalidated315        assert trans.is_active316        assert_raises_message(317            tsa.exc.StatementError,318            "Can't reconnect until invalid transaction is "\319                "rolled back",320            conn.execute, select([1])321        )322        assert trans.is_active323        try:324            trans.commit()325            assert False326        except tsa.exc.InvalidRequestError, e:327            assert str(e) \328                == "Can't reconnect until invalid transaction is "\329                "rolled back"330        assert trans.is_active331        trans.rollback()332        assert not trans.is_active333        assert conn.invalidated334        eq_(conn.execute(select([1])).scalar(), 1)335        assert not conn.invalidated336class RecycleTest(fixtures.TestBase):337    def test_basic(self):338        for threadlocal in False, True:339            engine = engines.reconnecting_engine(340                        options={'pool_threadlocal': threadlocal})341            conn = engine.contextual_connect()342            eq_(conn.execute(select([1])).scalar(), 1)343            conn.close()344            # set the pool recycle down to 1.345            # we aren't doing this inline with the346            # engine create since cx_oracle takes way347            # too long to create the 1st connection and don't348            # want to build a huge delay into this test.349            engine.pool._recycle = 1350            # kill the DB connection351            engine.test_shutdown()352            # wait until past the recycle period353            time.sleep(2)354            # can connect, no exception355            conn = engine.contextual_connect()356            eq_(conn.execute(select([1])).scalar(), 1)357            conn.close()358meta, table, engine = None, None, None359class InvalidateDuringResultTest(fixtures.TestBase):360    def setup(self):361        global meta, table, engine362        engine = engines.reconnecting_engine()363        meta = MetaData(engine)364        table = Table('sometable', meta,365            Column('id', Integer, primary_key=True),366            Column('name', String(50)))367        meta.create_all()368        table.insert().execute(369            [{'id':i, 'name':'row %d' % i} for i in range(1, 100)]370        )371    def teardown(self):372        meta.drop_all()373        engine.dispose()374    @testing.fails_on('+pymysql',375                      "Buffers the result set and doesn't check for "376                      "connection close")377    @testing.fails_on('+mysqldb',378                      "Buffers the result set and doesn't check for "379                      "connection close")380    @testing.fails_on('+pg8000',381                      "Buffers the result set and doesn't check for "382                      "connection close")383    @testing.fails_on('+informixdb',384                      "Wrong error thrown, fix in informixdb?")385    def test_invalidate_on_results(self):386        conn = engine.connect()387        result = conn.execute('select * from sometable')388        for x in xrange(20):389            result.fetchone()390        engine.test_shutdown()391        try:392            print 'ghost result: %r' % result.fetchone()393            assert False394        except tsa.exc.DBAPIError, e:395            if not e.connection_invalidated:396                raise...

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.