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

Notify object exposed in the extensions module.

Browse Source
This commit is contained in:
Daniele Varrazzo 2010-10-15 23:09:47 +01:00
parent 8db9c7085d
commit 4ec298e112
3 changed files with 25 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
lib/extensions.py
View File

@ -58,7 +58,7 @@ except:
from _psycopg import adapt, adapters, encodings, connection, cursor, lobject from _psycopg import adapt, adapters, encodings, connection, cursor, lobject
from _psycopg import string_types, binary_types, new_type, register_type from _psycopg import string_types, binary_types, new_type, register_type
from _psycopg import ISQLQuote from _psycopg import ISQLQuote, Notify
from _psycopg import QueryCanceledError, TransactionRollbackError from _psycopg import QueryCanceledError, TransactionRollbackError

1
psycopg/psycopgmodule.c
View File

@ -822,6 +822,7 @@ init_psycopg(void)
PyModule_AddObject(module, "connection", (PyObject*)&connectionType); PyModule_AddObject(module, "connection", (PyObject*)&connectionType);
PyModule_AddObject(module, "cursor", (PyObject*)&cursorType); PyModule_AddObject(module, "cursor", (PyObject*)&cursorType);
PyModule_AddObject(module, "ISQLQuote", (PyObject*)&isqlquoteType); PyModule_AddObject(module, "ISQLQuote", (PyObject*)&isqlquoteType);
PyModule_AddObject(module, "Notify", (PyObject*)&NotifyType);
#ifdef PSYCOPG_EXTENSIONS #ifdef PSYCOPG_EXTENSIONS
PyModule_AddObject(module, "lobject", (PyObject*)&lobjectType); PyModule_AddObject(module, "lobject", (PyObject*)&lobjectType);
#endif #endif

23
tests/test_notify.py
View File

@ -105,6 +105,14 @@ conn.close()
self.assertEqual(pid, self.conn.notifies[0][0]) self.assertEqual(pid, self.conn.notifies[0][0])
self.assertEqual('foo', self.conn.notifies[0][1]) self.assertEqual('foo', self.conn.notifies[0][1])
def test_notify_object(self):
self.autocommit(self.conn)
self.listen('foo')
self.notify('foo').communicate()
self.conn.poll()
notify = self.conn.notifies[0]
self.assert_(isinstance(notify, psycopg2.extensions.Notify))
def test_notify_attributes(self): def test_notify_attributes(self):
self.autocommit(self.conn) self.autocommit(self.conn)
self.listen('foo') self.listen('foo')
@ -131,6 +139,21 @@ conn.close()
self.assertEqual('foo', notify.channel) self.assertEqual('foo', notify.channel)
self.assertEqual('Hello, world!', notify.payload) self.assertEqual('Hello, world!', notify.payload)
def test_notify_init(self):
n = psycopg2.extensions.Notify(10, 'foo')
self.assertEqual(10, n.pid)
self.assertEqual('foo', n.channel)
self.assertEqual(None, n.payload)
(pid, channel) = n
self.assertEqual((pid, channel), (10, 'foo'))
n = psycopg2.extensions.Notify(42, 'bar', 'baz')
self.assertEqual(42, n.pid)
self.assertEqual('bar', n.channel)
self.assertEqual('baz', n.payload)
(pid, channel) = n
self.assertEqual((pid, channel), (42, 'bar'))
def test_suite(): def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__) return unittest.TestLoader().loadTestsFromName(__name__)