File: test-connect.py.in

package info (click to toggle)
pydb 1.26-1
  • links: PTS, VCS
  • area: main
  • in suites: jessie, jessie-kfreebsd, squeeze, wheezy
  • size: 2,460 kB
  • ctags: 1,061
  • sloc: python: 4,200; perl: 2,479; sh: 873; lisp: 866; makefile: 634; ansic: 16
file content (145 lines) | stat: -rwxr-xr-x 4,812 bytes parent folder | download | duplicates (3)
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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
#!@PYTHON@ -t
# $Id: test-connect.py.in,v 1.1 2007/11/01 09:20:31 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
import sys
import 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'))

# Global vars
__addr__ = 'localhost:8002'
MAXTRIES = 100
TESTFN = 'device'

sys.path.append("..")
from connection import ConnectionServerTCP, ConnectionClientTCP, \
     ConnectionSerial, ConnectionFailed

def repeatedConnect(client, addr):
    """Try to connect the client to addr either until we've tried MAXTRIES
    times or until it succeeds."""
    for i in range(MAXTRIES):
        try:
            client.connect(addr)
            if client.connected: break
        except ConnectionFailed:
                pass

class TestTCPConnections(unittest.TestCase):
    def setUp(self):
        self.server = ConnectionServerTCP()
        self.client = ConnectionClientTCP()

##     def test_ErrorAddressAlreadyInUse(self):
##         thread.start_new_thread(repeatedConnect, (self.client, __addr__))
##         self.server.connect(__addr__)

##         # Set up second server on same port and do not reuse the addr
##         s = ConnectionServerTCP()
##         self.assertRaises(ConnectionFailed, s.connect, __addr__, False)

##     def test_ConnectionRefused(self):
##         self.assertRaises(ConnectionFailed, self.client.connect, __addr__)

##     def test_ServerReadError(self):
##         thread.start_new_thread(self.server.connect, (__addr__,))

##         while not self.server._sock:
##             time.sleep(0.1)
            
##         repeatedConnect(self.client, __addr__)

##         # Wait to make _absolutely_ sure that the client has connected
##         while not self.server.output:
##             time.sleep(0.1)
##         self.client.disconnect()
##         self.assertRaises(ReadError, self.server.readline)

##         self.server.disconnect()

##         thread.start_new_thread(self.client.connect, (__addr__,))
##         self.server.connect(__addr__)

##         self.server.disconnect()
##         self.assertRaises(ReadError, self.client.readline)

    def tear_Down(self):
        self.server.disconnect()
        self.client.disconnect()
    
class TestSerialConnections(unittest.TestCase):
    """ This test just uses a file instead of a serial device, which
    on *nix systems is just files anyway.
    """
    def setUp(self):
        self.server = ConnectionSerial()
        self.client = ConnectionSerial()
        fd = open(TESTFN, "wr+")
        fd.close()
        self.server.connect(TESTFN)
        self.client.connect(TESTFN)
        
    def test_ClientToServerConnect(self):
        self.client.disconnect()
        self.server.disconnect()

    def test_ClientWriteRead(self):
        self.client.write('success!')
        line = self.server.readline()
        self.assertEquals('success!\n', line, 'Could not read from client.')

        # Unfortunately the text file doesn't erase what we've written like a
        # device of stream, so we have to close the the file and re-open it.
        self.server.disconnect()
        self.server.connect(TESTFN)
        self.server.write('great!')
        line = self.client.readline()
        self.assertEquals('great!\n', line, 'Could not read from server.')

    def test_DisconnectDisconnected(self):
        self.server.disconnect()

    def test_Readline(self):
        self.client.write('success!\nNext line.')
        self.client.disconnect()
        line = self.server.readline()
        self.assertEquals('success!\n', line, 'Could not read first line')
        line = self.server.readline()
        self.assertEquals('Next line.\n', line, 'Could not read second line.')
        line = self.server.readline()
        self.assertEquals('', line, 'Could not read third line.')

    def test_InvalidFilename(self):
        client = ConnectionSerial()
        self.assertRaises(ConnectionFailed, client.connect,
                          '/dev/pleasepleasepleasedontexit')

    def tearDown(self):
        self.server.disconnect()
        self.client.disconnect()
        os.remove(TESTFN)

    def test_InvalidPipe(self):
        fname='invalid'
        self.assertRaises(ConnectionFailed,self.client.connect, fname)
        client_fname = fname+'0'
        if os.path.exists(client_fname):
            os.unlink(client_fname)

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

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