--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/test/transport_tcp.py Sun Aug 23 22:32:05 2009 +0300
@@ -0,0 +1,62 @@
+import unittest
+
+from qmsk.net.socket import socket as _socket, af_inet
+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, af=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.SockAddr(self.sockaddr)
+
+ def test_connect_unspec (self) :
+ cc = tcp.Client(self.addr)
+ cs = cc.connect()
+
+ self.assertEquals(cs.sock.getpeername(), self.sockaddr)
+
+ def test_connect_inet4 (self) :
+ cc = tcp.Client(self.addr, socket.AF_INET)
+ 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.SockAddr(sockaddr))
+ cs = cc.connect()
+
+ self.assertEquals(cs.sock.getsockname(), sockaddr)
+ self.assertEquals(cs.sock.getpeername(), self.sockaddr)
+
+if __name__ == '__main__' :
+ unittest.main()
+