← Back to branch summary

~ubuntu-branches/ubuntu/trusty/drizzle/trusty

« back to all changes in this revision

Viewing changes to plugin/mysql_protocol/prototest/prototest/mysql/command.py

  • Committer: Bazaar Package Importer
  • Author(s): Monty Taylor
  • Date: 2010-10-02 14:17:48 UTC
  • mfrom: (1.1.1 upstream)
  • mto: (2.1.17 sid)
  • mto: This revision was merged to the branch mainline in revision 3.
  • Revision ID: james.westby@ubuntu.com-20101002141748-m6vbfbfjhrw1153e
Tags: 2010.09.1802-1
* New upstream release.
* Removed pid-file argument hack.
* Updated GPL-2 address to be new address.
* Directly copy in drizzledump.1 since debian doesn't have sphinx 1.0 yet.
* Link to jquery from libjs-jquery. Add it as a depend.
* Add drizzled.8 symlink to the install files.

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/mysql_protocol/prototest/prototest/mysql/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
# Use and distribution licensed under the BSD license.  See
 
9
# the COPYING.BSD file in the root source directory for full text.
 
10
#
 
11
'''
 
12
MySQL Protocol Command Objects
 
13
'''
 
14
 
 
15
import unittest
 
16
 
 
17
class CommandID(object):
 
18
  SLEEP = 0
 
19
  QUIT = 1
 
20
  INIT_DB = 2
 
21
  QUERY = 3
 
22
  FIELD_LIST = 4
 
23
  CREATE_DB = 5
 
24
  DROP_DB = 6
 
25
  REFRESH = 7
 
26
  SHUTDOWN = 8
 
27
  STATISTICS = 9
 
28
  PROCESS_INFO = 10
 
29
  CONNECT = 11
 
30
  PROCESS_KILL = 12
 
31
  DEBUG = 13
 
32
  PING = 14
 
33
  TIME = 15
 
34
  DELAYED_INSERT = 16
 
35
  CHANGE_USER = 17
 
36
  BINLOG_DUMP = 18
 
37
  TABLE_DUMP = 19
 
38
  CONNECT_OUT = 20
 
39
  REGISTER_SLAVE = 21
 
40
  STMT_PREPARE = 22
 
41
  STMT_EXECUTE = 23
 
42
  STMT_SEND_LONG_DATA = 24
 
43
  STMT_CLOSE = 25
 
44
  STMT_RESET = 26
 
45
  SET_OPTION = 27
 
46
  STMT_FETCH = 28
 
47
  DAEMON = 29
 
48
  END = 30
 
49
 
 
50
class Command(object):
 
51
  '''This class represents a command packet sent from the client.'''
 
52
 
 
53
  def __init__(self, packed=None, command=CommandID.SLEEP, payload=''):
 
54
    if packed is None:
 
55
      self.command = command
 
56
      self.payload = payload
 
57
    else:
 
58
      self.command = ord(packed[0])
 
59
      self.payload = packed[1:]
 
60
 
 
61
  def pack(self):
 
62
    return chr(self.command) + self.payload
 
63
 
 
64
  def __str__(self):
 
65
    return '''Command
 
66
  command = %s
 
67
  payload = %s
 
68
''' % (self.command, self.payload)
 
69
 
 
70
class TestCommand(unittest.TestCase):
 
71
 
 
72
  def testDefaultInit(self):
 
73
    command = Command()
 
74
    self.assertEqual(command.command, CommandID.SLEEP)
 
75
    self.assertEqual(command.payload, '')
 
76
 
 
77
  def testKeywordInit(self):
 
78
    command = Command(command=CommandID.QUERY, payload='abc')
 
79
    self.assertEqual(command.command, CommandID.QUERY)
 
80
    self.assertEqual(command.payload, 'abc')
 
81
 
 
82
  def testUnpackInit(self):
 
83
    command = Command('\x03abc')
 
84
    self.assertEqual(command.command, CommandID.QUERY)
 
85
    self.assertEqual(command.payload, 'abc')
 
86
 
 
87
  def testPack(self):
 
88
    command = Command(Command(command=CommandID.QUERY, payload='abc').pack())
 
89
    self.assertEqual(command.command, CommandID.QUERY)
 
90
    self.assertEqual(command.payload, 'abc')
 
91
 
 
92
class QueryCommand(Command):
 
93
  def __init__(self, packed=None, query=''):
 
94
    super(QueryCommand, self).__init__(packed=packed, command=CommandID.QUERY,
 
95
                                       payload=query)
 
96
 
 
97
  def __str__(self):
 
98
    return '''Command
 
99
  command = %s
 
100
  query = %s
 
101
''' % (self.command, self.payload)
 
102
 
 
103
class TestQueryCommand(unittest.TestCase):
 
104
 
 
105
  def testDefaultInit(self):
 
106
    query = QueryCommand()
 
107
    self.assertEqual(query.command, CommandID.QUERY)
 
108
    self.assertEqual(query.payload, '')
 
109
 
 
110
  def testKeywordInit(self):
 
111
    query = QueryCommand(query='abc')
 
112
    self.assertEqual(query.command, CommandID.QUERY)
 
113
    self.assertEqual(query.payload, 'abc')
 
114
 
 
115
  def testUnpackInit(self):
 
116
    query = QueryCommand('\x03abc')
 
117
    self.assertEqual(query.command, CommandID.QUERY)
 
118
    self.assertEqual(query.payload, 'abc')
 
119
 
 
120
  def testPack(self):
 
121
    query = QueryCommand(QueryCommand(query='abc').pack())
 
122
    self.assertEqual(query.command, CommandID.QUERY)
 
123
    self.assertEqual(query.payload, 'abc')
 
124
 
 
125
if __name__ == '__main__':
 
126
  unittest.main()