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
|
#!/usr/bin/env python -O
# -*- coding: utf-8 -*-
import unittest
from datetime import datetime
import mariadb
import sys, traceback
from test.base_test import create_connection
class TestException(unittest.TestCase):
def setUp(self):
self.connection = create_connection()
def tearDown(self):
del self.connection
def test_exception(self):
cursor = self.connection.cursor()
try:
cursor.execute("WRONG QUERY")
except mariadb.ProgrammingError as err:
self.assertEqual(err.sqlstate, "42000")
self.assertEqual(err.errno, 1064)
self.assertTrue(err.errmsg.find("You have an error "
"in your SQL syntax") > -1)
if mariadb._have_asan:
tb = sys.exc_info()[2]
traceback.clear_frames(tb)
pass
del cursor
def test_db_unknown_exception(self):
try:
create_connection({"database": "unknown"})
except mariadb.ProgrammingError as err:
self.assertEqual(err.sqlstate, "42000")
self.assertEqual(err.errno, 1049)
self.assertTrue(err.errmsg.find("Unknown database 'unknown'") > -1)
if mariadb._have_asan:
tb = sys.exc_info()[2]
traceback.clear_frames(tb)
pass
@unittest.skip("requires network access")
def test_conn_timeout_exception(self):
start = datetime.today()
try:
create_connection({"connect_timeout": 1, "host": "8.8.8.8"})
except mariadb.OperationalError as err:
self.assertEqual(err.sqlstate, "HY000")
self.assertEqual(err.errno, 2002)
self.assertTrue(err.errmsg.find("server on '8.8.8.8'") > -1)
end = datetime.today()
difference = end - start
self.assertEqual(difference.days, 0)
self.assertTrue(difference.seconds, 1)
if mariadb._have_asan:
tb = sys.exc_info()[2]
traceback.clear_frames(tb)
pass
if __name__ == '__main__':
unittest.main()
|
