Source

txMysql / txmysql / test / test_protocol.py

Full commit
""" tests for the mysql protocol factory

"""
import StringIO

from twisted.internet import protocol
from twisted.internet import reactor
from twisted.trial import unittest

from txmysql import packet
from txmysql import protocol as myprotocols


class TestMysqlProtocolConnection(unittest.TestCase):
    def setUp(self):
        self.buff = StringIO.StringIO()
        transport = protocol.FileWrapper(self.buff)
        f = myprotocols.MysqlProtocolFactory("test",
                                                  "test")
        self.p = f.buildProtocol(None)
        self.p.transport = transport
        
    def test_buffering(self):
        self.p.dataReceived("Hello World")
        self.assertEqual(self.p.buffer, "Hello World")

    def test_authenticationMessage(self):
        self.p.connectionMade()
        self.assertEqual(self.buff.getvalue(), "")

    def _receiveHandshakeInitializationPacket(self):
        handshakeMessage = "\x085.1b1\x00\xe8\x03\x00\x00abcdefgh\x00\x00\x00!\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00abcdefghijklm"
        bytes = packet.packHeader(len(handshakeMessage), 0)
        bytes += handshakeMessage
        self.p.dataReceived(bytes)

    def test_awaitingToAuthenticating(self):
        self.assertEqual(self.p.state, myprotocols.protocolStates.AWAITING_HANDSHAKE)
        self._receiveHandshakeInitializationPacket()
        self.assertEqual(self.p.state, myprotocols.protocolStates.AUTHENTICATING)

    def test_sendAuthenticationPacket(self):
        self._receiveHandshakeInitializationPacket()
        self.assertEqual(self.buff.getvalue(), ".\x00\x00\x00")