copyleftie/gajim-plural
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Small cleanups to test_nonblockingtcp.py without much value.

Browse source
This commit is contained in:
Stephan Erb 2008-12-24 14:46:21 +00:00
parent c5aff6f37c
commit 53d1fa6f66
1 changed files with 13 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
test/test_nonblockingtcp.py
View file

@ -3,9 +3,8 @@ Unit test for NonBlockingTCP tranport.
''' '''
import unittest import unittest
from xmpp_mocks import * from xmpp_mocks import IdleQueueThread, IdleMock
import sys, os.path
import threading, sys, os.path, time
gajim_root = os.path.join(os.path.abspath(os.path.dirname(__file__)), '..') gajim_root = os.path.join(os.path.abspath(os.path.dirname(__file__)), '..')
@ -15,25 +14,27 @@ sys.path.append(gajim_root + '/src/common')
import transports_nb import transports_nb
from client import * from client import *
xmpp_server = ('gajim.org',5222) xmpp_server = ('gajim.org', 5222)
''' '''
2-tuple - (XMPP server hostname, c2s port) 2-tuple - (XMPP server hostname, c2s port)
Script will connect to the machine. Script will connect to the machine.
''' '''
import socket import socket
ips = socket.getaddrinfo(xmpp_server[0], xmpp_server[1], socket.AF_UNSPEC,socket.SOCK_STREAM) ips = socket.getaddrinfo(xmpp_server[0], xmpp_server[1],
socket.AF_UNSPEC,socket.SOCK_STREAM)
# change xmpp_server on real values # change xmpp_server on real values
ip = ips[0] ip = ips[0]
class MockClient(IdleMock): class MockClient(IdleMock):
def __init__(self, idlequeue): def __init__(self, idlequeue):
self.idlequeue=idlequeue self.idlequeue = idlequeue
IdleMock.__init__(self) IdleMock.__init__(self)
def do_connect(self): def do_connect(self):
self.socket=transports_nb.NonBlockingTCP( self.socket = transports_nb.NonBlockingTCP(
lambda(event_type, data): sys.stdout.write('raising event %s: %s' % ( lambda(event_type, data): sys.stdout.write('raising event %s: %s' % (
event_type, data)), lambda: self.on_success(mode='SocketDisconnect'), event_type, data)), lambda: self.on_success(mode='SocketDisconnect'),
self.idlequeue, False, None) self.idlequeue, False, None)
@ -70,12 +71,14 @@ class TestNonBlockingTCP(unittest.TestCase):
self.idlequeue_thread.stop_thread() self.idlequeue_thread.stop_thread()
self.idlequeue_thread.join() self.idlequeue_thread.join()
def testSth(self): def test_connect_disconnect(self):
self.client.do_connect() self.client.do_connect()
self.assert_(self.client.socket.state == 'CONNECTED') self.assert_(self.client.socket.state == 'CONNECTED')
self.client.do_disconnect() self.client.do_disconnect()
self.assert_(self.client.socket.state == 'DISCONNECTED') self.assert_(self.client.socket.state == 'DISCONNECTED')
if __name__ == '__main__': if __name__ == '__main__':
suite = unittest.TestLoader().loadTestsFromTestCase(TestNonBlockingTCP) unittest.main()
unittest.TextTestRunner(verbosity=2).run(suite)
# vim: se ts=3: