root/trunk/RBTelepathy/test/test_packet.py

#!/usr/bin/env python
##~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
##~ License 
##~ 
##- The RuneBlade Foundation library is intended to ease some 
##- aspects of writing intricate Jabber, XML, and User Interface (wxPython, etc.) 
##- applications, while providing the flexibility to modularly change the 
##- architecture. Enjoy.
##~ 
##~ Copyright (C) 2002  TechGame Networks, LLC.
##~ 
##~ This library is free software; you can redistribute it and/or
##~ modify it under the terms of the BSD style License as found in the 
##~ LICENSE file included with this distribution.
##~ 
##~ TechGame Networks, LLC can be reached at:
##~ 3578 E. Hartsel Drive #211
##~ Colorado Springs, Colorado, USA, 80920
##~
##~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#~ Imports 
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

# Path adjustmet so the test can run out of the box
import sys; sys.path.append('..')
from test import test_support
import unittest

from RBTelepathy import Packet

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#~ Definitions 
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

class AddressedPacketTestCase(unittest.TestCase):
    toAddresses = ('someone', 'act@network')
    XMLHeader = '''<message><to addr='%s'/><to addr='%s'/></message>''' % toAddresses
    Data = ''

    def setUp(self):
        self.builder = Packet.StandardStreamPacketBuilder()

    def testAddressesMatch(self):
        """Test parsed element addresses match inputs"""
        packet = self.builder.BuildPacket(self.XMLHeader)
        toNodes = packet._getElements(node='to')
        self.failUnlessEqual(len(toNodes), len(self.toAddresses))
        for value, expected in zip(toNodes, self.toAddresses):
            self.failUnlessEqual(str(value.addr), expected)

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

class TerminateStreamPacketTestCase(unittest.TestCase):
    DataStreams = [
        ("<stream format='term'/>", "Terminated data test #1", "\x04"),
        ("<stream format='term'/>", "Test #2 for terminated data", "\x04"),
        ("<stream format='term'/>", unittest.__doc__, "\x04"),

        ("<stream format='terminated' ordinal='31'/>", "Data test #3 with different terminator", "\x1f"),
        ("<stream format='terminated' ordinal='07'/>", "Test terminated data #4", "\x07"),
        ("<stream format='terminated' ordinal='27'/>", unittest.__doc__, "\x1B"),
        ]
    XMLHeader = '''<message>%s</message>''' % ''.join([each[0] for each in DataStreams])

    def setUp(self):
        self.builder = Packet.StandardStreamPacketBuilder()
        self.packet = self.builder.BuildPacket(self.XMLHeader)
        for xml, data, term in self.DataStreams:
            self.packet.OnStreamData(data + term)

    def testAllStreamsAccountedFor(self):
        """Test stream count input match output"""
        streams = self.packet._getElements(node='stream')
        self.failUnless(len(streams) == len(self.DataStreams))
        self.failUnless(len(streams) == len(self.packet.datastreams))
        for a,b in zip(self.packet.datastreams, streams):
            self.failUnless(a is b, "packet.datastream should be *identical* to message.stream elements")

    def testStreamContent(self):
        """Test stream content matches input"""
        for (stream, (xml, data, term)) in zip(self.packet.datastreams, self.DataStreams):
            self.failUnless(stream.terminator == term, "Terminator not preserved")
            self.failUnless(stream.StreamContent == data, "Binary data content not preserved")

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

class LenghtStreamPacketTestCase(unittest.TestCase):
    DataStreams = [
        ("<stream format='length' size='%d'/>", "Lenght data test #1"),
        ("<stream format='length' size='%d'/>", "Test #2 for terminated data"),
        ("<stream format='length' size='%d'/>", unittest.__doc__),
        ("<stream format='length' size='%d'/>", open(sys.argv[0], 'r').read()),
        ]
    XMLHeader = '''<message>%s</message>''' % ''.join([each[0]%len(each[1]) for each in DataStreams])

    def setUp(self):
        self.builder = Packet.StandardStreamPacketBuilder()
        self.packet = self.builder.BuildPacket(self.XMLHeader)
        for xml, data in self.DataStreams:
            self.packet.OnStreamData(data)

    def testAllStreamsAccountedFor(self):
        """Test stream count input match output"""
        streams = self.packet._getElements(node='stream')
        self.failUnless(len(streams) == len(self.DataStreams))
        self.failUnless(len(streams) == len(self.packet.datastreams))
        for a,b in zip(self.packet.datastreams, streams):
            self.failUnless(a is b, "packet.datastream should be *identical* to message.stream elements")

    def testStreamContent(self):
        """Test stream content matches input"""
        for (stream, (xml, data)) in zip(self.packet.datastreams, self.DataStreams):
            self.failUnless(stream.ContentLen == len(data), "length not preserved")
            self.failUnless(len(stream.StreamContent) == len(data), "length not preserved")
            self.failUnless(stream.StreamContent == data, "Binary data content not preserved")

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#~ Unittest Main 
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

if __name__=='__main__':
    unittest.main()