1

import unittest

     2

     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)

    68

    69

    def test_connect_inet (self) :

    70

        cc = tcp.Client(endpoint.InetAddr('localhost', self.sockaddr.port), socket.AF_INET)

    71

        cs = cc.connect()

    72

    73

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

    74

    76

if __name__ == '__main__' :

    77

    unittest.main()

    78