Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_gevent_socket_pymssql_callproc_wait_read_concurrency(self):
def wait_callback(read_fileno):
gevent.socket.wait_read(read_fileno)
pymssql.set_wait_callback(wait_callback)
with pymssqlconn() as conn:
cur = conn.cursor()
proc_name = 'my_proc'
cur.execute("IF OBJECT_ID('%s', 'P') IS NOT NULL DROP PROCEDURE %s" % (proc_name, proc_name))
cur.execute("""
CREATE PROCEDURE %s AS
BEGIN
SET NOCOUNT ON
WAITFOR DELAY '00:00:05' -- sleep for 5 seconds
SELECT CURRENT_TIMESTAMP
END
""" % (proc_name,))
conn.commit()
elapsed_time = self._run_all_greenlets(
def test_gevent_socket_mssql_execute_wait_read_concurrency(self):
def wait_callback(read_fileno):
gevent.socket.wait_read(read_fileno)
pymssql.set_wait_callback(wait_callback)
elapsed_time = self._run_all_greenlets(
self.greenlet_run_mssql_execute)
self.assertTrue(
elapsed_time < datetime.timedelta(seconds=20),
'elapsed_time < 20 seconds')
def test_gevent_socket_pymssql_execute_wait_read_concurrency(self):
def wait_callback(read_fileno):
gevent.socket.wait_read(read_fileno)
pymssql.set_wait_callback(wait_callback)
elapsed_time = self._run_all_greenlets(
self.greenlet_run_pymssql_execute)
self.assertTrue(
elapsed_time < datetime.timedelta(seconds=20),
'elapsed_time < 20 seconds')