1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
#!/usr/bin/env python -O
# -*- coding: utf-8 -*-
import unittest
import mariadb
from test.base_test import create_connection
class TestCA(unittest.TestCase):
def setUp(self):
self.connection = create_connection()
self.connection.autocommit = False
def tearDown(self):
del self.connection
def test_xid(self):
con = create_connection()
xid = con.xid(1, "foo", "bar")
self.assertEqual(xid, (1, "foo", "bar"))
# default for format_id is 1
xid = con.xid(0, "foo", "bar")
self.assertEqual(xid, (1, "foo", "bar"))
# parameter too long:
try:
xid = con.xid(0, "a" * 65, "bar")
except mariadb.ProgrammingError:
pass
try:
xid = con.xid(0, "foo", "b" * 65)
except mariadb.ProgrammingError:
pass
def test_tpc_begin(self):
con = create_connection()
xid = con.xid(0, "1234567890", "2345")
try:
con.tpc_begin(xid)
except mariadb.NotSupportedError:
pass
def test_tpc_commit(self):
con = create_connection()
xid = con.xid(0, "1234567891", "2345")
cursor = con.cursor()
cursor.execute("DROP TABLE IF EXISTS t1")
cursor.execute("CREATE TABLE t1 (a int)")
try:
con.tpc_begin(xid)
cursor.execute("INSERT INTO t1 VALUES (1),(2)")
cursor.close()
con.tpc_commit()
finally:
con.close()
def test_tpc_rollback_without_prepare(self):
con = create_connection()
try:
xid = con.xid(0, "1234567892", "2345")
con.tpc_begin(xid)
cursor = con.cursor()
cursor.execute("SELECT 1")
cursor.close()
con.tpc_rollback()
finally:
con.close()
def test_tpc_commit_with_prepare(self):
con = create_connection()
try:
xid = con.xid(0, "1234567893", "2345")
con.tpc_begin(xid)
cursor = con.cursor()
cursor.execute("SELECT 1")
cursor.close()
con.tpc_prepare()
con.tpc_commit()
finally:
con.close()
def test_tpc_rollback_with_prepare(self):
con = create_connection()
try:
xid = con.xid(0, "1234567894", "2345")
con.tpc_begin(xid)
cursor = con.cursor()
cursor.execute("SELECT 1")
cursor.close()
con.tpc_prepare()
con.tpc_rollback()
finally:
con.close()
def test_tpc_begin_in_transaction_fails(self):
con = create_connection()
try:
xid = con.xid(0, "1234567895", "2345")
cursor = con.cursor()
cursor.execute("BEGIN")
cursor.execute("SELECT 1")
cursor.close()
self.assertRaises(mariadb.IntegrityError,
con.tpc_begin, xid)
finally:
con.close()
def test_commit_in_tpc_fails(self):
con = create_connection()
try:
xid = con.xid(0, "1234567897", "2345")
con.tpc_begin(xid)
self.assertRaises(mariadb.ProgrammingError, con.commit)
finally:
con.close()
def test_rollback_in_tpc_fails(self):
# calling rollback() within a TPC transaction fails with
# ProgrammingError.
con = create_connection()
try:
xid = con.xid(0, "1234567898", "2345")
con.tpc_begin(xid)
self.assertRaises(mariadb.ProgrammingError, con.rollback)
finally:
con.close()
if __name__ == '__main__':
unittest.main()
|
