event keeps a ref to its event_base - needed for later error propagation, and to ensure method calls remain valid
import unittest
from qmsk.net.socket import socket as _socket, af_inet, af_inet6
from qmsk.net.transport import tcp, socket, endpoint
class TestService (unittest.TestCase) :
def setUp (self) :
# create service on random port
self.ss = tcp.Service(None, family=socket.AF_INET)
self.addr = self.ss.sock.getsockname()
def test_bind_addr (self) :
# ensure it binds on the right AF
self.assertEquals(self.addr.family, socket.AF_INET)
def test_connect_accept (self) :
# ensure we can connect to it...
sock = _socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(self.addr)
# and accept
sc = self.ss.accept()
# test both ways
self.assertEquals(sock.getpeername().port, self.addr.port)
self.assertEquals(sc.sock.getpeername().port, sock.getsockname().port)
class TestClient (unittest.TestCase) :
def setUp (self) :
# create listening socket on random port
self.ls = _socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.ls.listen(1)
self.sockaddr = af_inet.sockaddr_in('127.0.0.1', self.ls.getsockname().port)
self.addr = endpoint.Address(self.sockaddr)
def test_connect (self) :
cc = tcp.Client(self.addr)
cs = cc.connect()
self.assertEquals(cs.sock.getpeername(), self.sockaddr)
def test_connect_bind (self) :
sockaddr = af_inet.sockaddr_in('127.0.0.1', self.sockaddr.port + 1)
# connect with bind()
cc = tcp.Client(self.addr, bind_endpoint=endpoint.Address(sockaddr))
cs = cc.connect()
self.assertEquals(cs.sock.getsockname(), sockaddr)
self.assertEquals(cs.sock.getpeername(), self.sockaddr)
def test_connect_bind6 (self) :
sockaddr = af_inet6.sockaddr_in6('::1', self.sockaddr.port + 1)
# connect with bind()
cc = tcp.Client(self.addr, bind_endpoint=endpoint.Address(sockaddr))
# should fail (IPv6 -> IPv4)
self.assertRaises(socket.SocketConnectEndpointError, cc.connect)
def test_connect_inet (self) :
cc = tcp.Client(endpoint.InetEndpoint('localhost', self.sockaddr.port))
cs = cc.connect()
self.assertEquals(cs.sock.getpeername(), self.sockaddr)
if __name__ == '__main__' :
unittest.main()