test_timeout.py

# Copyright (C) 2016-present the asyncpg authors and contributors
# <see AUTHORS file>
#
# This module is part of asyncpg and is released under
# the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0


importasyncio

importasyncpg
fromasyncpgimportconnectionaspg_connection
fromasyncpgimport_testbaseastb


MAX_RUNTIME=0.5


classTestTimeout(tb.ConnectedTestCase):

asyncdeftest_timeout_01(self):
formethnamein {'fetch', 'fetchrow', 'fetchval', 'execute'}:
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
meth=getattr(self.con, methname)
awaitmeth('select pg_sleep(10)', timeout=0.02)
self.assertEqual(awaitself.con.fetch('select 1'), [(1,)])

asyncdeftest_timeout_02(self):
st=awaitself.con.prepare('select pg_sleep(10)')

formethnamein {'fetch', 'fetchrow', 'fetchval'}:
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
meth=getattr(st, methname)
awaitmeth(timeout=0.02)
self.assertEqual(awaitself.con.fetch('select 1'), [(1,)])

asyncdeftest_timeout_03(self):
task=self.loop.create_task(
self.con.fetch('select pg_sleep(10)', timeout=0.2))
awaitasyncio.sleep(0.05)
task.cancel()
withself.assertRaises(asyncio.CancelledError), \
self.assertRunUnder(MAX_RUNTIME):
awaittask
self.assertEqual(awaitself.con.fetch('select 1'), [(1,)])

asyncdeftest_timeout_04(self):
st=awaitself.con.prepare('select pg_sleep(10)', timeout=0.1)
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
asyncwithself.con.transaction():
asyncfor_inst.cursor(timeout=0.1): # NOQA
pass
self.assertEqual(awaitself.con.fetch('select 1'), [(1,)])

st=awaitself.con.prepare('select pg_sleep(10)', timeout=0.1)
asyncwithself.con.transaction():
cur=awaitst.cursor()
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
awaitcur.fetch(1, timeout=0.1)
self.assertEqual(awaitself.con.fetch('select 1'), [(1,)])

asyncdeftest_timeout_05(self):
# Stress-test timeouts - try to trigger a race condition
# between a cancellation request to Postgres and next
# query (SELECT 1)
for_inrange(500):
withself.assertRaises(asyncio.TimeoutError):
awaitself.con.fetch('SELECT pg_sleep(1)', timeout=1e-10)
self.assertEqual(awaitself.con.fetch('SELECT 1'), [(1,)])

asyncdeftest_timeout_06(self):
asyncwithself.con.transaction():
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
asyncfor_inself.con.cursor( # NOQA
'select pg_sleep(10)', timeout=0.1):
pass
self.assertEqual(awaitself.con.fetch('select 1'), [(1,)])

asyncwithself.con.transaction():
cur=awaitself.con.cursor('select pg_sleep(10)')
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
awaitcur.fetch(1, timeout=0.1)

asyncwithself.con.transaction():
cur=awaitself.con.cursor('select pg_sleep(10)')
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
awaitcur.forward(1, timeout=1e-10)

asyncwithself.con.transaction():
cur=awaitself.con.cursor('select pg_sleep(10)')
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
awaitcur.fetchrow(timeout=0.1)

asyncwithself.con.transaction():
cur=awaitself.con.cursor('select pg_sleep(10)')
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
awaitcur.fetchrow(timeout=0.1)

withself.assertRaises(asyncpg.InFailedSQLTransactionError):
awaitcur.fetch(1)

self.assertEqual(awaitself.con.fetch('select 1'), [(1,)])

asyncdeftest_invalid_timeout(self):
forcommand_timeoutin ('a', False, -1):
withself.subTest(command_timeout=command_timeout):
withself.assertRaisesRegex(ValueError,
'invalid command_timeout'):
awaitself.connect(command_timeout=command_timeout)

# Note: negative timeouts are OK for method calls.
formethnamein {'fetch', 'fetchrow', 'fetchval', 'execute'}:
fortimeoutin ('a', False):
withself.subTest(timeout=timeout):
withself.assertRaisesRegex(ValueError, 'invalid timeout'):
awaitself.con.execute('SELECT 1', timeout=timeout)


classTestConnectionCommandTimeout(tb.ConnectedTestCase):

@tb.with_connection_options(command_timeout=0.2)
asyncdeftest_command_timeout_01(self):
formethnamein {'fetch', 'fetchrow', 'fetchval', 'execute'}:
withself.assertRaises(asyncio.TimeoutError), \
self.assertRunUnder(MAX_RUNTIME):
meth=getattr(self.con, methname)
awaitmeth('select pg_sleep(10)')
self.assertEqual(awaitself.con.fetch('select 1'), [(1,)])


classSlowPrepareConnection(pg_connection.Connection):
"""Connection class to test timeouts."""
asyncdef_get_statement(self, query, timeout, **kwargs):
awaitasyncio.sleep(0.3)
returnawaitsuper()._get_statement(query, timeout, **kwargs)


classTestTimeoutCoversPrepare(tb.ConnectedTestCase):

@tb.with_connection_options(connection_class=SlowPrepareConnection,
command_timeout=0.3)
asyncdeftest_timeout_covers_prepare_01(self):
formethnamein {'fetch', 'fetchrow', 'fetchval', 'execute'}:
withself.assertRaises(asyncio.TimeoutError):
meth=getattr(self.con, methname)
awaitmeth('select pg_sleep($1)', 0.2)