psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-08-01 19:00:09 +03:00
Code Issues Packages Projects Releases Wiki Activity

the test should not run in green case

Browse Source
This commit is contained in:
Richard Bann 2019-06-18 14:52:57 +02:00
parent 7ece08ac7a
commit e83ae9d69f
1 changed files with 5 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
tests/test_connection.py
View File

@ -45,7 +45,8 @@ from psycopg2 import extensions as ext
from .testutils import ( from .testutils import (
PY2, unittest, skip_if_no_superuser, skip_before_postgres, PY2, unittest, skip_if_no_superuser, skip_before_postgres,
skip_after_postgres, skip_before_libpq, skip_after_libpq, skip_after_postgres, skip_before_libpq, skip_after_libpq,
ConnectingTestCase, skip_if_tpc_disabled, skip_if_windows, slow) ConnectingTestCase, skip_if_tpc_disabled, skip_if_windows, slow,
skip_if_green)
from .testconfig import dbhost, dsn, dbname from .testconfig import dbhost, dsn, dbname
@ -411,12 +412,13 @@ t.join()
shutil.rmtree(dir, ignore_errors=True) shutil.rmtree(dir, ignore_errors=True)
@slow @slow
@skip_if_green
def test_handles_keyboardinterrupt(self): def test_handles_keyboardinterrupt(self):
def conn(queue): def conn(queue):
host = "10.255.255.1" # will timeout host = "10.255.255.1" # will timeout
queue.put(1) queue.put(1)
try: try:
self.connect(host=host, password="x", connect_timeout=1) self.connect(host=host, password="x", connect_timeout=0.5)
except KeyboardInterrupt: except KeyboardInterrupt:
queue.put("KeyboardInterrupt") queue.put("KeyboardInterrupt")
except psycopg2.OperationalError: except psycopg2.OperationalError:
@ -426,7 +428,7 @@ t.join()
process = multiprocessing.Process(target=conn, args=(queue,)) process = multiprocessing.Process(target=conn, args=(queue,))
process.start() process.start()
queue.get() queue.get()
time.sleep(0.9) time.sleep(0.3)
os.kill(process.pid, signal.SIGINT) os.kill(process.pid, signal.SIGINT)
process.join() process.join()
self.assertEqual(queue.get(), "KeyboardInterrupt") self.assertEqual(queue.get(), "KeyboardInterrupt")