Method connect

lib/sqlalchemy/testing/engines.py:250–261

(self, *args, **kwargs)

        return getattr(self.dbapi, key)

    def connect(self, *args, **kwargs):
        conn = self.dbapi.connect(*args, **kwargs)
        if self.is_stopped:
            self._safe(conn.close)
            curs = conn.cursor()  # should fail on Oracle etc.
            # should fail for everything that didn't fail
            # above, connection is closed
            curs.execute("select 1")
            assert False, "simulated connect failure didn't work"
        else:
            self.connections.append(conn)
            return conn

    def _safe(self, fn):
        try: