180 lines
7.3 KiB
Python
180 lines
7.3 KiB
Python
"""
|
|
@file test_udpconnection.py
|
|
@date 2008-09-16
|
|
Contributors can be viewed at:
|
|
http://svn.secondlife.com/svn/linden/projects/2008/pyogp/CONTRIBUTORS.txt
|
|
|
|
$LicenseInfo:firstyear=2008&license=apachev2$
|
|
|
|
Copyright 2008, Linden Research, Inc.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License").
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
or in
|
|
http://svn.secondlife.com/svn/linden/projects/2008/pyogp/LICENSE.txt
|
|
|
|
$/LicenseInfo$
|
|
"""
|
|
|
|
#standard libraries
|
|
import unittest, doctest
|
|
import pprint
|
|
|
|
# pyogp
|
|
from pyogp.lib.base.settings import Settings
|
|
#from pyogp.lib.base.message.udp_connection import MessageSystem
|
|
from pyogp.lib.base.message.types import MsgType
|
|
from pyogp.lib.base.tests.mockup_net import MockupUDPServer, MockupUDPClient
|
|
from pyogp.lib.base.message.message import Message, Block
|
|
from pyogp.lib.base.message.circuit import Host
|
|
from pyogp.lib.base.message.udpdispatcher import UDPDispatcher
|
|
|
|
class TestUDPConnection(unittest.TestCase):
|
|
|
|
def tearDown(self):
|
|
pass
|
|
|
|
def setUp(self):
|
|
|
|
self.settings = Settings()
|
|
self.settings.ENABLE_DEFERRED_PACKET_PARSING = False
|
|
|
|
self.udp_connection = UDPDispatcher(MockupUDPClient(), settings = self.settings)
|
|
self.host = Host( (MockupUDPServer(), 80) )
|
|
|
|
def test_find_circuit(self):
|
|
host = Host((MockupUDPServer(), 80))
|
|
host2 = Host((MockupUDPServer(), 80))
|
|
udp_connection = UDPDispatcher()
|
|
assert len(udp_connection.circuit_manager.circuit_map) == 0, \
|
|
"Circuit map has incorrect circuits"
|
|
circuit1 = udp_connection.find_circuit(host)
|
|
assert len(udp_connection.circuit_manager.circuit_map) == 1, \
|
|
"Circuit map has incorrect circuits 2"
|
|
circuit2 = udp_connection.find_circuit(host2)
|
|
assert len(udp_connection.circuit_manager.circuit_map) == 2, \
|
|
"Circuit map has incorrect circuits 3"
|
|
circuit3 = udp_connection.find_circuit(host2)
|
|
assert circuit2 == circuit3, "Didn't save circuit"
|
|
assert len(udp_connection.circuit_manager.circuit_map) == 2, \
|
|
"Circuit map has incorrect circuits 4"
|
|
|
|
|
|
def test_send_variable(self):
|
|
msg = Message('PacketAck',
|
|
Block('Packets', ID=0x00000003)
|
|
)
|
|
buf = self.udp_connection.send_message(msg, self.host)
|
|
assert buf == \
|
|
'\x00' + '\x00\x00\x00\x01' + '\x00' + '\xff\xff\xff\xfb' + \
|
|
'\x01' + '\x03\x00\x00\x00', \
|
|
'Received: ' + repr(buf) + ' ' + \
|
|
'Expected: ' + repr('\x00' + '\x00\x00\x00\x01' + '\x00' + \
|
|
'\xff\xff\xff\xfb' + '\x01' + '\x03\x00\x00\x00')
|
|
|
|
def test_send_same_host(self):
|
|
msg = Message('PacketAck',
|
|
Block('Packets', ID=0x00000003)
|
|
)
|
|
ret1 = self.udp_connection.send_message(msg, self.host)
|
|
|
|
msg2 = Message('PacketAck',
|
|
Block('Packets', ID=0x00000003)
|
|
)
|
|
ret2 = self.udp_connection.send_message(msg2, self.host)
|
|
|
|
#strings to test for
|
|
test_str = '\x00' + '\x00\x00\x00\x01' + '\x00' + '\xff\xff\xff\xfb' + \
|
|
'\x01' + '\x03\x00\x00\x00'
|
|
test_str2 = '\x00' + '\x00\x00\x00\x02' + '\x00' + '\xff\xff\xff\xfb' + \
|
|
'\x01' + '\x03\x00\x00\x00'
|
|
|
|
assert ret1 == \
|
|
test_str, \
|
|
'Received: ' + repr(ret1) + ' ' + \
|
|
'Expected: ' + repr(test_str)
|
|
|
|
assert ret2 == \
|
|
test_str2, \
|
|
'Received: ' + repr(ret2) + ' ' + \
|
|
'Expected: ' + repr(test_str2)
|
|
|
|
def test_send_reliable(self):
|
|
msg = Message('PacketAck',
|
|
Block('Packets', ID=0x00000003)
|
|
)
|
|
host = Host((MockupUDPServer(), 80))
|
|
ret_msg = self.udp_connection.send_reliable(msg, host, 10)
|
|
test_str = '\x40' + '\x00\x00\x00\x01' + '\x00' + '\xff\xff\xff\xfb' + \
|
|
'\x01' + '\x03\x00\x00\x00'
|
|
assert ret_msg == \
|
|
test_str ,\
|
|
'Received: ' + repr(msg) + ' ' + \
|
|
'Expected: ' + repr(test_str)
|
|
|
|
def test_receive(self):
|
|
out_message = '\x00' + '\x00\x00\x00\x01' + '\x00' + \
|
|
'\xff\xff\xff\xfb' + '\x01' + '\x01\x00\x00\x00'
|
|
server = MockupUDPServer()
|
|
server.send_message(self.udp_connection.udp_client, out_message)
|
|
|
|
data, data_size = self.udp_connection.udp_client.receive_packet(0)
|
|
packet = self.udp_connection.receive_check(self.udp_connection.udp_client.sender,
|
|
data, data_size)
|
|
assert packet.name == 'PacketAck'
|
|
data = packet.message_data.blocks['Packets'][0].vars['ID'].data
|
|
assert data == 1, "ID Data incorrect: " + str(data)
|
|
|
|
def test_receive_zero(self):
|
|
out_message = '\x80' + '\x00\x00\x00\x01' + '\x00' + \
|
|
'\xff\xff\xff\xfb' + '\x01' + '\x01\x00\x03'
|
|
server = MockupUDPServer()
|
|
server.send_message(self.udp_connection.udp_client, out_message)
|
|
|
|
data, data_size = self.udp_connection.udp_client.receive_packet(0)
|
|
packet = self.udp_connection.receive_check(self.udp_connection.udp_client.sender,
|
|
data, data_size)
|
|
assert packet.name == 'PacketAck'
|
|
data = packet.message_data.blocks['Packets'][0].vars['ID'].data
|
|
assert data == 1, "ID Data incorrect: " + str(data)
|
|
|
|
def test_receive_reliable(self):
|
|
out_message = '\x40' + '\x00\x00\x00\x05' + '\x00' + \
|
|
'\xff\xff\xff\xfb' + '\x01' + '\x01\x00\x00\x00'
|
|
server = MockupUDPServer()
|
|
server.send_message(self.udp_connection.udp_client, out_message)
|
|
|
|
data, data_size = self.udp_connection.udp_client.receive_packet(0)
|
|
packet = self.udp_connection.receive_check(self.udp_connection.udp_client.sender,
|
|
data, data_size)
|
|
sender_host = self.udp_connection.udp_client.get_sender()
|
|
circuit = self.udp_connection.circuit_manager.get_circuit(sender_host)
|
|
assert len(circuit.acks) == 1, "Ack not collected"
|
|
assert circuit.acks[0] == 5, "Ack ID not correct, got " + str(circuit.acks[0])
|
|
|
|
def test_acks(self):
|
|
out_message = '\x40' + '\x00\x00\x00\x05' + '\x00' + \
|
|
'\xff\xff\xff\xfb' + '\x01' + '\x00\x00\x00\x01'
|
|
server = MockupUDPServer()
|
|
server.send_message(self.udp_connection.udp_client, out_message)
|
|
|
|
data, data_size = self.udp_connection.udp_client.receive_packet(0)
|
|
packet = self.udp_connection.receive_check(self.udp_connection.udp_client.sender,
|
|
data, data_size)
|
|
assert server.rec_buffer == '', "ERROR: server has message without " + \
|
|
"receiving one"
|
|
self.udp_connection.process_acks()
|
|
assert server.rec_buffer != '', "Ack not sent"
|
|
test_msg = '\x00' + '\x00\x00\x00\x01' + '\x00' + \
|
|
'\xff\xff\xff\xfb' + '\x01' + '\x05\x00\x00\x00'
|
|
assert server.rec_buffer == test_msg, "Ack received incorrect, got " + \
|
|
repr(server.rec_buffer)
|
|
|
|
def test_suite():
|
|
from unittest import TestSuite, makeSuite
|
|
suite = TestSuite()
|
|
suite.addTest(makeSuite(TestUDPConnection))
|
|
return suite
|