FM / tests

/opt/cloudlinux/venv/lib/python3.11/site-packages/psycopg2/tests/

/opt/cloudlinux/venv/lib/python3.11/site-packages/psycopg2/tests

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
dbapi20.py	32286	0644	edit dl rm
dbapi20_tpc.py	4143	0644	edit dl rm
testconfig.py	1249	0644	edit dl rm
testutils.py	11133	0644	edit dl rm
test_async.py	15729	0644	edit dl rm
test_bugX000.py	1678	0644	edit dl rm
test_bug_gc.py	1719	0644	edit dl rm
test_cancel.py	3705	0644	edit dl rm
test_connection.py	39595	0644	edit dl rm
test_copy.py	11864	0644	edit dl rm
test_cursor.py	21326	0644	edit dl rm
test_dates.py	24059	0644	edit dl rm
test_errcodes.py	2107	0644	edit dl rm
test_extras_dictcursor.py	17404	0644	edit dl rm
test_green.py	4198	0644	edit dl rm
test_lobject.py	15262	0644	edit dl rm
test_module.py	11123	0644	edit dl rm
test_notify.py	7137	0644	edit dl rm
test_psycopg2_dbapi20.py	1932	0644	edit dl rm
test_quote.py	8252	0644	edit dl rm
test_transaction.py	9214	0644	edit dl rm
test_types_basic.py	18623	0644	edit dl rm
test_types_extras.py	62578	0644	edit dl rm
test_with.py	7365	0644	edit dl rm
__init__.py	2966	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/psycopg2/tests/dbapi20_tpc.py (4143B)

""" Python DB API 2.0 driver Two Phase Commit compliance test suite.

"""

import unittest

class TwoPhaseCommitTests(unittest.TestCase):

driver = None

def connect(self):
        """Make a database connection."""
        raise NotImplementedError

_last_id = 0
    _global_id_prefix = "dbapi20_tpc:"

def make_xid(self, con):
        id = TwoPhaseCommitTests._last_id
        TwoPhaseCommitTests._last_id += 1
        return con.xid(42, "%s%d" % (self._global_id_prefix, id), "qualifier")

def test_xid(self):
        con = self.connect()
        try:
            xid = con.xid(42, "global", "bqual")
        except self.driver.NotSupportedError:
            self.fail("Driver does not support transaction IDs.")

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:
            xid = self.make_xid(con)
            try:
                con.tpc_begin(xid)
            except self.driver.NotSupportedError:
                self.fail("Driver does not support tpc_begin()")
        finally:
            con.close()

def test_tpc_commit_without_prepare(self):
        con = self.connect()
        try:
            xid = self.make_xid(con)
            con.tpc_begin(xid)
            cursor = con.cursor()
            cursor.execute("SELECT 1")
            con.tpc_commit()
        finally:
            con.close()

def test_tpc_rollback_without_prepare(self):
        con = self.connect()
        try:
            xid = self.make_xid(con)
            con.tpc_begin(xid)
            cursor = con.cursor()
            cursor.execute("SELECT 1")
            con.tpc_rollback()
        finally:
            con.close()

def test_tpc_commit_with_prepare(self):
        con = self.connect()
        try:
            xid = self.make_xid(con)
            con.tpc_begin(xid)
            cursor = con.cursor()
            cursor.execute("SELECT 1")
            con.tpc_prepare()
            con.tpc_commit()
        finally:
            con.close()

def test_tpc_rollback_with_prepare(self):
        con = self.connect()
        try:
            xid = self.make_xid(con)
            con.tpc_begin(xid)
            cursor = con.cursor()
            cursor.execute("SELECT 1")
            con.tpc_prepare()
            con.tpc_rollback()
        finally:
            con.close()

def test_tpc_begin_in_transaction_fails(self):
        con = self.connect()
        try:
            xid = self.make_xid(con)

cursor = con.cursor()
            cursor.execute("SELECT 1")
            self.assertRaises(self.driver.ProgrammingError,
                              con.tpc_begin, xid)
        finally:
            con.close()

def test_tpc_begin_in_tpc_transaction_fails(self):
        con = self.connect()
        try:
            xid = self.make_xid(con)

def test_commit_in_tpc_fails(self):
        # calling commit() within a TPC transaction fails with
        # ProgrammingError.
        con = self.connect()
        try:
            xid = self.make_xid(con)
            con.tpc_begin(xid)

self.assertRaises(self.driver.ProgrammingError, con.commit)
        finally:
            con.close()

def test_rollback_in_tpc_fails(self):
        # calling rollback() within a TPC transaction fails with
        # ProgrammingError.
        con = self.connect()
        try:
            xid = self.make_xid(con)
            con.tpc_begin(xid)

self.assertRaises(self.driver.ProgrammingError, con.rollback)
        finally:
            con.close()

""" Python DB API 2.0 driver Two Phase Commit compliance test suite. """ import unittest class TwoPhaseCommitTests(unittest.TestCase): driver = None def connect(self): """Make a database connection.""" raise NotImplementedError _last_id = 0 _global_id_prefix = "dbapi20_tpc:" def make_xid(self, con): id = TwoPhaseCommitTests._last_id TwoPhaseCommitTests._last_id += 1 return con.xid(42, "%s%d" % (self._global_id_prefix, id), "qualifier") def test_xid(self): con = self.connect() try: xid = con.xid(42, "global", "bqual") except self.driver.NotSupportedError: self.fail("Driver does not support transaction IDs.") 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: xid = self.make_xid(con) try: con.tpc_begin(xid) except self.driver.NotSupportedError: self.fail("Driver does not support tpc_begin()") finally: con.close() def test_tpc_commit_without_prepare(self): con = self.connect() try: xid = self.make_xid(con) con.tpc_begin(xid) cursor = con.cursor() cursor.execute("SELECT 1") con.tpc_commit() finally: con.close() def test_tpc_rollback_without_prepare(self): con = self.connect() try: xid = self.make_xid(con) con.tpc_begin(xid) cursor = con.cursor() cursor.execute("SELECT 1") con.tpc_rollback() finally: con.close() def test_tpc_commit_with_prepare(self): con = self.connect() try: xid = self.make_xid(con) con.tpc_begin(xid) cursor = con.cursor() cursor.execute("SELECT 1") con.tpc_prepare() con.tpc_commit() finally: con.close() def test_tpc_rollback_with_prepare(self): con = self.connect() try: xid = self.make_xid(con) con.tpc_begin(xid) cursor = con.cursor() cursor.execute("SELECT 1") con.tpc_prepare() con.tpc_rollback() finally: con.close() def test_tpc_begin_in_transaction_fails(self): con = self.connect() try: xid = self.make_xid(con) cursor = con.cursor() cursor.execute("SELECT 1") self.assertRaises(self.driver.ProgrammingError, con.tpc_begin, xid) finally: con.close() def test_tpc_begin_in_tpc_transaction_fails(self): con = self.connect() try: xid = self.make_xid(con) cursor = con.cursor() cursor.execute("SELECT 1") self.assertRaises(self.driver.ProgrammingError, con.tpc_begin, xid) finally: con.close() def test_commit_in_tpc_fails(self): # calling commit() within a TPC transaction fails with # ProgrammingError. con = self.connect() try: xid = self.make_xid(con) con.tpc_begin(xid) self.assertRaises(self.driver.ProgrammingError, con.commit) finally: con.close() def test_rollback_in_tpc_fails(self): # calling rollback() within a TPC transaction fails with # ProgrammingError. con = self.connect() try: xid = self.make_xid(con) con.tpc_begin(xid) self.assertRaises(self.driver.ProgrammingError, con.rollback) finally: con.close()