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
137
138
139
140
141
142
143
144
145
146
147
|
# -*- coding: utf-8 -*-
"""
Test user message hanler.
"""
import unittest
import pytest
from .helpers import config, mssqlconn
msgs = []
def user_msg_handler1(msgstate, severity, srvname, procname, line, msgtext):
global msgs
procname = procname.decode('ascii')
msgtext = msgtext.decode('ascii')
entry = ("msg_handler1: msgstate = %d, severity = %d, procname = '%s', "
"line = %d, msgtext = '%s'") % (msgstate, severity, procname, line, msgtext)
msgs.append(entry)
def user_msg_handler2(msgstate, severity, srvname, procname, line, msgtext):
global msgs
procname = procname.decode('ascii')
msgtext = msgtext.decode('ascii')
entry = ("msg_handler2: msgstate = %d, severity = %d, procname = '%s', "
"line = %d, msgtext = '%s'") % (msgstate, severity, procname, line, msgtext)
msgs.append(entry)
def wrong_signature_msg_handler():
pass
@pytest.mark.mssql_server_required
class TestUserMsgHandler(unittest.TestCase):
def test_basic_functionality(self):
cnx = mssqlconn()
try:
cnx.set_msghandler(user_msg_handler1)
msgs_before = len(msgs)
cnx.execute_non_query("USE master")
msgs_after = len(msgs)
delta = msgs_after - msgs_before
self.assertEqual(delta, 1)
expect = ("msg_handler1: msgstate = 1, severity = 0, procname = ''"
", line = 1, msgtext = 'Changed database context to 'master'.'")
self.assertEqual(expect, msgs[msgs_after - 1])
finally:
cnx.close()
def test_set_handler_to_none(self):
cnx = mssqlconn()
try:
cnx.set_msghandler(None)
msgs_before = len(msgs)
cnx.execute_non_query("USE master")
msgs_after = len(msgs)
delta = msgs_after - msgs_before
self.assertEqual(delta, 0)
finally:
cnx.close()
def test_change_handler(self):
cnx = mssqlconn()
try:
cnx.set_msghandler(user_msg_handler1)
msgs_before = len(msgs)
cnx.execute_non_query("USE master")
msgs_after = len(msgs)
delta = msgs_after - msgs_before
self.assertEqual(delta, 1)
expect = ("msg_handler1: msgstate = 1, severity = 0, procname = ''"
", line = 1, msgtext = 'Changed database context to 'master'.'")
self.assertEqual(expect, msgs[msgs_after - 1])
cnx.set_msghandler(user_msg_handler2)
msgs_before = len(msgs)
cnx.execute_non_query("USE %s" % config.database)
msgs_after = len(msgs)
delta = msgs_after - msgs_before
self.assertEqual(delta, 1)
expect = ("msg_handler2: msgstate = 1, severity = 0, procname = ''"
", line = 1, msgtext = 'Changed database context to '%s'.'") % config.database
self.assertEqual(expect, msgs[msgs_after - 1])
finally:
cnx.close()
def test_per_conn_handlers(self):
cnx1 = mssqlconn()
cnx2 = mssqlconn()
try:
cnx1.set_msghandler(user_msg_handler1)
msgs_before = len(msgs)
cnx1.execute_non_query("USE master")
msgs_after = len(msgs)
delta = msgs_after - msgs_before
self.assertEqual(delta, 1)
expect = ("msg_handler1: msgstate = 1, severity = 0, procname = ''"
", line = 1, msgtext = 'Changed database context to 'master'.'")
self.assertEqual(expect, msgs[msgs_after - 1])
cnx2.set_msghandler(user_msg_handler2)
msgs_before = len(msgs)
cnx2.execute_non_query("USE %s" % config.database)
msgs_after = len(msgs)
delta = msgs_after - msgs_before
self.assertEqual(delta, 1)
expect = ("msg_handler2: msgstate = 1, severity = 0, procname = ''"
", line = 1, msgtext = 'Changed database context to '%s'.'") % config.database
self.assertEqual(expect, msgs[msgs_after - 1])
finally:
cnx1.close()
cnx2.close()
@staticmethod
def user_msg_handler3(msgstate, severity, srvname, procname, line, msgtext):
global msgs
procname = procname.decode('ascii')
msgtext = msgtext.decode('ascii')
entry = ("msg_handler3 called")
msgs.append(entry)
def test_static_method_handler(self):
cnx = mssqlconn()
try:
cnx.set_msghandler(self.user_msg_handler3)
msgs_before = len(msgs)
cnx.execute_non_query("USE master")
msgs_after = len(msgs)
delta = msgs_after - msgs_before
self.assertEqual(delta, 1)
expect = ("msg_handler3 called")
self.assertEqual(expect, msgs[msgs_after - 1])
finally:
cnx.close()
def test_wrong_signature_handler(self):
cnx = mssqlconn()
try:
cnx.set_msghandler(wrong_signature_msg_handler)
cnx.execute_non_query("USE master")
finally:
cnx.close()
|
