1. Mike Steder
  2. txMysql

Source

txMysql / txmysql / test / test_protocol.py

""" 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 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 test_handshakeInitializationPacket(self):
    #     self.p.dataReceived()
    #     self.assertEqual(len(self.p.packets), 1)
    #     packet = self.p.packets[0]
    #     self.assertEqual(packet, None)