1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
import os
from tests import patcher_test, skip_unless
from tests import get_database_auth
from tests.db_pool_test import postgres_requirement
psycopg_test_file = """
import os
import sys
import eventlet
eventlet.monkey_patch()
from eventlet import patcher
if not patcher.is_monkey_patched('psycopg'):
print "Psycopg not monkeypatched"
sys.exit(0)
count = [0]
def tick(totalseconds, persecond):
for i in xrange(totalseconds*persecond):
count[0] += 1
eventlet.sleep(1.0/persecond)
dsn = os.environ['PSYCOPG_TEST_DSN']
import psycopg2
def fetch(num, secs):
conn = psycopg2.connect(dsn)
cur = conn.cursor()
for i in range(num):
cur.execute("select pg_sleep(%s)", (secs,))
f = eventlet.spawn(fetch, 2, 1)
t = eventlet.spawn(tick, 2, 100)
f.wait()
assert count[0] > 100, count[0]
print "done"
"""
class PatchingPsycopg(patcher_test.ProcessBase):
@skip_unless(postgres_requirement)
def test_psycopg_patched(self):
if 'PSYCOPG_TEST_DSN' not in os.environ:
# construct a non-json dsn for the subprocess
psycopg_auth = get_database_auth()['psycopg2']
if isinstance(psycopg_auth,str):
dsn = psycopg_auth
else:
dsn = " ".join(["%s=%s" % (k,v) for k,v, in psycopg_auth.iteritems()])
os.environ['PSYCOPG_TEST_DSN'] = dsn
self.write_to_tempfile("psycopg_patcher", psycopg_test_file)
output, lines = self.launch_subprocess('psycopg_patcher.py')
if lines[0].startswith('Psycopg not monkeypatched'):
print "Can't test psycopg2 patching; it's not installed."
return
# if there's anything wrong with the test program it'll have a stack trace
self.assert_(lines[0].startswith('done'), output)
|
