psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-11-26 02:43:43 +03:00
Code Issues Packages Projects Releases Wiki Activity

Check the correctness of the PID in the notify tests.

Browse Source
This commit is contained in:
Daniele Varrazzo 2010-10-15 17:14:47 +01:00
parent d9e49e940a
commit e651308287
1 changed files with 15 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
tests/test_notify.py
View File

@ -7,7 +7,7 @@ from psycopg2 import extensions
import time
import select
import signal
from subprocess import Popen
from subprocess import Popen, PIPE
import sys
if sys.version_info < (3,):
@ -43,6 +43,7 @@ import psycopg2
import psycopg2.extensions
conn = psycopg2.connect(%(dsn)r)
conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
print conn.get_backend_pid()
curs = conn.cursor()
curs.execute("NOTIFY " %(name)r)
curs.close()
@ -50,22 +51,24 @@ conn.close()
"""
% { 'dsn': tests.dsn, 'sec': sec, 'name': name})
return Popen([sys.executable, '-c', script])
return Popen([sys.executable, '-c', script], stdout=PIPE)
def test_notifies_received_on_poll(self):
self.autocommit(self.conn)
self.listen('foo')
self.notify('foo', 1);
proc = self.notify('foo', 1)
t0 = time.time()
ready = select.select([self.conn], [], [], 5)
t1 = time.time()
self.assert_(0.99 < t1 - t0 < 4, t1 - t0)
pid = int(proc.communicate()[0])
self.assertEqual(0, len(self.conn.notifies))
self.assertEqual(extensions.POLL_OK, self.conn.poll())
self.assertEqual(1, len(self.conn.notifies))
self.assertEqual(pid, self.conn.notifies[0][0])
self.assertEqual('foo', self.conn.notifies[0][1])
def test_many_notifies(self):
@ -73,23 +76,27 @@ conn.close()
for name in ['foo', 'bar', 'baz']:
self.listen(name)
pids = {}
for name in ['foo', 'bar', 'baz', 'qux']:
self.notify(name).wait()
pids[name] = int(self.notify(name).communicate()[0])
self.assertEqual(0, len(self.conn.notifies))
self.assertEqual(extensions.POLL_OK, self.conn.poll())
self.assertEqual(3, len(self.conn.notifies))
names = [n[1] for n in self.conn.notifies]
for name in ['foo', 'bar', 'baz']:
self.assert_(name in names, name)
names = dict.fromkeys(['foo', 'bar', 'baz'])
for (pid, name) in self.conn.notifies:
self.assertEqual(pids[name], pid)
names.pop(name) # raise if name found twice
def test_notifies_received_on_execute(self):
self.autocommit(self.conn)
self.listen('foo')
self.notify('foo').wait()
pid = int(self.notify('foo').communicate()[0])
self.assertEqual(0, len(self.conn.notifies))
self.conn.cursor().execute('select 1;')
self.assertEqual(1, len(self.conn.notifies))
self.assertEqual(pid, self.conn.notifies[0][0])
self.assertEqual('foo', self.conn.notifies[0][1])
def test_suite():