3 from eventlet.support import six
5 from tests import patcher_test, skip_unless
6 from tests import get_database_auth
7 from tests.db_pool_test import postgres_requirement
9 psycopg_test_file = """
13 eventlet.monkey_patch()
14 from eventlet import patcher
15 if not patcher.is_monkey_patched('psycopg'):
16 print("Psycopg not monkeypatched")
20 def tick(totalseconds, persecond):
21 for i in range(totalseconds*persecond):
23 eventlet.sleep(1.0/persecond)
25 dsn = os.environ['PSYCOPG_TEST_DSN']
28 conn = psycopg2.connect(dsn)
31 cur.execute("select pg_sleep(%s)", (secs,))
33 f = eventlet.spawn(fetch, 2, 1)
34 t = eventlet.spawn(tick, 2, 100)
36 assert count[0] > 100, count[0]
41 class PatchingPsycopg(patcher_test.ProcessBase):
42 @skip_unless(postgres_requirement)
43 def test_psycopg_patched(self):
44 if 'PSYCOPG_TEST_DSN' not in os.environ:
45 # construct a non-json dsn for the subprocess
46 psycopg_auth = get_database_auth()['psycopg2']
47 if isinstance(psycopg_auth, str):
50 dsn = " ".join(["%s=%s" % (k, v) for k, v in six.iteritems(psycopg_auth)])
51 os.environ['PSYCOPG_TEST_DSN'] = dsn
52 self.write_to_tempfile("psycopg_patcher", psycopg_test_file)
53 output, lines = self.launch_subprocess('psycopg_patcher.py')
54 if lines[0].startswith('Psycopg not monkeypatched'):
55 print("Can't test psycopg2 patching; it's not installed.")
57 # if there's anything wrong with the test program it'll have a stack trace
58 assert lines[0].startswith('done'), output