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

hook up two phase commit tests.

Browse Source 
By James Henstridge on 2008-07-24.

Merged from lp:~jamesh/psycopg/two-phase-commit/revision/359
This commit is contained in:
Daniele Varrazzo 2010-10-06 01:39:58 +01:00
parent 7a9d678050
commit 8bfd34faf2
4 changed files with 24 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
ChangeLog
View File

@ -14,6 +14,9 @@
2008-07-24 James Henstridge <james@jamesh.id.au>
* tests/test_psycopg2_dbapi20.py (Psycopg2TPCTests): hook up two
phase commit tests.
* psycopg/xid_type.c (xid_len, xid_getitem): implement sequence
behaviour, as required for transaction IDs.
(XidType): There is no point in allowing subclasses of Xid.

2
psycopg/xid_type.c
View File

@ -75,7 +75,7 @@ xid_init(XidObject *self, PyObject *args, PyObject *kwargs)
&format_id, &gtrid, &bqual))
return -1;
if (format_id < 0 || format_id >= 0xffffffff) {
if (format_id < 0 || format_id > 0x7fffffff) {
PyErr_SetString(PyExc_ValueError,
"format_id must be a non-negative 32-bit integer");
return -1;

16
tests/dbapi20_tpc.py
View File

@ -5,7 +5,7 @@
import unittest
class TwoPhaseCommitTest(unittest.TestCase):
class TwoPhaseCommitTests(unittest.TestCase):
driver = None
@ -17,8 +17,8 @@ class TwoPhaseCommitTest(unittest.TestCase):
_global_id_prefix = "dbapi20_tpc:"
def make_xid(self, con):
id = TwoPhaseCommitTest._last_id
TwoPhaseCommitTest._last_id += 1
id = TwoPhaseCommitTests._last_id
TwoPhaseCommitTests._last_id += 1
return con.xid(42, "%s%d" % (self._global_id_prefix, id), "qualifier")
def test_xid(self):
@ -28,10 +28,16 @@ class TwoPhaseCommitTest(unittest.TestCase):
except self.driver.NotSupportedError:
self.fail("Driver does not support transaction IDs.")
self.assertEuqals(xid[0], 42)
self.assertEuqals(xid[1], "global")
self.assertEquals(xid[0], 42)
self.assertEquals(xid[1], "global")
self.assertEquals(xid[2], "bqual")
# Try some extremes for the transaction ID:
xid = con.xid(0, "", "")
self.assertEquals(tuple(xid), (0, "", ""))
xid = con.xid(0x7fffffff, "a" * 64, "b" * 64)
self.assertEquals(tuple(xid), (0x7fffffff, "a" * 64, "b" * 64))
def test_tpc_begin(self):
con = self.connect()
try:

10
tests/test_psycopg2_dbapi20.py
View File

@ -1,11 +1,12 @@
#!/usr/bin/env python
import dbapi20
import dbapi20_tpc
import unittest
import psycopg2
import tests
class Psycopg2TestCase(dbapi20.DatabaseAPI20Test):
class Psycopg2Tests(dbapi20.DatabaseAPI20Test):
driver = psycopg2
connect_args = ()
connect_kw_args = {'dsn': tests.dsn}
@ -21,6 +22,13 @@ class Psycopg2TestCase(dbapi20.DatabaseAPI20Test):
pass
class Psycopg2TPCTests(dbapi20_tpc.TwoPhaseCommitTests):
driver = psycopg2
def connect(self):
return psycopg2.connect(dsn=tests.dsn)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)