tests: ProcessBase -> run_pythonrun_python

Extracted inline code to permanent files
Python3 compatibility of test files
tests.run_python: new_tmp flag to create new temporary directory and
    pass it to child process via environ[TMP].
    The directory with any files is removed when child process is
    finished.
tests.rmtree: silent version of shutil.rmtree()

1 files changed, 37 insertions, 0 deletions

diff --git a/tests/patcher_psycopg_test_patched.py b/tests/patcher_psycopg_test_patched.py
new file mode 100644
index 0000000..9be75ef
--- /dev/null
+++ b/tests/patcher_psycopg_test_patched.py
@@ -0,0 +1,37 @@
+from __future__ import print_function
+# no standard tests in this file, ignore
+__test__ = False
+
+
+def fetch(dsn, num, secs):
+    conn = psycopg2.connect(dsn)
+    cur = conn.cursor()
+    for i in range(num):
+        cur.execute("select pg_sleep(%s)", [secs])
+
+
+def tick(count, totalseconds, persecond):
+    for i in range(totalseconds * persecond):
+        count[0] += 1
+        eventlet.sleep(1.0 / persecond)
+
+
+if __name__ == '__main__':
+    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)
+
+    dsn = os.environ['PSYCOPG_TEST_DSN']
+    import psycopg2
+
+    count = [0]
+    f = eventlet.spawn(fetch, dsn, 2, 1)
+    t = eventlet.spawn(tick, count, 2, 100)
+    f.wait()
+    assert count[0] > 100, count[0]
+    print("done")