1 import unittest

     2 

     3 from qmsk.net.socket import socket as _socket, af_inet

     4 from qmsk.net.transport import tcp, socket, endpoint

     5 

     6 class TestService (unittest.TestCase) :

     7     def setUp (self) :

     8         # create service on random port

     9         self.ss = tcp.Service(None, af=socket.AF_INET)

    10 

    11         self.addr = self.ss.sock.getsockname()

    12     

    13     def test_bind_addr (self) :

    14         # ensure it binds on the right AF

    15         self.assertEquals(self.addr.family, socket.AF_INET)

    16         

    17     def test_connect_accept (self) :

    18         # ensure we can connect to it...

    19         sock = _socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    20         sock.connect(self.addr)

    21 

    22         # and accept

    23         sc = self.ss.accept()

    24         

    25         # test both ways

    26         self.assertEquals(sock.getpeername().port, self.addr.port)

    27         self.assertEquals(sc.sock.getpeername().port, sock.getsockname().port)

    28     

    29 class TestClient (unittest.TestCase) :

    30     def setUp (self) :

    31         # create listening socket on random port

    32         self.ls = _socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    33         self.ls.listen(1)

    34 

    35         self.sockaddr = af_inet.sockaddr_in('127.0.0.1', self.ls.getsockname().port)

    36         self.addr = endpoint.SockAddr(self.sockaddr)

    37    

    38     def test_connect_unspec (self) :

    39         cc = tcp.Client(self.addr)

    40         cs = cc.connect()

    41 

    42         self.assertEquals(cs.sock.getpeername(), self.sockaddr)

    43 

    44     def test_connect_inet4 (self) :

    45         cc = tcp.Client(self.addr, socket.AF_INET)

    46         cs = cc.connect()

    47 

    48         self.assertEquals(cs.sock.getpeername(), self.sockaddr)

    49 

    50     def test_connect_bind (self) :

    51         sockaddr = af_inet.sockaddr_in('127.0.0.1', self.sockaddr.port + 1)

    52         

    53         # connect with bind()

    54         cc = tcp.Client(self.addr, bind_endpoint=endpoint.SockAddr(sockaddr))

    55         cs = cc.connect()

    56 

    57         self.assertEquals(cs.sock.getsockname(), sockaddr)

    58         self.assertEquals(cs.sock.getpeername(), self.sockaddr)

    59     

    60 if __name__ == '__main__' :

    61     unittest.main()

    62