eventlet/tests/patcher_psycopg_test.py

   1 import os
   2
   3 from eventlet.support import six
   4
   5 from tests import patcher_test, skip_unless
   6 from tests import get_database_auth
   7 from tests.db_pool_test import postgres_requirement
   8
   9 psycopg_test_file = """
  10 import os
  11 import sys
  12 import eventlet
  13 eventlet.monkey_patch()
  14 from eventlet import patcher
  15 if not patcher.is_monkey_patched('psycopg'):
  16     print("Psycopg not monkeypatched")
  17     sys.exit(0)
  18
  19 count = [0]
  20 def tick(totalseconds, persecond):
  21     for i in range(totalseconds*persecond):
  22         count[0] += 1
  23         eventlet.sleep(1.0/persecond)
  24
  25 dsn = os.environ['PSYCOPG_TEST_DSN']
  26 import psycopg2
  27 def fetch(num, secs):
  28     conn = psycopg2.connect(dsn)
  29     cur = conn.cursor()
  30     for i in range(num):
  31         cur.execute("select pg_sleep(%s)", (secs,))
  32
  33 f = eventlet.spawn(fetch, 2, 1)
  34 t = eventlet.spawn(tick, 2, 100)
  35 f.wait()
  36 assert count[0] > 100, count[0]
  37 print("done")
  38 """
  39
  40
  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):
  48                 dsn = psycopg_auth
  49             else:
  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.")
  56             return
  57         # if there's anything wrong with the test program it'll have a stack trace
  58         assert lines[0].startswith('done'), output