38 MySQL Protocol Result Objects
48 '''This class represents an OK result packet sent from the server.'''
50 def __init__(self, packed=None, affected_rows=0, insert_id=0, status=0,
51 warning_count=0, message=
'', version_40=
False):
53 self.affected_rows = affected_rows
54 self.insert_id = insert_id
56 self.message = message
57 self.version_40 = version_40
58 if version_40
is False:
59 self.warning_count = warning_count
61 self.version_40 = version_40
62 if ord(packed[0]) != 0:
63 raise BadFieldCount(
'Expected 0, received ' + str(ord(packed[0])))
64 self.affected_rows = ord(packed[1])
65 self.insert_id = ord(packed[2])
66 if version_40
is True:
71 data = struct.unpack(
'<H', packed[3:5])
73 self.message = packed[5:]
75 data = struct.unpack(
'<HH', packed[3:7])
77 self.warning_count = data[1]
78 self.message = packed[7:]
81 if self.version_40
is True:
88 ''' % (self.affected_rows, self.insert_id, self.status, self.message,
98 ''' % (self.affected_rows, self.insert_id, self.status, self.warning_count,
99 self.message, self.version_40)
103 def testDefaultInit(self):
105 self.assertEqual(result.affected_rows, 0)
106 self.assertEqual(result.insert_id, 0)
107 self.assertEqual(result.status, 0)
108 self.assertEqual(result.warning_count, 0)
109 self.assertEqual(result.message,
'')
110 self.assertEqual(result.version_40,
False)
113 def testDefaultInit40(self):
115 self.assertEqual(result.affected_rows, 0)
116 self.assertEqual(result.insert_id, 0)
117 self.assertEqual(result.status, 0)
118 self.assertEqual(result.message,
'')
119 self.assertEqual(result.version_40,
True)
122 def testKeywordInit(self):
123 result =
OkResult(affected_rows=3, insert_id=5, status=2,
124 warning_count=7, message=
'test', version_40=
False)
125 self.assertEqual(result.affected_rows, 3)
126 self.assertEqual(result.insert_id, 5)
127 self.assertEqual(result.status, 2)
128 self.assertEqual(result.warning_count, 7)
129 self.assertEqual(result.message,
'test')
130 self.assertEqual(result.version_40,
False)
132 def testUnpackInit(self):
133 data = struct.pack(
'BBB', 0, 3, 5)
134 data += struct.pack(
'<HH', 2, 7)
138 self.assertEqual(result.affected_rows, 3)
139 self.assertEqual(result.insert_id, 5)
140 self.assertEqual(result.status, 2)
141 self.assertEqual(result.warning_count, 7)
142 self.assertEqual(result.message,
'test')
143 self.assertEqual(result.version_40,
False)
146 def testUnpackInit40(self):
147 data = struct.pack(
'BBB', 0, 3, 5)
148 data += struct.pack(
'<H', 2)
151 result =
OkResult(data, version_40=
True)
152 self.assertEqual(result.affected_rows, 3)
153 self.assertEqual(result.insert_id, 5)
154 self.assertEqual(result.status, 2)
155 self.assertEqual(result.message,
'test')
156 self.assertEqual(result.version_40,
True)
160 '''This class represents an error result packet sent from the server.'''
162 def __init__(self, packed=None, error_code=0, sqlstate_marker='#',
163 sqlstate=
'XXXXX', message=
'', version_40=
False):
168 if version_40
is False:
173 if ord(packed[0]) != 255:
174 raise BadFieldCount(
'Expected 255, received ' + str(ord(packed[0])))
175 data = struct.unpack(
'<H', packed[1:3])
177 if version_40
is True:
186 return '''ErrorResult
192 return '''ErrorResult
203 def testDefaultInit(self):
205 self.assertEqual(result.error_code, 0)
206 self.assertEqual(result.sqlstate_marker,
'#')
207 self.assertEqual(result.sqlstate,
'XXXXX')
208 self.assertEqual(result.message,
'')
209 self.assertEqual(result.version_40,
False)
212 def testDefaultInit40(self):
214 self.assertEqual(result.error_code, 0)
215 self.assertEqual(result.message,
'')
216 self.assertEqual(result.version_40,
True)
219 def testKeywordInit(self):
220 result =
ErrorResult(error_code=3, sqlstate_marker=
'@', sqlstate=
'ABCDE',
221 message=
'test', version_40=
False)
222 self.assertEqual(result.error_code, 3)
223 self.assertEqual(result.sqlstate_marker,
'@')
224 self.assertEqual(result.sqlstate,
'ABCDE')
225 self.assertEqual(result.message,
'test')
226 self.assertEqual(result.version_40,
False)
229 def testUnpackInit(self):
231 data += struct.pack(
'<H', 1234)
236 self.assertEqual(result.error_code, 1234)
237 self.assertEqual(result.sqlstate_marker,
'#')
238 self.assertEqual(result.sqlstate,
'ABCDE')
239 self.assertEqual(result.message,
'test')
240 self.assertEqual(result.version_40,
False)
243 def testUnpackInit40(self):
245 data += struct.pack(
'<H', 1234)
249 self.assertEqual(result.error_code, 1234)
250 self.assertEqual(result.message,
'test')
251 self.assertEqual(result.version_40,
True)
255 '''This class represents an EOF result packet sent from the server.'''
257 def __init__(self, packed=None, warning_count=0, status=0, version_40=False):
265 if ord(packed[0]) != 254:
266 raise BadFieldCount(
'Expected 254, received ' + str(ord(packed[0])))
267 if version_40
is False:
268 data = struct.unpack(
'<HH', packed[1:])
286 def testDefaultInit(self):
288 self.assertEqual(result.warning_count, 0)
289 self.assertEqual(result.status, 0)
290 self.assertEqual(result.version_40,
False)
293 def testDefaultInit40(self):
295 self.assertEqual(result.version_40,
True)
298 def testKeywordInit(self):
299 result =
EofResult(warning_count=3, status=5, version_40=
False)
300 self.assertEqual(result.warning_count, 3)
301 self.assertEqual(result.status, 5)
302 self.assertEqual(result.version_40,
False)
305 def testUnpackInit(self):
307 data += struct.pack(
'<HH', 3, 5)
310 self.assertEqual(result.warning_count, 3)
311 self.assertEqual(result.status, 5)
312 self.assertEqual(result.version_40,
False)
315 def testUnpackInit40(self):
316 result =
EofResult(chr(254), version_40=
True)
317 self.assertEqual(result.version_40,
True)
321 '''This class represents an count result packet sent from the server.'''
323 def __init__(self, packed=None, count=0):
327 self.
count = ord(packed[0])
329 raise BadFieldCount(
'Expected 1-253, received ' + str(ord(packed[0])))
332 return '''CountResult
338 def testDefaultInit(self):
340 self.assertEqual(result.count, 0)
343 def testKeywordInit(self):
345 self.assertEqual(result.count, 3)
348 def testUnpackInit(self):
350 self.assertEqual(result.count, 3)
353 def create_result(packed, version_40=False):
354 '''This function creates the appropriate result object instance depending on
356 count = ord(packed[0])
358 return OkResult(packed, version_40=version_40)
360 return EofResult(packed, version_40=version_40)
365 if __name__ ==
'__main__':