1

import unittest

     2

     3

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

     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

    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

    39

        cc = tcp.Client(self.addr)

    40

        cs = cc.connect()

    41

    42

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

    43

    44

    def test_connect_bind (self) :

    45

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

    46

    47

        # connect with bind()

    48

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

    49

        cs = cc.connect()

    50

    51

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

    52

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

    53

    54

    def test_connect_bind6 (self) :

    55

        sockaddr = af_inet6.sockaddr_in6('::1', self.sockaddr.port + 1)

    56

    57

        # connect with bind()

    58

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

    59

    60

        # should fail (IPv6 -> IPv4)

    61

        self.assertRaises(socket.SocketConnectEndpointError, cc.connect)

    62

    63

    def test_connect_inet (self) :

    65

        cs = cc.connect()

    66

    67

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

    68

    69

    70

if __name__ == '__main__' :

    71

    unittest.main()

    72