00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037 '''
00038 MySQL Protocol Command Objects
00039 '''
00040
00041 import unittest
00042
00043 class CommandID(object):
00044 SLEEP = 0
00045 QUIT = 1
00046 INIT_DB = 2
00047 QUERY = 3
00048 FIELD_LIST = 4
00049 CREATE_DB = 5
00050 DROP_DB = 6
00051 REFRESH = 7
00052 SHUTDOWN = 8
00053 STATISTICS = 9
00054 PROCESS_INFO = 10
00055 CONNECT = 11
00056 PROCESS_KILL = 12
00057 DEBUG = 13
00058 PING = 14
00059 TIME = 15
00060 DELAYED_INSERT = 16
00061 CHANGE_USER = 17
00062 BINLOG_DUMP = 18
00063 TABLE_DUMP = 19
00064 CONNECT_OUT = 20
00065 REGISTER_SLAVE = 21
00066 STMT_PREPARE = 22
00067 STMT_EXECUTE = 23
00068 STMT_SEND_LONG_DATA = 24
00069 STMT_CLOSE = 25
00070 STMT_RESET = 26
00071 SET_OPTION = 27
00072 STMT_FETCH = 28
00073 DAEMON = 29
00074 END = 30
00075
00076 class Command(object):
00077 '''This class represents a command packet sent from the client.'''
00078
00079 def __init__(self, packed=None, command=CommandID.SLEEP, payload=''):
00080 if packed is None:
00081 self.command = command
00082 self.payload = payload
00083 else:
00084 self.command = ord(packed[0])
00085 self.payload = packed[1:]
00086
00087 def pack(self):
00088 return chr(self.command) + self.payload
00089
00090 def __str__(self):
00091 return '''Command
00092 command = %s
00093 payload = %s
00094 ''' % (self.command, self.payload)
00095
00096 class TestCommand(unittest.TestCase):
00097
00098 def testDefaultInit(self):
00099 command = Command()
00100 self.assertEqual(command.command, CommandID.SLEEP)
00101 self.assertEqual(command.payload, '')
00102
00103 def testKeywordInit(self):
00104 command = Command(command=CommandID.QUERY, payload='abc')
00105 self.assertEqual(command.command, CommandID.QUERY)
00106 self.assertEqual(command.payload, 'abc')
00107
00108 def testUnpackInit(self):
00109 command = Command('\x03abc')
00110 self.assertEqual(command.command, CommandID.QUERY)
00111 self.assertEqual(command.payload, 'abc')
00112
00113 def testPack(self):
00114 command = Command(Command(command=CommandID.QUERY, payload='abc').pack())
00115 self.assertEqual(command.command, CommandID.QUERY)
00116 self.assertEqual(command.payload, 'abc')
00117
00118 class QueryCommand(Command):
00119 def __init__(self, packed=None, query=''):
00120 super(QueryCommand, self).__init__(packed=packed, command=CommandID.QUERY,
00121 payload=query)
00122
00123 def __str__(self):
00124 return '''Command
00125 command = %s
00126 query = %s
00127 ''' % (self.command, self.payload)
00128
00129 class TestQueryCommand(unittest.TestCase):
00130
00131 def testDefaultInit(self):
00132 query = QueryCommand()
00133 self.assertEqual(query.command, CommandID.QUERY)
00134 self.assertEqual(query.payload, '')
00135
00136 def testKeywordInit(self):
00137 query = QueryCommand(query='abc')
00138 self.assertEqual(query.command, CommandID.QUERY)
00139 self.assertEqual(query.payload, 'abc')
00140
00141 def testUnpackInit(self):
00142 query = QueryCommand('\x03abc')
00143 self.assertEqual(query.command, CommandID.QUERY)
00144 self.assertEqual(query.payload, 'abc')
00145
00146 def testPack(self):
00147 query = QueryCommand(QueryCommand(query='abc').pack())
00148 self.assertEqual(query.command, CommandID.QUERY)
00149 self.assertEqual(query.payload, 'abc')
00150
00151 if __name__ == '__main__':
00152 unittest.main()