38 MySQL Protocol Packet Objects
48 '''This class represents a packet header.'''
50 def __init__(self, packed=None, size=0, sequence=0):
53 self.sequence = sequence
55 data = struct.unpack(
'4B', packed)
56 self.size = data[0] | (data[1] << 8) | (data[2] << 16)
57 self.sequence = data[3]
63 return struct.pack(
'4B',
65 (self.size >> 8) & 0xFF,
66 (self.size >> 16) & 0xFF,
70 if self.size >= 16777216:
78 ''' % (self.size, self.sequence)
82 def testDefaultInit(self):
84 self.assertEqual(packet.size, 0)
85 self.assertEqual(packet.sequence, 0)
88 def testKeywordInit(self):
89 packet =
Packet(size=1234, sequence=5)
90 self.assertEqual(packet.size, 1234)
91 self.assertEqual(packet.sequence, 5)
94 def testUnpackInit(self):
95 packet =
Packet(struct.pack(
'4B', 210, 4, 0, 5))
96 self.assertEqual(packet.size, 1234)
97 self.assertEqual(packet.sequence, 5)
101 self.assertEqual(packet.size, 1234)
103 def testPackRange(self):
104 for x
in range(0, 300):
106 self.assertEqual(packet.size, x)
107 self.assertEqual(packet.sequence, x % 256)
110 for x
in range(300, 16777216, 997):
112 self.assertEqual(packet.size, x)
113 self.assertEqual(packet.sequence, x % 256)
116 self.assertEqual(packet.size, 16777215)
117 self.assertEqual(packet.sequence, 0)
119 self.assertRaises(PacketException, Packet, size=16777216)
120 self.assertRaises(PacketException, Packet, size=16777217)
121 self.assertRaises(PacketException, Packet, size=4294967295)
122 self.assertRaises(PacketException, Packet, size=4294967296)
123 self.assertRaises(PacketException, Packet, size=4294967297)
125 def parse_row(count, data):
129 if ord(data[0]) == 251:
133 (size, packed_size) = parse_encoded_size(data)
134 row.append(data[packed_size:packed_size+size])
135 data = data[packed_size+size:]
141 def parse_encoded_size(data):
145 size = struct.unpack(
'<H', data[1:3])[0]
148 data = struct.unpack(
'<HB', data[1:4])
149 size = data[0] | (data[1] << 16)
152 data = struct.unpack(
'<II', data[1:9])
153 size = data[0] | (data[1] << 32)
156 raise BadSize(str(size))
158 return (size, packed_size)
160 if __name__ ==
'__main__':