Python-сообщество

Привет. Пишу маленькую утилитку, которая должна:

1) Ставится на каждый из узлов кластера
2) Посылать сообщения на другие узлы кластера. Нагрузку можно менять.
3) Принимать сообщения с других узлов кластера.

Выбрал Twisted. Вот что получилось:

Multicast.py:

from twisted.internet.protocol import DatagramProtocol, Factory
from twisted.internet import defer
from time import time

class MulticastServerUDP(DatagramProtocol):
def __init__(self, adress, changefunction):
self.adress = adress
self.changefunction = changefunction

def startProtocol(self):
print ‘Started Listening’
# Join a specific multicast group
self.transport.joinGroup(self.adress)

def datagramReceived(self, datagram, address):
if datagram == “System”:
print " ' + repr(datagram) + “ from ” + repr(address)
self.changefunction(datagram)

class MulticastClient(DatagramProtocol):
def __init__(self, data, adress):
self.data = data
self.adress = adress

def SendPacket(self):
self.transport.write(self.data, self.adress)

main.py

from twisted.internet import reactor, task
from twisted.application.internet import MulticastServer
from twisted.internet.protocol import DatagramProtocol
import LoadingProgram.Multicast

SendPacketLoopTime = 1
MulticastGroupAdress = ('224.0.0.1', 8005)
MulticastListeningInterface = ‘0.0.0.0’

def ChangeFunction(data):
SendPacketLoopTime = float(data)
loopSendPacket.stop()
loopSendPacket.start(SendPacketLoopTime)

# Listen for multicast on MulticastGroupAdress
serverprotocol = LoadingProgram.Multicast.MulticastServerUDP(MulticastGroupAdress, ChangeFunction)
reactor.listenMulticast(8005, serverprotocol, MulticastListeningInterface)

# Send multicast on MulticastGroupAdress from our dynamically allocated port
clientprotocol = LoadingProgram.Multicast.MulticastClient(u'UIR', MulticastGroupAdress)
reactor.listenUDP(0, clientprotocol)
loopSendPacket = task.LoopingCall(clientprotocol.SendPacket)
loopSendPacket.start(SendPacketLoopTime)

reactor.run()

Программа замечательно выполняет свои ф-ии, однако, вдохновившись XP программированием и статьёй pythy (http://gorod-omsk.ru/blog/pythy/2006/07/30/from-perl-script-to-twisted-app-writing-unit-tests/) решил написать для неё тесты.

Однако никак не могу понять что тут нужно проверять и как проверять?

Может быть, гуру, посоветуете что-то?

Отредактировано (Апрель 1, 2007 17:51:20)

1) М.б. Kamaelia лучше подойдет для распределенных приложений (честно - не пробовал)
2) Я бы тестировал логику (поведение сервера/клиента), т.е. модуль MultiCast
3) Тестируется, заменяя транспорт StringIO, проверяются ответы ожидаемые и реальные (как это делается, описано в ссылаемой статье)
4) Неплохо бы использовать {code:python} для кода

j2a
2) Я бы тестировал логику (поведение сервера/клиента), т.е. модуль MultiCast

А что там конкретно надо тестировать?
1) Удачно ли посылаются данные на multicast адрес?
2) Приходит ли то же, что и было послано?

j2a
4) Неплохо бы использовать {code:python} для кода

Исправил. Не знал про тег.

Вот то что я начал делать:

from twisted.trial import unittest
from twisted.internet import protocol, defer
from LoadingProgram import Multicast
from twisted.test.test_protocols import StringIOWithoutClosing as SIOWOC

class ServerProtoTest(unittest.TestCase):
def setUP(self):
self.p = Multicast.MulticastServerUDP()

def test_datagramRecieved(self):
s = SIOWOC()
self.p.makeConnection(protocol.FileWrapper(s))


Но не могу понять вообще то я собираюсь проверять что надо или не то?
Как посылать сообщение, если для посылки мне надо привязаться к локальному порту и только потом посылать? А для привязки нужен reactor, можно ли его в тесты засовывать? Сомневаюсь…

В статье ни одного комментария по поводу тестов. Увы, по ней я не смог разобраться.