terom@34: import unittest
terom@34: 
terom@36: from qmsk.net.socket import socket as _socket, af_inet, af_inet6
terom@34: from qmsk.net.transport import tcp, socket, endpoint
terom@34: 
terom@34: class TestService (unittest.TestCase) :
terom@34:     def setUp (self) :
terom@34:         # create service on random port
terom@37:         self.ss = tcp.Service(None, family=socket.AF_INET)
terom@34: 
terom@34:         self.addr = self.ss.sock.getsockname()
terom@34:     
terom@34:     def test_bind_addr (self) :
terom@34:         # ensure it binds on the right AF
terom@34:         self.assertEquals(self.addr.family, socket.AF_INET)
terom@34:         
terom@34:     def test_connect_accept (self) :
terom@34:         # ensure we can connect to it...
terom@34:         sock = _socket.socket(socket.AF_INET, socket.SOCK_STREAM)
terom@34:         sock.connect(self.addr)
terom@34: 
terom@34:         # and accept
terom@34:         sc = self.ss.accept()
terom@34:         
terom@34:         # test both ways
terom@34:         self.assertEquals(sock.getpeername().port, self.addr.port)
terom@34:         self.assertEquals(sc.sock.getpeername().port, sock.getsockname().port)
terom@34:     
terom@34: class TestClient (unittest.TestCase) :
terom@34:     def setUp (self) :
terom@34:         # create listening socket on random port
terom@34:         self.ls = _socket.socket(socket.AF_INET, socket.SOCK_STREAM)
terom@34:         self.ls.listen(1)
terom@34: 
terom@34:         self.sockaddr = af_inet.sockaddr_in('127.0.0.1', self.ls.getsockname().port)
terom@34:         self.addr = endpoint.SockAddr(self.sockaddr)
terom@34:    
terom@37:     def test_connect (self) :
terom@34:         cc = tcp.Client(self.addr)
terom@34:         cs = cc.connect()
terom@34: 
terom@34:         self.assertEquals(cs.sock.getpeername(), self.sockaddr)
terom@34: 
terom@34:     def test_connect_bind (self) :
terom@34:         sockaddr = af_inet.sockaddr_in('127.0.0.1', self.sockaddr.port + 1)
terom@34:         
terom@34:         # connect with bind()
terom@34:         cc = tcp.Client(self.addr, bind_endpoint=endpoint.SockAddr(sockaddr))
terom@34:         cs = cc.connect()
terom@34: 
terom@34:         self.assertEquals(cs.sock.getsockname(), sockaddr)
terom@34:         self.assertEquals(cs.sock.getpeername(), self.sockaddr)
terom@36:  
terom@36:     def test_connect_bind6 (self) :
terom@36:         sockaddr = af_inet6.sockaddr_in6('::1', self.sockaddr.port + 1)
terom@36:         
terom@36:         # connect with bind()
terom@36:         cc = tcp.Client(self.addr, bind_endpoint=endpoint.SockAddr(sockaddr))
terom@36:         
terom@36:         # should fail (IPv6 -> IPv4)
terom@36:         self.assertRaises(socket.SocketConnectEndpointError, cc.connect)
terom@34:     
terom@35:     def test_connect_inet (self) :
terom@37:         cc = tcp.Client(endpoint.InetAddr('localhost', self.sockaddr.port))
terom@35:         cs = cc.connect()
terom@35: 
terom@35:         self.assertEquals(cs.sock.getpeername(), self.sockaddr)
terom@35: 
terom@36:     
terom@34: if __name__ == '__main__' :
terom@34:     unittest.main()
terom@34: