psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-02-12 07:10:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

subprocess test function moved into a module

Browse Source 
It won't work on windows if it's in the script: failing with errors
such as:

    AttributeError: 'module' object has no attribute 'process'

or:

    Can't get attribute 'process' on <module '__main__' (built-in)>
This commit is contained in:
Daniele Varrazzo 2019-03-17 23:55:04 +00:00
parent 7c5afd6977
commit 62a078fe0c
1 changed files with 29 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
tests/test_connection.py
View File

@ -22,14 +22,16 @@
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public # FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
# License for more details. # License for more details.
import ctypes
import gc import gc
import os import os
import re import re
import subprocess as sp
import sys import sys
import threading
import time import time
import ctypes
import shutil
import tempfile
import threading
import subprocess as sp
from collections import deque from collections import deque
from operator import attrgetter from operator import attrgetter
from weakref import ref from weakref import ref
@ -359,10 +361,11 @@ class ConnectionTests(ConnectingTestCase):
@slow @slow
def test_multiprocess_close(self): def test_multiprocess_close(self):
script = ("""\ dir = tempfile.mkdtemp()
try:
with open(os.path.join(dir, "mptest.py"), 'w') as f:
f.write("""\
import time import time
import threading
import multiprocessing
import psycopg2 import psycopg2
def thread(): def thread():
@ -374,16 +377,28 @@ def thread():
def process(): def process():
time.sleep(0.2) time.sleep(0.2)
t = threading.Thread(target=thread, name='mythread')
t.start()
time.sleep(0.2)
multiprocessing.Process(target=process, name='myprocess').start()
t.join()
""" % {'dsn': dsn}) """ % {'dsn': dsn})
out = sp.check_output([sys.executable, '-c', script], stderr=sp.STDOUT) script = ("""\
self.assertEqual(out, b'', out.decode('ascii')) import sys
sys.path.insert(0, %(dir)r)
import time
import threading
import multiprocessing
import mptest
t = threading.Thread(target=mptest.thread, name='mythread')
t.start()
time.sleep(0.2)
multiprocessing.Process(target=mptest.process, name='myprocess').start()
t.join()
""" % {'dir': dir})
out = sp.check_output(
[sys.executable, '-c', script], stderr=sp.STDOUT)
self.assertEqual(out, b'', out)
finally:
shutil.rmtree(dir, ignore_errors=True)
class ParseDsnTestCase(ConnectingTestCase): class ParseDsnTestCase(ConnectingTestCase):