Drizzled Public API Documentation

command.py
1 #!/usr/bin/env python
2 #
3 # Drizzle Client & Protocol Library
4 #
5 # Copyright (C) 2008 Eric Day (eday@oddments.org)
6 # All rights reserved.
7 #
8 # Redistribution and use in source and binary forms, with or without
9 # modification, are permitted provided that the following conditions are
10 # met:
11 #
12 # * Redistributions of source code must retain the above copyright
13 # notice, this list of conditions and the following disclaimer.
14 #
15 # * Redistributions in binary form must reproduce the above
16 # copyright notice, this list of conditions and the following disclaimer
17 # in the documentation and/or other materials provided with the
18 # distribution.
19 #
20 # * The names of its contributors may not be used to endorse or
21 # promote products derived from this software without specific prior
22 # written permission.
23 #
24 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
25 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
26 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
27 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
28 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
29 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
30 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
31 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
32 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
33 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
34 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
35 #
36 
37 '''
38 MySQL Protocol Command Objects
39 '''
40 
41 import unittest
42 
43 class CommandID(object):
44  SLEEP = 0
45  QUIT = 1
46  INIT_DB = 2
47  QUERY = 3
48  FIELD_LIST = 4
49  CREATE_DB = 5
50  DROP_DB = 6
51  REFRESH = 7
52  SHUTDOWN = 8
53  STATISTICS = 9
54  PROCESS_INFO = 10
55  CONNECT = 11
56  PROCESS_KILL = 12
57  DEBUG = 13
58  PING = 14
59  TIME = 15
60  DELAYED_INSERT = 16
61  CHANGE_USER = 17
62  BINLOG_DUMP = 18
63  TABLE_DUMP = 19
64  CONNECT_OUT = 20
65  REGISTER_SLAVE = 21
66  STMT_PREPARE = 22
67  STMT_EXECUTE = 23
68  STMT_SEND_LONG_DATA = 24
69  STMT_CLOSE = 25
70  STMT_RESET = 26
71  SET_OPTION = 27
72  STMT_FETCH = 28
73  DAEMON = 29
74  END = 30
75 
76 class Command(object):
77  '''This class represents a command packet sent from the client.'''
78 
79  def __init__(self, packed=None, command=CommandID.SLEEP, payload=''):
80  if packed is None:
81  self.command = command
82  self.payload = payload
83  else:
84  self.command = ord(packed[0])
85  self.payload = packed[1:]
86 
87  def pack(self):
88  return chr(self.command) + self.payload
89 
90  def __str__(self):
91  return '''Command
92  command = %s
93  payload = %s
94 ''' % (self.command, self.payload)
95 
96 class TestCommand(unittest.TestCase):
97 
98  def testDefaultInit(self):
99  command = Command()
100  self.assertEqual(command.command, CommandID.SLEEP)
101  self.assertEqual(command.payload, '')
102 
103  def testKeywordInit(self):
104  command = Command(command=CommandID.QUERY, payload='abc')
105  self.assertEqual(command.command, CommandID.QUERY)
106  self.assertEqual(command.payload, 'abc')
107 
108  def testUnpackInit(self):
109  command = Command('\x03abc')
110  self.assertEqual(command.command, CommandID.QUERY)
111  self.assertEqual(command.payload, 'abc')
112 
113  def testPack(self):
114  command = Command(Command(command=CommandID.QUERY, payload='abc').pack())
115  self.assertEqual(command.command, CommandID.QUERY)
116  self.assertEqual(command.payload, 'abc')
117 
119  def __init__(self, packed=None, query=''):
120  super(QueryCommand, self).__init__(packed=packed, command=CommandID.QUERY,
121  payload=query)
122 
123  def __str__(self):
124  return '''Command
125  command = %s
126  query = %s
127 ''' % (self.command, self.payload)
128 
129 class TestQueryCommand(unittest.TestCase):
130 
131  def testDefaultInit(self):
132  query = QueryCommand()
133  self.assertEqual(query.command, CommandID.QUERY)
134  self.assertEqual(query.payload, '')
135 
136  def testKeywordInit(self):
137  query = QueryCommand(query='abc')
138  self.assertEqual(query.command, CommandID.QUERY)
139  self.assertEqual(query.payload, 'abc')
140 
141  def testUnpackInit(self):
142  query = QueryCommand('\x03abc')
143  self.assertEqual(query.command, CommandID.QUERY)
144  self.assertEqual(query.payload, 'abc')
145 
146  def testPack(self):
147  query = QueryCommand(QueryCommand(query='abc').pack())
148  self.assertEqual(query.command, CommandID.QUERY)
149  self.assertEqual(query.payload, 'abc')
150 
151 if __name__ == '__main__':
152  unittest.main()