#!@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()
