File: tcp_connect.py.in

package info (click to toggle)
pydb 1.19-1
  • links: PTS
  • area: main
  • in suites: etch, etch-m68k
  • size: 1,952 kB
  • ctags: 1,412
  • sloc: python: 3,065; perl: 2,479; sh: 586; makefile: 555; lisp: 265; ansic: 16
file content (113 lines) | stat: -rw-r--r-- 4,045 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
#!@PYTHON@ -t
# $Id: tcp_connect.py.in,v 1.4 2006/09/21 01:13:41 rockyb Exp $ -*- Python -*-

# This unit test doesn't use any of the debugger code. It is meant solely
# to test the connection classes.

import os, sys, threading, time, unittest

top_builddir = "@top_builddir@"
if top_builddir[-1] != os.path.sep:
    top_builddir += os.path.sep
sys.path.insert(0, os.path.join(top_builddir, 'pydb'))
top_srcdir = "@top_srcdir@"
if top_srcdir[-1] != os.path.sep:
    top_srcdir += os.path.sep
sys.path.insert(0, os.path.join(top_srcdir, 'pydb'))

import connection

debug=True

class TestTCPConnections(unittest.TestCase):

    def setUp(self):
        self.port = 8000
        self.host = '127.0.0.1'
        self.server = connection.ConnectionServerTCP()
        self.client = connection.ConnectionClientTCP()

    def tear_Down(self):
        self.server.disconnect()
        self.client.disconnect()

    def repeated_connect(self, MAXTRIES=15):
        """Try to connect the client to self.addr either until we've
        tried MAXTRIES times or until it succeeds."""
        for i in range(MAXTRIES):
            if self.server._sock is None:
                time.sleep(0.01)
            else:
                for j in range(MAXTRIES):
                    try:
                        self.client.connect(self.addr)
                        if self.client.connected:
                            if debug:
                                print "client connected"
                            return True
                    except connection.ConnectionFailed, e:
                        if debug: print e
            
        if debug: print "Client can't connect"
        return False
        
    def server_listen(self, MAXPORTS=20):
        """Try to start the server listening over a range of ports
        starting with self.port"""
        self.addr = "%s:%d" % (self.host, self.port)
        for i in range(MAXPORTS):
            try:
                self.server.connect(self.addr)
                if debug: print "Listening on socket %s" % self.addr
                return True
            except connection.ConnectionFailed, e:
                if e[0] == 'Address already in use':
                    self.port += 1
                    print "Trying port %d" % self.port
                    self.addr = "%s:%d" % (self.host, self.port)
        if debug: print "No free port found"
        return False
                
    def setup_client_and_server(self):
        t=threading.Thread(target=self.server_listen)
        t.start()
        if not self.repeated_connect(): return False
        for i in range(15):
            if self.server._sock is not None and \
                   self.server.output is not None:
                return True
            time.sleep(0.01)

    def test_client_server(self):
        """Test that the we can read and write between a TCP client and
        server"""
        if not self.setup_client_and_server():
            self.server.disconnect()
            self.client.disconnect()
            return 
        self.server.write("good")
        line = self.client.readline()
        self.assertEqual("good", line, "Server write to client read bad")
        self.client.write('success')
        line = self.server.readline()
        self.assertEqual('success\n', line, 'Client write to server read bad')

    def test_disconnect_disconnected(self):
        """Test disconnecting twice """
        self.server.disconnect()
        self.server.disconnect()

    def test_invalid_address_port_pair(self):
        """Test an erroneous connection name"""
        addr = 'localhost 8000'
        self.assertRaises(connection.ConnectionFailed,
                          self.server.connect, addr)

    def test_invalid_server_address(self):
        """Test an erroneous server address"""
        addr = 'fff.209320909xcmnm2iu3-=0-0-z.,x.,091209:2990'
        self.assertRaises(connection.ConnectionFailed,
                          self.server.connect, addr)

if __name__ == '__main__':
    unittest.main()